1

2025-09-26 20:01:19 +08:00 · 2023-05-22 19:06:34 +08:00
parent 6ab19c0c2e
commit c12fa2a731
6 changed files with 251 additions and 9 deletions
--- a/go.mod
+++ b/go.mod
@@ -22,6 +22,7 @@ require (
 	github.com/nsqio/go-nsq v1.0.8
 	github.com/olivere/elastic v6.2.31+incompatible
 	github.com/pkg/errors v0.9.1
+	github.com/rabbitmq/amqp091-go v1.8.1
 	github.com/spf13/cobra v1.0.0
 	github.com/syndtr/goleveldb v1.0.0
 	github.com/usthooz/gutil v0.0.0-20220415021411-9948fc1f4e88
@@ -59,7 +60,6 @@ require (
 	github.com/patrickmn/go-cache v2.1.0+incompatible // indirect
 	github.com/sirupsen/logrus v1.4.0 // indirect
 	github.com/spf13/pflag v1.0.3 // indirect
-	github.com/stretchr/testify v1.6.1 // indirect
 	github.com/tidwall/gjson v1.13.0 // indirect
 	github.com/tidwall/match v1.1.1 // indirect
 	github.com/tidwall/pretty v1.2.0 // indirect
--- a/go.sum
+++ b/go.sum
@@ -220,6 +220,8 @@ github.com/prometheus/common v0.4.0/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y8
 github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
 github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
+github.com/rabbitmq/amqp091-go v1.8.1 h1:RejT1SBUim5doqcL6s7iN6SBmsQqyTgXb1xMlH0h1hA=
+github.com/rabbitmq/amqp091-go v1.8.1/go.mod h1:+jPrT9iY2eLjRaMSRHUhc3z14E/l85kv/f+6luSD3pc=
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
@@ -242,11 +244,13 @@ github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnIn
 github.com/spf13/viper v1.4.0/go.mod h1:PTJ7Z/lr49W6bUbkmS1V3by4uWynFiR9p7+dSq/yZzE=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
-github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
-github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/syndtr/goleveldb v1.0.0 h1:fBdIW9lB4Iz0n9khmH8w27SJ3QEJ7+IgjPEwGSZiFdE=
 github.com/syndtr/goleveldb v1.0.0/go.mod h1:ZVVdQEZoIme9iO1Ch2Jdy24qqXrMMOU6lpPAyBWyWuQ=
 github.com/tidwall/gjson v1.13.0 h1:3TFY9yxOQShrvmjdM76K+jc66zJeT6D3/VFFYCGQf7M=
@@ -273,6 +277,8 @@ go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.5.1/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
 go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
+go.uber.org/goleak v1.2.1 h1:NBol2c7O1ZokfZ0LEU9K6Whx/KnwvepVetCUhtKja4A=
+go.uber.org/goleak v1.2.1/go.mod h1:qlT2yGI9QafXHhZZLxlSuNsMw3FFLxBr+tBRlmO1xH4=
 go.uber.org/multierr v1.1.0 h1:HoEmRHQPVSqub6w2z2d2EOVs2fjyFRGyofhKuyDq0QI=
 go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
 go.uber.org/zap v1.10.0 h1:ORx85nbTijNz8ljznvCMR1ZBIPKFn3jQrag10X2AsuM=
@@ -420,8 +426,9 @@ gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
-gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gorm.io/driver/mysql v1.0.1 h1:omJoilUzyrAp0xNoio88lGJCroGdIOen9hq2A/+3ifw=
 gorm.io/driver/mysql v1.0.1/go.mod h1:KtqSthtg55lFp3S5kUXqlGaelnWpKitn4k1xZTnoiPw=
 gorm.io/gorm v1.9.19/go.mod h1:0HFTzE/SqkGTzK6TlDPPQbAYCluiVvhzoA1+aVyzenw=
--- a/myrabbitmq/myrocketmq.go
+++ b/myrabbitmq/myrocketmq.go
@@ -0,0 +1,180 @@
+package myrabbitmq
+
+import (
+	"fmt"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+	"github.com/xxjwxc/public/message"
+)
+
+type HandlerRocketMqRead func(msg []byte) error //ID , MESSAGEID,内容
+
+// MQURL 格式 amqp://账号：密码@rabbitmq服务器地址：端口号/vhost (默认是5672端口)
+// 端口可在 /etc/rabbitmq/rabbitmq-env.conf 配置文件设置，也可以启动后通过netstat -tlnp查看
+const MQURL = "amqp://admin:huan91uncc@172.21.138.131:5672/"
+
+type MyRabbitMQ struct {
+	Conn    *amqp.Connection
+	Channel *amqp.Channel
+	// 队列名称
+	QueueName string
+	// 交换机
+	Exchange string
+	// routing Key
+	RoutingKey string
+	//MQ链接字符串
+	Mqurl string
+}
+
+// NewRabbitMQ 创建结构体实例
+func NewRabbitMQ(mqurl string) (*MyRabbitMQ, error) {
+	rabbitMQ := MyRabbitMQ{
+		Mqurl: mqurl,
+	}
+	var err error
+	//创建rabbitmq连接
+	rabbitMQ.Conn, err = amqp.Dial(rabbitMQ.Mqurl)
+	if err != nil {
+		return nil, err
+	}
+
+	//创建Channel
+	rabbitMQ.Channel, err = rabbitMQ.Conn.Channel()
+	if err != nil {
+		return nil, err
+	}
+
+	return &rabbitMQ, nil
+}
+
+// 释放资源,建议NewRabbitMQ获取实例后 配合defer使用
+func (mq *MyRabbitMQ) Close() error {
+	if mq.Conn == nil || mq.Channel == nil {
+		return message.GetError(message.InValidOp)
+	}
+
+	err := mq.Conn.Close()
+	if err != nil {
+		return err
+	}
+
+	return mq.Channel.Close()
+}
+
+// NewProducer 创建生产者
+func (mq *MyRabbitMQ) NewProducer(queueName, exchange, routingKey string) (*amqp.Channel, error) {
+	mq.QueueName = queueName
+	mq.Exchange = exchange
+	mq.RoutingKey = routingKey
+	// 1.声明队列
+	/*
+	  如果只有一方声明队列，可能会导致下面的情况：
+	   a)消费者是无法订阅或者获取不存在的MessageQueue中信息
+	   b)消息被Exchange接受以后，如果没有匹配的Queue，则会被丢弃
+
+	  为了避免上面的问题，所以最好选择两方一起声明
+	  ps:如果客户端尝试建立一个已经存在的消息队列，Rabbit MQ不会做任何事情，并返回客户端建立成功的
+	*/
+	_, err := mq.Channel.QueueDeclare( // 返回的队列对象内部记录了队列的一些信息，这里没什么用
+		mq.QueueName, // 队列名
+		true,         // 是否持久化
+		false,        // 是否自动删除(前提是至少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时，才会自动删除。注意：生产者客户端创建这个队列，或者没有消费者客户端与这个队列连接时，都不会自动删除这个队列)
+		false,        // 是否为排他队列（排他的队列仅对“首次”声明的conn可见[一个conn中的其他channel也能访问该队列]，conn结束后队列删除）
+		false,        // 是否阻塞
+		nil,          // 额外属性（我还不会用）
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	// 2.声明交换器
+	err = mq.Channel.ExchangeDeclare(
+		mq.Exchange, //交换器名
+		"topic",     //exchange type：一般用fanout、direct、topic
+		true,        // 是否持久化
+		false,       // 是否自动删除（自动删除的前提是至少有一个队列或者交换器与这和交换器绑定，之后所有与这个交换器绑定的队列或者交换器都与此解绑）
+		false,       // 设置是否内置的。true表示是内置的交换器，客户端程序无法直接发送消息到这个交换器中，只能通过交换器路由到交换器这种方式
+		false,       // 是否阻塞
+		nil,         // 额外属性
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	// 3.建立Binding(可随心所欲建立多个绑定关系)
+	err = mq.Channel.QueueBind(
+		mq.QueueName,  // 绑定的队列名称
+		mq.RoutingKey, // bindkey 用于消息路由分发的key
+		mq.Exchange,   // 绑定的exchange名
+		false,         // 是否阻塞
+		nil,           // 额外属性
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	return mq.Channel, nil
+}
+
+// SendMessage 发送消息(level 代表延迟级别)
+func (mq *MyRabbitMQ) SendMessage(msg []byte) error {
+	// 4.发送消息
+	return mq.Channel.Publish(
+		mq.Exchange,   // 交换器名
+		mq.RoutingKey, // routing key
+		false,         // 是否返回消息(匹配队列)，如果为true, 会根据binding规则匹配queue，如未匹配queue，则把发送的消息返回给发送者
+		false,         // 是否返回消息(匹配消费者)，如果为true, 消息发送到queue后发现没有绑定消费者，则把发送的消息返回给发送者
+		amqp.Publishing{ // 发送的消息，固定有消息体和一些额外的消息头，包中提供了封装对象
+			ContentType: "text/plain", // 消息内容的类型
+			Body:        msg,          // 消息内容
+		},
+	)
+}
+
+// 消费者
+func (mq *MyRabbitMQ) NewConsumer(queueName string) (<-chan amqp.Delivery, error) {
+	mq.QueueName = queueName
+	// 1.声明队列（两端都要声明，原因在生产者处已经说明）
+	_, err := mq.Channel.QueueDeclare( // 返回的队列对象内部记录了队列的一些信息，这里没什么用
+		mq.QueueName, // 队列名
+		true,         // 是否持久化
+		false,        // 是否自动删除(前提是至少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时，才会自动删除。注意：生产者客户端创建这个队列，或者没有消费者客户端与这个队列连接时，都不会自动删除这个队列)
+		false,        // 是否为排他队列（排他的队列仅对“首次”声明的conn可见[一个conn中的其他channel也能访问该队列]，conn结束后队列删除）
+		false,        // 是否阻塞
+		nil,          // 额外属性（我还不会用）
+	)
+	if err != nil {
+		fmt.Println("声明队列失败", err)
+		return nil, err
+	}
+
+	// 2.从队列获取消息（消费者只关注队列）consume方式会不断的从队列中获取消息
+	msgChanl, err := mq.Channel.Consume(
+		mq.QueueName, // 队列名
+		"",           // 消费者名，用来区分多个消费者，以实现公平分发或均等分发策略
+		false,        // 是否自动应答
+		false,        // 是否排他
+		false,        // 是否接收只同一个连接中的消息，若为true，则只能接收别的conn中发送的消息
+		false,        // 队列消费是否阻塞
+		nil,          // 额外属性
+	)
+	if err != nil {
+		fmt.Println("获取消息失败", err)
+		return nil, err
+	}
+
+	return msgChanl, nil
+}
+
+func (m *MyRabbitMQ) Start(msgChanl <-chan amqp.Delivery, hand HandlerRocketMqRead) error { // 阻塞模式
+	for msg := range msgChanl {
+		// 这里写你的处理逻辑
+		// 获取到的消息是amqp.Delivery对象，从中可以获取消息信息
+		if hand(msg.Body) == nil {
+			msg.Ack(true) // 主动应答
+		}
+
+	}
+
+	return nil
+}
--- a/myrabbitmq/myrocketmq_test.go
+++ b/myrabbitmq/myrocketmq_test.go
@@ -0,0 +1,57 @@
+package myrabbitmq
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/xxjwxc/public/message"
+)
+
+func Test_NewAdmin(t *testing.T) {
+	topic := "xxjtest"
+	tag := "tagtest"
+	host := "amqp://admin:admin@192.155.1.151:5672/"
+	group := "nlp_cmd_train"
+	// 初始化mq
+	mq, err := NewRabbitMQ(host)
+	if err != nil {
+		fmt.Println(err)
+	}
+	defer mq.Close() // 完成任务释放资源
+	_, err = mq.NewProducer(topic, tag, group)
+	fmt.Println(err)
+	// go func() {
+	// 	for i := 0; i < 1000; i++ {
+	// 		mq.SendMessage([]byte(fmt.Sprintf("this is xxj test %v", i)))
+	// 		time.Sleep(1 * time.Microsecond)
+	// 	}
+	// }()
+	// time.Sleep(10 * time.Minute)
+
+	go func() {
+		ch, err := mq.NewConsumer(topic) // 消費者
+		fmt.Println(err)
+		mq.Start(ch, func(msg []byte) error {
+			fmt.Println("消费者1", string(msg))
+			time.Sleep(1 * time.Second)
+			return nil
+		})
+		time.Sleep(10 * time.Minute)
+	}()
+
+	go func() {
+		ch, err := mq.NewConsumer(topic) // 消費者
+		fmt.Println(err)
+		mq.Start(ch, func(msg []byte) error {
+			fmt.Println("消费者2", string(msg))
+			time.Sleep(1 * time.Second)
+			return message.GetError(message.ActvFailure)
+		})
+		time.Sleep(10 * time.Minute)
+
+	}()
+
+	time.Sleep(10 * time.Minute)
+	mq.Close()
+}
--- a/myrocketmq/myrocketmq.go
+++ b/myrocketmq/myrocketmq.go
@@ -44,10 +44,10 @@ func (m *MyRocketAdmin) CreateTopic(topic string, readQueueNums int, writeQueueN

 func (m *MyRocketAdmin) FetchPublishMessageQueues(topic string) ([]*primitive.MessageQueue, error) {
 	if m.admin == nil {
-		return message.GetError(message.StateError)
+		return nil, message.GetError(message.StateError)
 	}
-	
-	return m.admin.FetchPublishMessageQueues(context.Background(), admin.WithTopicCreate(topic))
+
+	return m.admin.FetchPublishMessageQueues(context.Background(), topic)
 }

 func (m *MyRocketAdmin) Close() error {
--- a/myrocketmq/myrocketmq_test.go
+++ b/myrocketmq/myrocketmq_test.go
@@ -15,8 +15,6 @@ func Test_NewAdmin(t *testing.T) {
 	fmt.Println(err)
 	adm.CreateTopic(topic, 0, 0)

-	
-
 	pwd, err := NewProducer(host, group, 2) // 生产者
 	fmt.Println(err)
 	go func() {