消息

notify/producer.go

@@ -19,13 +19,11 @@ var (
 
 func New(config *NsqConfig) {
 	nsqConfig = config
-	NsqProducers = InitProducer(config)
 }
 
-func InitProducer(config *NsqConfig) *NsqProducer {
+func InitProducer(config *NsqConfig) {
 	producer, err := nsq.NewProducer(config.Addr, nsq.NewConfig())
 	if logs.CheckErr(err, "InitProducer") {
-		//panic(err)
 		producer.Stop()
 	}
 
@@ -34,21 +32,20 @@ func InitProducer(config *NsqConfig) *NsqProducer {
 		producer.Stop()
 	}
 
-	return &NsqProducer{
+	NsqProducers = &NsqProducer{
 		producer: producer,
 	}
 }
 
-//发布消息
+//Publish 发布消息
 func (p *NsqProducer) Publish(topic string, message string) (err error) {
-	//重试连接
-	NsqProducers = InitProducer(nsqConfig)
+	InitProducer(nsqConfig)
 
-	if p.producer != nil {
+	if NsqProducers.producer != nil {
 		if message == "" { //不能发布空串，否则会导致error
 			return nil
 		}
-		err = p.producer.Publish(topic, []byte(message)) // 发布消息
+		err = NsqProducers.producer.Publish(topic, []byte(message)) // 发布消息
 		if err != nil {
 			g.Log().Cat("Producer").Cat("error").Infof(`消息内容【%v】错误【%v】`, message, err.Error())
 		} else {
@@ -60,16 +57,14 @@ func (p *NsqProducer) Publish(topic string, message string) (err error) {
 	return fmt.Errorf("producer is nil", err)
 }
 
-//发布延迟消息
-func (p *NsqProducer) DeferredPublish(topic string, delay time.Duration, message string) error {
-
-	NsqProducers = InitProducer(nsqConfig)
-	var err error
-	if p.producer != nil {
+//DeferredPublish 发布延迟消息
+func (p *NsqProducer) DeferredPublish(topic string, delay time.Duration, message string) (err error) {
+	InitProducer(nsqConfig)
+	if NsqProducers.producer != nil {
 		if message == "" { //不能发布空串，否则会导致error
 			return nil
 		}
-		err = p.producer.DeferredPublish(topic, delay, []byte(message)) // 发布消息
+		err = NsqProducers.producer.DeferredPublish(topic, delay, []byte(message)) // 发布消息
 		if err != nil {
 			g.Log().Cat("Producer").Cat("error").Infof(`消息内容【%v】错误【%v】`, message, err.Error())
 		} else {