add: init mq

cmd/follow/dal/mq/follow_mq.go

@@ -74,20 +74,30 @@ func (f *FollowMQ) Consume(ctx context.Context) (<-chan amqp.Delivery, error) {
 }
 
 // TODO:sync resolve msg
+func (s *SyncFollow) SyncFollowMQ(ctx context.Context) error {
+	defer FollowMQCli.ReleaseRes()
+	msgs, err := FollowMQCli.Consume(ctx)
+	if err != nil {
+		return err
+	}
 
-// type SyncFollow struct {
-// }
+	var forever chan struct{}
 
-// func (s *SyncFollow) RunTaskCreate(ctx context.Context) error {
-// 	msgs, err := FollowMQCli.Consume(ctx)
-// 	if err != nil {
-// 		return err
-// 	}
-// 	var forever chan struct{}
+	go func() {
+		for msg := range msgs {
+			klog.Infof("Resolve msg: %s", msg.Body)
+			// TODO:落库处理
+			msg.Ack(false)
+		}
+	}()
 
-// 	//TODO:
+	<-forever
 
-// 	<-forever
+	return nil
+}
 
-// 	return nil
-// }
+// 释放资源,建议获取实例后配合defer使用
+func (f *FollowMQ) ReleaseRes() {
+	f.conn.Close()
+	f.channel.Close()
+}

cmd/follow/dal/mq/init.go

@@ -1,6 +1,8 @@
 package mq
 
 import (
+	"context"
+
 	"github.com/cloudwego/kitex/pkg/klog"
 	"github.com/ozline/tiktok/pkg/utils"
 	amqp "github.com/rabbitmq/amqp091-go"
@@ -18,6 +20,9 @@ type FollowMQ struct {
 	queueName string
 }
 
+type SyncFollow struct {
+}
+
 var (
 	Rmq         *RabbitMQ
 	FollowMQCli *FollowMQ
@@ -43,6 +48,9 @@ func Init() {
 		klog.Error(err)
 		return
 	}
+
+	ctx := context.Background()
+	go run(ctx)
 }
 
 func FollowMQInit() (*FollowMQ, error) {
@@ -57,3 +65,11 @@ func FollowMQInit() (*FollowMQ, error) {
 		queueName: "FollowQueue", // 加入consts
 	}, nil
 }
+
+func run(ctx context.Context) {
+	fSync := new(SyncFollow)
+	err := fSync.SyncFollowMQ(ctx)
+	if err != nil {
+		klog.Infof("RunTaskCreate:%s", err)
+	}
+}