cmb

Dockerfile

@@ -5,8 +5,7 @@ ENV GOCACHE /root/.cache/go-build
 COPY . /src
 WORKDIR /src
 
-
+RUN make build
-RUN  make build
 
 FROM registry.cn-chengdu.aliyuncs.com/lansexiongdi/work:v1
 

configs/config.yaml

@@ -59,16 +59,15 @@ cmb:
   notifyUrl: "https://sandbox.cdcc.cmbchina.com/AccessGateway/transIn/updateCodeStatus.json" # 招行测试回调地址
 
 wechatNotifyMQ:
-  isOpenConsumer: false #是否启动消费 true/false
   accessKeyId: "LTAI5tPyV7FynQNTfEvbEBuX"
   accessKeySecret: "tZmTh8cV98xAQgtlRU0soWcb6Tpd4T"
   endPoint: "http://1389288909295870.mqrest.cn-hangzhou.aliyuncs.com"
   regionId: "hangzhou"
   instanceId: "MQ_INST_1389288909295870_BYSoMttI"
   topic: "notify"
-  groupId: "market_pro"
+  groupId: "GID_market_pro"
   tag: "voucher_notify_dev"
-  debug: false
+  isOpenConsumer: false #是否启动消费 true/false
   registerTagUrl: "https://wpcallbacks.api.1688sup.com/wechatPay/register_tag"
 
 #配置日志

internal/conf/conf.proto

@@ -93,7 +93,6 @@ message WechatNotifyMQ {
   string topic = 6;
   string tag = 7;
   string groupId = 8;
-  bool debug = 9;
   string registerTagUrl = 10;
   bool isOpenConsumer = 11;
 }

internal/data/mq.go

@@ -2,12 +2,9 @@ package data
 
 import (
 	"fmt"
-	mq_http_sdk "github.com/aliyunmq/mq-http-go-sdk"
 	"github.com/apache/rocketmq-client-go/v2/primitive"
 	"github.com/apache/rocketmq-client-go/v2/producer"
 	"github.com/go-kratos/kratos/v2/log"
-	"strconv"
-	"time"
 	"voucher/internal/conf"
 	"voucher/internal/pkg/mq"
 )
@@ -36,8 +33,10 @@ func buildMqProducer(c *conf.RocketMQ) (*mq.Producer, error) {
 	if c == nil {
 		return nil, nil
 	}
+
 	var p *mq.Producer
 	var err error
+
 	if c.AccessKey != "" && c.SecretKey != "" {
 		p, err = mq.NewProducer(c.Addr, producer.WithCredentials(primitive.Credentials{
 			AccessKey: c.AccessKey,
@@ -46,62 +45,18 @@ func buildMqProducer(c *conf.RocketMQ) (*mq.Producer, error) {
 	} else {
 		p, err = mq.NewProducer(c.Addr)
 	}
+
 	if err != nil {
 		fmt.Println("创建 rocketMQ producer 失败: ", err)
 		return nil, err
 	}
+
 	err = p.Start()
 	if err != nil {
 		fmt.Println("rocketMQ producer start 失败: ", err)
 		return nil, err
 	}
+
 	//此时并没有发起连接，在使用时才会
 	return p, nil
 }
-
-func wechatNotifyProducer() {
-	// 设置HTTP协议客户端接入点，进入消息队列RocketMQ版控制台实例详情页面的接入点区域查看。
-	endpoint := "http://1389288909295870.mqrest.cn-hangzhou.aliyuncs.com"
-	// 请确保环境变量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已设置。
-	// AccessKey ID，阿里云身份验证标识。
-	accessKey := "LTAI5tPyV7FynQNTfEvbEBuX"
-	// AccessKey Secret，阿里云身份验证密钥。
-	secretKey := "tZmTh8cV98xAQgtlRU0soWcb6Tpd4T"
-	// 消息所属的Topic，在消息队列RocketMQ版控制台创建。
-	topic := "notify"
-	// Topic所属的实例ID，在消息队列RocketMQ版控制台创建。
-	// 若实例有命名空间，则实例ID必须传入；若实例无命名空间，则实例ID传入null空值或字符串空值。实例的命名空间可以在消息队列RocketMQ版控制台的实例详情页面查看。
-	instanceId := "MQ_INST_1389288909295870_BYSoMttI"
-
-	tag := "voucher_notify_dev"
-	//tag := "voucher_notify_pro"
-
-	client := mq_http_sdk.NewAliyunMQClient(endpoint, accessKey, secretKey, "")
-
-	mqProducer := client.GetProducer(instanceId, topic)
-
-	// 循环发送2条消息。
-	for i := 0; i < 2; i++ {
-		var msg mq_http_sdk.PublishMessageRequest
-
-		msg = mq_http_sdk.PublishMessageRequest{
-			MessageBody: "hello mq!",         //消息内容。
-			MessageTag:  tag,                 // 消息标签。
-			Properties:  map[string]string{}, // 消息属性。
-		}
-		// 设置消息的Key。
-		msg.MessageKey = "MessageKey"
-		// 设置消息自定义属性。
-		msg.Properties["a"] = strconv.Itoa(i)
-
-		ret, err := mqProducer.PublishMessage(msg)
-
-		if err != nil {
-			fmt.Println(err)
-			return
-		} else {
-			fmt.Printf("Publish ---->\n\tMessageId:%s, BodyMD5:%s, \n", ret.MessageId, ret.MessageBodyMD5)
-		}
-		time.Sleep(time.Duration(100) * time.Millisecond)
-	}
-}

internal/data/mq_test.go

@@ -76,7 +76,3 @@ func Test_NotifyProducer(t *testing.T) {
 		return
 	}
 }
-
-func Test_WechatNotifyProducer(t *testing.T) {
-	wechatNotifyProducer()
-}

internal/pkg/mq_http/mq_http.go

@@ -0,0 +1,54 @@
+package mq_http
+
+import (
+	"fmt"
+	mq_http_sdk "github.com/aliyunmq/mq-http-go-sdk"
+	"strconv"
+)
+
+func wechatNotifyProducer(bodyStr string) error {
+	// 设置HTTP协议客户端接入点，进入消息队列RocketMQ版控制台实例详情页面的接入点区域查看。
+	endpoint := "http://1389288909295870.mqrest.cn-hangzhou.aliyuncs.com"
+	// 请确保环境变量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已设置。
+	// AccessKey ID，阿里云身份验证标识。
+	accessKey := "LTAI5tPyV7FynQNTfEvbEBuX"
+	// AccessKey Secret，阿里云身份验证密钥。
+	secretKey := "tZmTh8cV98xAQgtlRU0soWcb6Tpd4T"
+	// 消息所属的Topic，在消息队列RocketMQ版控制台创建。
+	topic := "notify"
+	// Topic所属的实例ID，在消息队列RocketMQ版控制台创建。
+	// 若实例有命名空间，则实例ID必须传入；若实例无命名空间，则实例ID传入null空值或字符串空值。实例的命名空间可以在消息队列RocketMQ版控制台的实例详情页面查看。
+	instanceId := "MQ_INST_1389288909295870_BYSoMttI"
+
+	tag := "voucher_notify_dev"
+	//tag := "voucher_notify_pro"
+
+	client := mq_http_sdk.NewAliyunMQClient(endpoint, accessKey, secretKey, "")
+
+	mqProducer := client.GetProducer(instanceId, topic)
+
+	// 循环发送2条消息。
+	for i := 0; i < 1; i++ {
+		var msg mq_http_sdk.PublishMessageRequest
+
+		msg = mq_http_sdk.PublishMessageRequest{
+			MessageBody: bodyStr,             //消息内容。
+			MessageTag:  tag,                 // 消息标签。
+			Properties:  map[string]string{}, // 消息属性。
+			MessageKey:  "MessageKey",        // 设置消息的Key。
+		}
+
+		// 设置消息自定义属性。
+		msg.Properties["a"] = strconv.Itoa(i)
+
+		ret, err := mqProducer.PublishMessage(msg)
+
+		if err != nil {
+			return err
+		}
+
+		fmt.Printf("Publish ---->\n\tMessageId:%s, BodyMD5:%s, \n", ret.MessageId, ret.MessageBodyMD5)
+	}
+
+	return nil
+}

internal/pkg/mq_http/mq_http_test.go

@@ -0,0 +1,13 @@
+package mq_http
+
+import "testing"
+
+func Test_WechatNotifyProducer(t *testing.T) {
+
+	bodyStr := `{"id":"5465699d-de6a-5414-a8df-283167b577ca","create_time":"2025-03-07T15:57:24+08:00","resource_type":"encrypt-resource","event_type":"COUPON.USE","summary":"代金券核销通知","original_type":"coupon","associated_data":"coupon","plain_text":{"stock_creator_mchid":"1605446142","stock_id":"19990623","coupon_id":"95952277058","coupon_name":"萧山农商新客激活礼","description":"","status":"USED","create_time":"2025-03-07T15:49:31+08:00","coupon_type":"NORMAL","no_cash":false,"singleitem":false,"consume_information":{"consume_time":"2025-03-07T15:57:24+08:00","consume_mchid":"1800002761","transaction_id":"4200002544202503077103159055"}`
+	err := wechatNotifyProducer(bodyStr)
+	if err != nil {
+		t.Errorf("入队失败 error = %v", err)
+		return
+	}
+}

internal/server/wechat_notify_consume.go

@@ -28,25 +28,57 @@ func (w *WechatNotifyConsumer) Start(ctx context.Context) error {
 		return nil
 	}
 
+	//// 设置HTTP协议客户端接入点，进入消息队列RocketMQ版控制台实例详情页面的接入点区域查看。
+	//endpoint := w.conf.WechatNotifyMQ.EndPoint
+	//// 请确保环境变量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已设置。
+	//// AccessKey ID，阿里云身份验证标识。
+	//accessKey := w.conf.WechatNotifyMQ.AccessKeyId
+	//// AccessKey Secret，阿里云身份验证密钥。
+	//secretKey := w.conf.WechatNotifyMQ.AccessKeySecret
+	//// 消息所属的Topic，在消息队列RocketMQ版控制台创建。
+	////不同消息类型的Topic不能混用，例如普通消息的Topic只能用于收发普通消息，不能用于收发其他类型的消息。
+	//topic := w.conf.WechatNotifyMQ.Topic
+	//// Topic所属的实例ID，在消息队列RocketMQ版控制台创建。
+	//// 若实例有命名空间，则实例ID必须传入；若实例无命名空间，则实例ID传入null空值或字符串空值。实例的命名空间可以在消息队列RocketMQ版控制台的实例详情页面查看。
+	//instanceId := w.conf.WechatNotifyMQ.InstanceId
+	//// 您在控制台创建的Group ID。
+	//groupId := w.conf.WechatNotifyMQ.GroupId
+	//
+	//tag := w.conf.WechatNotifyMQ.Tag
+
+	/**
+	ACCESS_KEY_ID=LTAI5tPyV7FynQNTfEvbEBuX
+	ACCESS_KEY_SECRET=tZmTh8cV98xAQgtlRU0soWcb6Tpd4T
+	END_POINT=http://1389288909295870.mqrest.cn-hangzhou.aliyuncs.com
+	REGION_ID=cn-hangzhou
+	INSTANCE_ID= MQ_INST_1389288909295870_BYSoMttI
+	TOPIC=notify
+	TAG=coupon_usage_notify_market
+	REGISTER_TAG_URL= https://wpcallbacks.api.1688sup.com/wechatPay/register_tag
+	GROUP_ID=market_pro
+	DEBUG=false
+	*/
+
 	// 设置HTTP协议客户端接入点，进入消息队列RocketMQ版控制台实例详情页面的接入点区域查看。
-	endpoint := w.conf.WechatNotifyMQ.EndPoint
+	endpoint := "http://1389288909295870.mqrest.cn-hangzhou.aliyuncs.com"
 	// 请确保环境变量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已设置。
 	// AccessKey ID，阿里云身份验证标识。
-	accessKey := w.conf.WechatNotifyMQ.AccessKeyId
+	accessKeyId := "LTAI5tPyV7FynQNTfEvbEBuX"
 	// AccessKey Secret，阿里云身份验证密钥。
-	secretKey := w.conf.WechatNotifyMQ.AccessKeySecret
+	accessKeySecret := "tZmTh8cV98xAQgtlRU0soWcb6Tpd4T"
 	// 消息所属的Topic，在消息队列RocketMQ版控制台创建。
-	//不同消息类型的Topic不能混用，例如普通消息的Topic只能用于收发普通消息，不能用于收发其他类型的消息。
+	topic := "notify"
-	topic := w.conf.WechatNotifyMQ.Topic
 	// Topic所属的实例ID，在消息队列RocketMQ版控制台创建。
 	// 若实例有命名空间，则实例ID必须传入；若实例无命名空间，则实例ID传入null空值或字符串空值。实例的命名空间可以在消息队列RocketMQ版控制台的实例详情页面查看。
-	instanceId := w.conf.WechatNotifyMQ.InstanceId
+	instanceId := "MQ_INST_1389288909295870_BYSoMttI"
-	// 您在控制台创建的Group ID。
-	groupId := w.conf.WechatNotifyMQ.GroupId
 
-	tag := w.conf.WechatNotifyMQ.Tag
+	//groupId := "market_pro"
+	groupId := "GID_market_pro"
 
-	client := mq_http_sdk.NewAliyunMQClient(endpoint, accessKey, secretKey, "")
+	tag := "voucher_notify_dev"
+	//tag := "voucher_notify_pro"
+
+	client := mq_http_sdk.NewAliyunMQClient(endpoint, accessKeyId, accessKeySecret, "")
 
 	mqConsumer := client.GetConsumer(instanceId, topic, groupId, tag)
 
@@ -63,19 +95,8 @@ func (w *WechatNotifyConsumer) Start(ctx context.Context) error {
 					fmt.Printf("Consume %d messages---->\n", len(resp.Messages))
 					for _, v := range resp.Messages {
 						handles = append(handles, v.ReceiptHandle)
-						fmt.Printf("\tMessageID: %s, PublishTime: %d, MessageTag: %s\n"+
+						log.Warnf("接收消息成功 wechat notify messageTag:%s, message: %s", v.MessageTag, v.MessageBody)
-							"\tConsumedTimes: %d, FirstConsumeTime: %d, NextConsumeTime: %d\n"+
+						fmt.Printf("接收消息成功 wechat notify messageTag:%s, message: %s", v.MessageTag, v.MessageBody)
-							"\tBody: %s\n"+
-							"\tProps: %s\n",
-							v.MessageId,
-							v.PublishTime,
-							v.MessageTag,
-							v.ConsumedTimes,
-							v.FirstConsumeTime,
-							v.NextConsumeTime,
-							v.MessageBody,
-							v.Properties,
-						)
 					}
 
 					// NextConsumeTime前若不确认消息消费成功，则消息会被重复消费。
@@ -83,13 +104,13 @@ func (w *WechatNotifyConsumer) Start(ctx context.Context) error {
 					ackerr := mqConsumer.AckMessage(handles)
 					if ackerr != nil {
 						// 某些消息的句柄可能超时，会导致消息消费状态确认不成功。
-						fmt.Println(ackerr)
 						if errAckItems, ok := ackerr.(errors.ErrCode).Context()["Detail"].([]mq_http_sdk.ErrAckItem); ok {
 							for _, errAckItem := range errAckItems {
-								fmt.Printf("\tErrorHandle:%s, ErrorCode:%s, ErrorMsg:%s\n",
+								log.Errorf("\tErrorHandle:%s, ErrorCode:%s, ErrorMsg:%s\n", errAckItem.ErrorHandle, errAckItem.ErrorCode, errAckItem.ErrorMsg)
-									errAckItem.ErrorHandle, errAckItem.ErrorCode, errAckItem.ErrorMsg)
+								fmt.Printf("\tErrorHandle:%s, ErrorCode:%s, ErrorMsg:%s\n", errAckItem.ErrorHandle, errAckItem.ErrorCode, errAckItem.ErrorMsg)
 							}
 						} else {
+							log.Errorf("ackerr:%+v\n", ackerr)
 							fmt.Println("ack err =", ackerr)
 						}
 						time.Sleep(time.Duration(3) * time.Second)
@@ -121,14 +142,14 @@ func (w *WechatNotifyConsumer) Start(ctx context.Context) error {
 		// 长轮询消费消息，网络超时时间默认为35s。
 		// 长轮询表示如果Topic没有消息，则客户端请求会在服务端挂起3s，3s内如果有消息可以消费则立即返回响应。
 		mqConsumer.ConsumeMessage(respChan, errChan,
-			3, // 一次最多消费3条（最多可设置为16条）。
+			3,  // 一次最多消费3条（最多可设置为16条）。
-			3, // 长轮询时间3s（最多可设置为30s）。
+			10, // 长轮询时间3s（最多可设置为30s）。
 		)
 		<-endChan
 	}
 }
 
-func (w WechatNotifyConsumer) Stop(ctx context.Context) error {
+func (w *WechatNotifyConsumer) Stop(ctx context.Context) error {
 	fmt.Println("关闭 wechat consumer 中...")
 	return nil
 }