fix kq client

app/mqueue/cmd/rpc/etc/mqueue.yaml

@@ -24,6 +24,14 @@ Redis:
   Type: node
   Pass: G62m50oigInC30sf
 
-KqServerConf:
+#支付成功回调通知
+KqPaymentUpdatePayStatusConf:
   Brokers:
     - kafka:9092
+  Topic: payment-update-paystatus-topic
+
+#修改订单状态后发送小程序通知用户
+KqSendWxMiniTplMessageConf:
+  Brokers:
+    - kafka:9092
+  Topic: send-wx-mini-tpl-message

app/mqueue/cmd/rpc/internal/config/config.go

@@ -5,7 +5,6 @@ import "github.com/zeromicro/go-zero/zrpc"
 type Config struct {
 	zrpc.RpcServerConf
 
-	KqServerConf struct {
-		Brokers []string
-	}
+	KqPaymentUpdatePayStatusConf KqConfig
+	KqSendWxMiniTplMessageConf KqConfig
 }

app/mqueue/cmd/rpc/internal/config/kqConfig.go

@@ -0,0 +1,6 @@
+package config
+
+type KqConfig struct {
+	Brokers []string
+	Topic   string
+}

app/mqueue/cmd/rpc/internal/logic/kqPaymenStatusUpdateLogic.go

@@ -2,6 +2,9 @@ package logic
 
 import (
 	"context"
+	"github.com/pkg/errors"
+	"looklook/common/xerr"
+	"encoding/json"
 
 	"looklook/app/mqueue/cmd/rpc/internal/svc"
 	"looklook/app/mqueue/cmd/rpc/pb"
@@ -32,7 +35,13 @@ func (l *KqPaymenStatusUpdateLogic) KqPaymenStatusUpdate(in *pb.KqPaymenStatusUp
 		PayStatus: in.PayStatus,
 	}
 
-	if err := l.svcCtx.KqueueClient.Push(kqueue.PAYMENT_UPDATE_PAYSTATUS, m); err != nil {
+	//2、序列化
+	body, err := json.Marshal(m)
+	if err != nil {
+		return nil, errors.Wrapf(xerr.NewErrMsg("kq kqPaymenStatusUpdateLogic  task marshal error "), "kq kqPaymenStatusUpdateLogic  task marshal error , v : %+v", m)
+	}
+
+	if err := l.svcCtx.KqueuePaymentUpdatePayStatusClient.Push(string(body)); err != nil {
 		return nil, err
 	}
 

app/mqueue/cmd/rpc/internal/logic/sendWxMiniSubMessageLogic.go

@@ -2,6 +2,9 @@ package logic
 
 import (
 	"context"
+	"github.com/pkg/errors"
+	"looklook/common/xerr"
+	"encoding/json"
 
 	"looklook/app/mqueue/cmd/rpc/internal/svc"
 	"looklook/app/mqueue/cmd/rpc/pb"
@@ -34,7 +37,13 @@ func (l *SendWxMiniSubMessageLogic) SendWxMiniSubMessage(in *pb.SendWxMiniSubMes
 		Page:       in.Page,
 	}
 
-	if err := l.svcCtx.KqueueClient.Push(kqueue.SEND_WX_MINI_TPL_MESSAGE, m); err != nil {
+	//2、序列化
+	body, err := json.Marshal(m)
+	if err != nil {
+		return nil, errors.Wrapf(xerr.NewErrMsg("kq sendWxMiniSubMessageLogic task marshal error "), "kq sendWxMiniSubMessageLogic task marshal error , v : %+v", m)
+	}
+
+	if err := l.svcCtx.KqueueSendWxMiniTplMessageClient.Push(string(body)); err != nil {
 		return nil, err
 	}
 

app/mqueue/cmd/rpc/internal/svc/serviceContext.go

@@ -1,25 +1,28 @@
 package svc
 
 import (
-	"looklook/app/mqueue/cmd/rpc/internal/config"
-	"looklook/common/kqueue"
-
 	"github.com/hibiken/asynq"
+	"github.com/zeromicro/go-queue/kq"
+	"looklook/app/mqueue/cmd/rpc/internal/config"
 )
 
 type ServiceContext struct {
 	Config config.Config
 
-	KqueueClient kqueue.KqueueClient
 	AsynqClient  *asynq.Client
+
+	KqueuePaymentUpdatePayStatusClient *kq.Pusher
+	KqueueSendWxMiniTplMessageClient   *kq.Pusher
+
 }
 
 func NewServiceContext(c config.Config) *ServiceContext {
 
 	svc := &ServiceContext{
 		Config:       c,
-		KqueueClient: kqueue.NewKqueueSvcClient(c.KqServerConf.Brokers),
 		AsynqClient:  asynq.NewClient(asynq.RedisClientOpt{Addr: c.Redis.Host, Password: c.Redis.Pass}),
+		KqueuePaymentUpdatePayStatusClient: kq.NewPusher(c.KqPaymentUpdatePayStatusConf.Brokers,c.KqPaymentUpdatePayStatusConf.Topic),
+		KqueueSendWxMiniTplMessageClient: kq.NewPusher(c.KqSendWxMiniTplMessageConf.Brokers,c.KqSendWxMiniTplMessageConf.Topic),
 	}
 
 	return svc