From 1ee1fd6e5b89b47626aad07dde601286aa26bc15 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E5=AD=90=E9=93=AD?= Date: Fri, 7 Mar 2025 16:28:59 +0800 Subject: [PATCH] cmb --- Dockerfile | 3 +- configs/config.yaml | 5 +- internal/conf/conf.proto | 1 - internal/data/mq.go | 55 ++--------------- internal/data/mq_test.go | 4 -- internal/pkg/mq_http/mq_http.go | 54 ++++++++++++++++ internal/pkg/mq_http/mq_http_test.go | 13 ++++ internal/server/wechat_notify_consume.go | 79 +++++++++++++++--------- 8 files changed, 125 insertions(+), 89 deletions(-) create mode 100644 internal/pkg/mq_http/mq_http.go create mode 100644 internal/pkg/mq_http/mq_http_test.go diff --git a/Dockerfile b/Dockerfile index 424251f..0668fe5 100644 --- a/Dockerfile +++ b/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 diff --git a/configs/config.yaml b/configs/config.yaml index b76c112..e514c98 100644 --- a/configs/config.yaml +++ b/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" #配置日志 diff --git a/internal/conf/conf.proto b/internal/conf/conf.proto index e8c9c5e..f32402f 100644 --- a/internal/conf/conf.proto +++ b/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; } diff --git a/internal/data/mq.go b/internal/data/mq.go index 0d027a2..295faec 100644 --- a/internal/data/mq.go +++ b/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) - } -} diff --git a/internal/data/mq_test.go b/internal/data/mq_test.go index 0105c2d..ed69ef7 100644 --- a/internal/data/mq_test.go +++ b/internal/data/mq_test.go @@ -76,7 +76,3 @@ func Test_NotifyProducer(t *testing.T) { return } } - -func Test_WechatNotifyProducer(t *testing.T) { - wechatNotifyProducer() -} diff --git a/internal/pkg/mq_http/mq_http.go b/internal/pkg/mq_http/mq_http.go new file mode 100644 index 0000000..a411d36 --- /dev/null +++ b/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 +} diff --git a/internal/pkg/mq_http/mq_http_test.go b/internal/pkg/mq_http/mq_http_test.go new file mode 100644 index 0000000..69c4189 --- /dev/null +++ b/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 + } +} diff --git a/internal/server/wechat_notify_consume.go b/internal/server/wechat_notify_consume.go index 9ce945d..61d7b37 100644 --- a/internal/server/wechat_notify_consume.go +++ b/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 := w.conf.WechatNotifyMQ.Topic + topic := "notify" // Topic所属的实例ID，在消息队列RocketMQ版控制台创建。 // 若实例有命名空间，则实例ID必须传入；若实例无命名空间，则实例ID传入null空值或字符串空值。实例的命名空间可以在消息队列RocketMQ版控制台的实例详情页面查看。 - instanceId := w.conf.WechatNotifyMQ.InstanceId - // 您在控制台创建的Group ID。 - groupId := w.conf.WechatNotifyMQ.GroupId + instanceId := "MQ_INST_1389288909295870_BYSoMttI" - 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"+ - "\tConsumedTimes: %d, FirstConsumeTime: %d, NextConsumeTime: %d\n"+ - "\tBody: %s\n"+ - "\tProps: %s\n", - v.MessageId, - v.PublishTime, - v.MessageTag, - v.ConsumedTimes, - v.FirstConsumeTime, - v.NextConsumeTime, - v.MessageBody, - v.Properties, - ) + log.Warnf("接收消息成功 wechat notify messageTag:%s, message: %s", v.MessageTag, v.MessageBody) + fmt.Printf("接收消息成功 wechat notify messageTag:%s, message: %s", v.MessageTag, v.MessageBody) } // 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", - errAckItem.ErrorHandle, errAckItem.ErrorCode, errAckItem.ErrorMsg) + log.Errorf("\tErrorHandle:%s, ErrorCode:%s, ErrorMsg:%s\n", 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, // 长轮询时间3s（最多可设置为30s）。 + 3, // 一次最多消费3条（最多可设置为16条）。 + 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 }