Merge remote-tracking branch 'origin/master'

logs/logs.go

@@ -2,7 +2,6 @@ package logs
 
 import (
 	"context"
-	"fmt"
 	"github.com/gogf/gf/errors/gerror"
 	"github.com/gogf/gf/frame/g"
 )
@@ -36,12 +35,12 @@ func RequestLog(req RequestLogReq) {
 
 //记录info日志
 func Info(path string, format string, v ...interface{}) {
-	g.Log().Async().Cat(path).Println(fmt.Sprintf(format, v...))
+	g.Log().Async().Cat(path).Infof(format, v...)
 }
 
 //记录error日志
 func Error(format string, v ...interface{}) {
-	g.Log().Async().Cat("error").Println(fmt.Sprintf(format, v...))
+	g.Log().Async().Cat("error").Infof(format, v...)
 }
 
 //检查错误

notify/consumer.go

@@ -40,9 +40,10 @@ func InitBatchConsumer(consumer *NsqConfig, clientNum int) {
 	if clientNum == 0 {
 		clientNum = 1
 	}
-	for i := 0; i < clientNum; i++ {
-		c.AddHandler(consumer.Handler) // 批量添加消费者接口
-	}
+
+	c.ChangeMaxInFlight(clientNum) //可以根据nsqds数量来配置
+	c.AddHandler(consumer.Handler) // 批量添加消费者接口
+
 	//建立NSQLookupd连接
 	err = c.ConnectToNSQD(consumer.Addr)
 	if logs.CheckErr(err, "ConnectToNSQD") {

notify/message.go

@@ -24,7 +24,6 @@ const (
 	//API请求日志
 	ApiRequestTopic = "apiRequest"
 
-
 	//消息类型
 
 	//商品消息
@@ -38,10 +37,14 @@ const (
 	OrderComfirmReceiveGoods = 202 //确认收货
 	OrderSuccess             = 203 //交易成功
 	OrderCancel              = 204 //订单取消
-	RefundApply              = 300 //申请售后
-	RefundAgree              = 301 //商家同意售后
-	RefundRefuse             = 302 //商家拒绝售后
-	RefundSuccess            = 303 //售后成功
+	OrderPay                 = 205 //延迟支付
+	OrderRejection           = 206 //已拒收
+	OrderReturn              = 207 //拒收已入库
+
+	RefundApply   = 300 //申请售后
+	RefundAgree   = 301 //商家同意售后
+	RefundRefuse  = 302 //商家拒绝售后
+	RefundSuccess = 303 //售后成功
 
 	//标签消息
 	TagsImport = 401 //导入标签
@@ -52,7 +55,7 @@ const (
 
 	ErrMsgParamEmpty = "该类型消息对应字段不能为空"
 	ErrMsgAppIDEmpty = "该类型消息appid字段不能为空"
-	ErrMsgTpye 		 = "错误的消息类型"
+	ErrMsgTpye       = "错误的消息类型"
 )
 
 //推送下游对应字段
@@ -207,9 +210,8 @@ func (p *NsqProducer) NotifyServer(notifyServer *NotifyServer) (err error) {
 	return
 }
 
-
 /*
-	 api请求日志 20220310 gk
+ api请求日志 20220310 gk
 */
 
 func (p *NsqProducer) NotifyApiLog(notifyApiLog *NotifyApiLog) (err error) {
@@ -219,4 +221,4 @@ func (p *NsqProducer) NotifyApiLog(notifyApiLog *NotifyApiLog) (err error) {
 		logs.Info("NotifyApiLog", "消息内容:【%v】", string(jsonBytes))
 	}
 	return
-}
\ No newline at end of file
+}

notify/producer.go

@@ -11,9 +11,13 @@ type NsqProducer struct {
 	producer *nsq.Producer
 }
 
-var NsqProducers *NsqProducer
+var (
+	NsqProducers *NsqProducer
+	nsqConfig  *NsqConfig //重试
+)
 
 func New(config *NsqConfig) {
+	nsqConfig = config
 	NsqProducers = InitProducer(config)
 }
 
@@ -38,9 +42,16 @@ func InitProducer(config *NsqConfig) *NsqProducer {
 //发布消息
 func (p *NsqProducer) Publish(topic string, message string) error {
 
-	logs.Info("topic","topic:%v,message:%v",topic,message)
-
+	//logs.Info("topic","topic:%v,message:%v",topic,message)
 	var err error
+	defer func() {
+		if err != nil {
+			//重试连接
+			InitProducer(nsqConfig)
+		}
+	}()
+
+
 	if p.producer != nil {
 		if message == "" { //不能发布空串，否则会导致error
 			return nil
@@ -56,7 +67,7 @@ func (p *NsqProducer) Publish(topic string, message string) error {
 //发布延迟消息
 func (p *NsqProducer) DeferredPublish(topic string,delay time.Duration, message string) error {
 
-	logs.Info("topic",message)
+	//logs.Infof(context.Background(),"topic",message)
 
 	var err error
 	if p.producer != nil {