fix

initalize/rabbitmq.go

@@ -1,9 +1,11 @@
 package initalize
 
 import (
+	"context"
 	"crypto/tls"
 	"errors"
 	"fusenapi/constants"
+	"fusenapi/server/websocket/consumer"
 	"github.com/streadway/amqp"
 	"github.com/zeromicro/go-zero/core/logx"
 	"log"
@@ -82,11 +84,17 @@ func (h *RabbitMqHandle) SendMsg(queueName constants.RABBIT_MQ, message []byte)
 }
 
 // 消费消息
-func (h *RabbitMqHandle) Consume(queueName constants.RABBIT_MQ, handleFunc func(data []byte) error) error {
+func (h *RabbitMqHandle) Consume(ctx context.Context, queueName constants.RABBIT_MQ, handle consumer.MqHandle) error {
 	object, ok := mapMq[queueName]
 	if !ok {
 		return errors.New("unknown queue")
 	}
+	go func() {
+		select {
+		case <-ctx.Done():
+			panic("err ctx deadline")
+		}
+	}()
 	msgs, err := object.ch.Consume(
 		object.queue.Name, // 队列名
 		object.queue.Name, // 消费者名，如果为空，则是随机生成一个
@@ -115,7 +123,7 @@ func (h *RabbitMqHandle) Consume(queueName constants.RABBIT_MQ, handleFunc func(
 				<-limit
 				wait.Done()
 			}()
-			if err = handleFunc(m.Body); err != nil {
+			if err = handle.Run(m.Body); err != nil {
 				logx.Error("failed to deal with MQ message:", string(m.Body))
 				return
 			}

server/websocket/consumer/consumer.go

@@ -0,0 +1,16 @@
+package consumer
+
+import "fmt"
+
+type MqHandle interface {
+	Run(data []byte) error
+}
+
+// 消费渲染结果数据
+type MqConsumerRenderResult struct {
+}
+
+func (m *MqConsumerRenderResult) Run(data []byte) error {
+	fmt.Println("收到消息:" + string(data))
+	return nil
+}

server/websocket/websocket.go

@@ -1,8 +1,11 @@
 package main
 
 import (
+	"context"
 	"flag"
 	"fmt"
+	"fusenapi/constants"
+	"fusenapi/server/websocket/consumer"
 	"net/http"
 
 	"fusenapi/utils/auth"
@@ -27,6 +30,11 @@ func main() {
 	defer server.Stop()
 
 	ctx := svc.NewServiceContext(c)
+	//消费组装队列
+	ctx1 := context.Background()
+	ctx2, cancel := context.WithCancel(ctx1)
+	defer cancel()
+	go ctx.RabbitMq.Consume(ctx2, constants.RABBIT_MQ_ASSEMBLE_RENDER_DATA, &consumer.MqConsumerRenderResult{})
 	handler.RegisterHandlers(server, ctx)
 	fmt.Printf("Starting server at %s:%d...\n", c.Host, c.Port)
 	server.Start()