update

README.md

@@ -0,0 +1,57 @@
+# 智莺RabbitMq模块包
+## 版本
+`RabbitMQ 3.8.8` `Erlang 22.0.7`
+> RabbitMQ 必须基于Erlang 环境才能运行，不同版本的RabbitMq对Erlang 版本要求不同。
+## 两种设计使用模式
+> `直连模式:`每次使用Mq，单独连接，单独开辟信道进行操作 [简单、易控、可用于测试及部分对tcp请求不频繁的情况(如单脚本)]。
+
+> `连接池模式:`
+>> - 每次使用Mq，多连接，多信道 [高复用、稳定、节约资源、避免异常情况产生、可用于生产环境中对mq推拉操作频繁的情况]。
+>> - rabbitmq 官网建议,多线程尽量共享连接，独享信道(大部分连接包目前只实现了单连接的多信道连接池，这种方式在高并发时，单链接就不够用了)。
+>> 考虑到高并发时的情况，同时结合实际mq情况，把tcp控制在合理范围内，可以适当放宽请求连接数。
+>> - 实现了多连接多信道的连接池、实现了连接过期时间设置(连接的最后一个信道释放/关闭时，对应的连接如果过期，才进行关闭)、日志打印，可以利用具体框架实现的日志模块打印，默认是标准输出。
+
+
+## 功能
+- 支持 `简单(simple)` 、 `工作(work)` 、 `订阅/发布(fanout)` 、 `路由routing(direct)` 、 `部分匹配(topic)` 等常用mq工作模式类型
+- 支持 两种身份的日常操作 [生产者(product)、消费者(consume)]
+- 生产者使用很简单，这里不做赘诉（获取连接池内的信道后直接进行相应mq操作接口）。
+- 消费者这里支持两种模式`推模式`、`拉模式` [推模式下将信道置为接收模式，直到取消队列的订阅为止。在接收模式期间，RabbitMQ会不断地推送消息给消费者（当然推送消息的个数还是会受到Qos的限制）。如果只想从队列获得单条消息而不是持续订阅，建议还是使用拉模式。但是不能将Get方法放在一个循环里来代替推模式，这样做会严重影响RabbitMQ的性能。如果要实现高吞吐，消费者理应使用推模式)]。
+
+## 层级介绍
+
+- `md` 结构体
+- `rabbit` 封装了mq连接池及日常操作的处理方法
+- `rabbitmq` 封装了mq直接连接及日常操作的处理方法
+- `test` 提供了mq基本使用几种模式的使用测试用例及关于模式的介绍md文档
+- `utils` 工具类
+
+## 留言
+### 存储机制
+RabbitMQ消息有两种类型:
+- 持久化消息和非持久化消息。
+- 这两种消息都会被写入磁盘。
+- 持久化消息在到达队列时写入磁盘，同时会内存中保存一份备份，当内存吃紧时，消息从内存中清除。这会提高一定的性能。 非持久化消息一般只存于内存中，当内存压力大时数据刷盘处理，以节省内存空间。
+
+![image-20220113153420444](/images/md/storage.png)
+
+> 惰性队列：RabbitMQ从3.6.0版本开始引入了惰性队列（Lazy Queue）的概念。惰性队列会尽可能的将消息存入磁盘中，而在消费者消费到相应的消息时才会被加载到内存中，它的一个重要的设计目标是能够支持更长的队列，即支持更多的消息存储。当消费者由于各种各样的原因（比如消费者下线、宕机亦或者是由于维护而关闭等）而致使长时间内不能消费消息造成堆积时，惰性队列就很有必要了。
+
+### 名词释义
+- `Broker`:broker 是指一个或多个 erlang node 的逻辑分组，且 node 上运行着 RabbitMQ 应用程序（可以理解为mq服务器的意思，或者再简单点就是mq节点的意思）。
+- `Channel`:信道
+- `Message`:消息，由消息头和消息体组成
+- `Binding`:绑定，用于建立Exchange和Queue之间的关联
+- `Queue`:消息队列
+- `Exchange`: 交换器/路由器
+- `ACK`:ACK机制是消费者从RabbitMQ收到消息并处理完成后，反馈给RabbitMQ，MQ收到反馈后才将此消息从队列中删除。消息的ACK确认机制默认是打开的。
+- `QOS`:RabbitMQ提供了一种qos（服务质量保证）功能，即在非自动确认消息的前提下，如果一定数目的消息（通过基于consume或者channel设置Qos的值）未被确认前，不进行消费新的消息。
+这种机制一方面可以实现限速（将消息暂存到RabbitMQ内存中）的作用，一方面可以保证消息确认质量（比如确认了但是处理有异常的情况）。
+- `Durable（持久化）`: RabbitMq对交换器，队列，消息都可以声明持久化属性，交换器和队列持久化属性为durable（其属性值为false代表不持久化，属性值为true代表持久化），消息持久化属性为deliveryMode（其属性值为1代表不持久化，属性值为2代表持久化）。
+> - RabbitMq实例以broker表示，当broker重启时，所有未申明durable的交换器和队列都会被删除。
+> - RabbitMq中消息都被保存在队列中，所以如果队列都删除了，消息不管有没有设置deliveryMode=2都不管用了。
+> - 综上所述，可以得出的结论为，交换器未声明durable属性不会影响队列的持久化（但是发送方的producer会被影响，无法正常发送消息）；只声明队列持久化，重启之后消息会丢失；只声明消息的持久化，重启之后消息随队列一起丢失。单单设置消息持久化而不设置队列的持久化没有任何意义。
+- `autoDelete`:RabbitMq在声明交换器和队列时，都有一个共同的属性为autoDelete(自动删除)，autoDelete属性针对的是曾经有过但后来没有的事物（除非常不重要且临时性数据，否则建议设置为`false`，不使用）。
+- `exclusive（排他性）`:RabbitMq中真正处理消息是队列，自然exclusive也只对队列生效。声明了exclusive属性的队列只对首次声明它的连接可见，并且在连接断开时自动删除。`强烈建议，非极端特殊情况，不要使用该属性`
+- `noWait`: 是否非阻塞，true表示是。阻塞：表示创建交换器的请求发送后，阻塞等待MQ Server返回信息。非阻塞：不会阻塞等待MQ
+- `noLocal`: 是否本地化，设置为true表示不能将同一个connection中生产者发送的消息传递给这个connection中的消费者。

main.go

@@ -0,0 +1,47 @@
+package main
+
+import (
+	"code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbitmq"
+	"fmt"
+	"log"
+)
+
+func main() {
+	var host, port, user, pwd string
+	log.SetFlags(log.Lshortfile | log.LstdFlags)
+	log.Println("input host,port,user,pwd")
+	fmt.Scanf("%s %s %s %s", &host, &port, &user, &pwd)
+	if host == "" && port == "" && user == "" && pwd == "" {
+		host = "119.23.182.117"
+		port = "5672"
+		user = "admin"
+		pwd = "123456"
+	}
+	//ch, err := GetChannel
+	rabbitmq.Init(host, port, user, pwd)
+	//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
+	receiveMq := rabbitmq.New(rabbitmq.Cfg.Uri, "test_single_queue")
+	//for {
+	receiveMq.Qos(1)
+	//接收消息时，指定
+	messages := receiveMq.Consume()
+
+	//forever := make(chan bool) //开辟管道
+	//启用协程处理消息
+	//go func() {
+	var i = 0
+	for msg := range messages {
+		log.Printf("Received a message: %s", msg.Body)
+
+		msg.Ack(false)
+		i++
+		if i >= 5 {
+			receiveMq.CloseChannel()
+			return
+		}
+		//msg.Reject(true)
+	}
+	//}()
+	log.Printf("[*] Waiting for message,To exit press CTRL+C")
+	//<-forever
+}

rabbit/pool.go

@@ -283,6 +283,20 @@ func (ch *Channel) QueueDelete(queue string) {
 	failOnError(e, "删除队列失败！")
 }
 
+//初始化队列
+func (ch *Channel) NewQueue(name string, typename string) {
+	q, e := ch.Channel.QueueDeclare(
+		name,  //队列名
+		false, //是否开启持久化
+		false, //不使用时删除
+		false, //排他
+		false, //不等待
+		nil,   //参数
+	)
+	failOnError(e, "初始化队列失败！")
+	ch.Name = q.Name
+}
+
 // NewExchange 初始化交换机
 //s：rabbitmq服务器的链接，name：交换机名字，typename：交换机类型
 func NewExchange(s string, name string, typename string) {

rabbitmq/tools.go

@@ -39,7 +39,7 @@ func New(s string, name string) *RabbitMQ {
 	q, e := ch.QueueDeclare(
 		name,  //队列名
 		false, //是否开启持久化
-		true,  //不使用时删除
+		false, //不使用时删除
 		false, //排他
 		false, //不等待
 		nil,   //参数
@@ -55,7 +55,7 @@ func New(s string, name string) *RabbitMQ {
 
 // QueueDeclare 声明交换机
 func (q *RabbitMQ) QueueDeclare(queue string) {
-	_, e := q.channel.QueueDeclare(queue, false, true, false, false, nil)
+	_, e := q.channel.QueueDeclare(queue, false, false, false, false, nil)
 	failOnError(e, "声明交换机！")
 }
 
@@ -162,6 +162,13 @@ func (q *RabbitMQ) Consume() <-chan amqp.Delivery {
 	return c
 }
 
+//拉取某个消息队列的消息
+func (q *RabbitMQ) Get() (messages amqp.Delivery, ok bool, err error) {
+	messages, ok, err = q.channel.Get(q.Name, false)
+	failOnError(err, "拉取消息失败！")
+	return
+}
+
 // 关闭信道
 func (q *RabbitMQ) CloseChannel() {
 	q.channel.Close()

test/fanout_exchange_test.go

@@ -2,6 +2,7 @@ package test
 
 import (
 	"code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbitmq"
+	"fmt"
 	"log"
 	"strconv"
 	"strings"
@@ -12,8 +13,20 @@ import (
 const FanoutExchangeName = "test_fanout_exchange"
 
 func TestFanoutExchangeSend(t *testing.T) {
-	ch := rabbitmq.Connect("amqp://user:password@ip:port/")
-	rabbitmq.NewExchange("amqp://user:password@ip:port/", FanoutExchangeName, "fanout")
+	var host, port, user, pwd string
+	log.SetFlags(log.Lshortfile | log.LstdFlags)
+	log.Println("input host,port,user,pwd")
+	fmt.Scanf("%s %s %s %s", &host, &port, &user, &pwd)
+	if host == "" && port == "" && user == "" && pwd == "" {
+		host = "119.23.182.117"
+		port = "5672"
+		user = "admin"
+		pwd = "123456"
+	}
+	rabbitmq.Init(host, port, user, pwd)
+	ch := rabbitmq.Connect(rabbitmq.Cfg.Uri)
+	rabbitmq.NewExchange(rabbitmq.Cfg.Uri, FanoutExchangeName, "fanout")
+
 	i := 0
 	for {
 		time.Sleep(1)
@@ -28,23 +41,45 @@ func TestFanoutExchangeReceive1(t *testing.T) {
 	// 2.创建交换机
 	// 3.将自己绑定到交换机上
 	// 4.接收交换机上发过来的消息
-
+	var host, port, user, pwd string
+	log.SetFlags(log.Lshortfile | log.LstdFlags)
+	log.Println("input host,port,user,pwd")
+	fmt.Scanf("%s %s %s %s", &host, &port, &user, &pwd)
+	if host == "" && port == "" && user == "" && pwd == "" {
+		host = "119.23.182.117"
+		port = "5672"
+		user = "admin"
+		pwd = "123456"
+	}
+	rabbitmq.Init(host, port, user, pwd)
 	//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
-	receiveMq := rabbitmq.New("amqp://user:password@ip:port/", "test_fanout_exchange_receive_queue_1")
+	receiveMq := rabbitmq.New(rabbitmq.Cfg.Uri, "test_fanout_exchange_receive_queue_1")
 	//第一个参数：rabbitmq服务器的链接，第二个参数：交换机名字，第三个参数：交换机类型
-	rabbitmq.NewExchange("amqp://user:password@ip:port/", FanoutExchangeName, "fanout")
+	rabbitmq.NewExchange(rabbitmq.Cfg.Uri, FanoutExchangeName, "fanout")
 	// 队列绑定到exchange
 	receiveMq.Bind(FanoutExchangeName, "")
 	//4
-	for {
-		//接收消息时，指定
-		msgs := receiveMq.Consume()
-		go func() {
-			for d := range msgs {
-				log.Printf("recevie1  Received a message: %s", d.Body)
+	forever := make(chan bool) //开辟管道
+	var i = 0
+	//for {
+	//接收消息时，指定
+	messages := receiveMq.Consume()
+	go func() {
+		for msg := range messages {
+			log.Printf("recevie1  Received a message: %s", msg.Body)
+			i++
+			if i >= 5 {
+				msg.Reject(true)
+				receiveMq.CloseChannel()
+				return
+			} else {
+				msg.Ack(false)
 			}
-		}()
-	}
+		}
+	}()
+	log.Printf("[*] Waiting for message,To exit press CTRL+C")
+	<-forever
+	//}
 }
 
 func TestFanoutExchangeReceive2(t *testing.T) {

test/single_queue_test.go

@@ -19,7 +19,7 @@ func TestSend(t *testing.T) {
 	log.Println("input host,port,user,pwd")
 	fmt.Scanf("%s %s %s %s", &host, &port, &user, &pwd)
 	if host == "" && port == "" && user == "" && pwd == "" {
-		host = "10.10.140.138"
+		host = "119.23.182.117"
 		port = "5672"
 		user = "admin"
 		pwd = "123456"
@@ -29,7 +29,7 @@ func TestSend(t *testing.T) {
 	sendMq := rabbitmq.New(rabbitmq.Cfg.Uri, SingleQueueName)
 	i := 0
 	for {
-		if i > 100 {
+		if i > 99 {
 			break
 		}
 		//time.Sleep(time.Second * 1)
@@ -45,7 +45,7 @@ func TestReceive(t *testing.T) {
 	log.Println("input host,port,user,pwd")
 	fmt.Scanf("%s %s %s %s", &host, &port, &user, &pwd)
 	if host == "" && port == "" && user == "" && pwd == "" {
-		host = "10.10.140.138"
+		host = "119.23.182.117"
 		port = "5672"
 		user = "admin"
 		pwd = "123456"
@@ -55,19 +55,24 @@ func TestReceive(t *testing.T) {
 	//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
 	receiveMq := rabbitmq.New(rabbitmq.Cfg.Uri, SingleQueueName)
 	//for {
-	receiveMq.Qos(5)
+	receiveMq.Qos(1)
 	//接收消息时，指定
 	messages := receiveMq.Consume()
 
 	forever := make(chan bool) //开辟管道
+	var i = 0
 	//启用协程处理消息
 	go func() {
 		for msg := range messages {
 			log.Printf("Received a message: %s", msg.Body)
-
-			msg.Ack(false)
-
-			//msg.Reject(true)
+			i++
+			if i >= 5 {
+				msg.Reject(true)
+				receiveMq.CloseChannel()
+				return
+			} else {
+				msg.Ack(false)
+			}
 		}
 	}()
 	log.Printf("[*] Waiting for message,To exit press CTRL+C")
@@ -80,7 +85,7 @@ func TestReceive1(t *testing.T) {
 	log.Println("input host,port,user,pwd")
 	fmt.Scanf("%s %s %s %s", &host, &port, &user, &pwd)
 	if host == "" && port == "" && user == "" && pwd == "" {
-		host = "10.10.140.138"
+		host = "119.23.182.117"
 		port = "5672"
 		user = "admin"
 		pwd = "123456"
@@ -109,3 +114,33 @@ func TestReceive1(t *testing.T) {
 	//}
 	fmt.Println("get msg done")
 }
+
+func TestReceive2(t *testing.T) {
+	var host, port, user, pwd string
+	log.SetFlags(log.Lshortfile | log.LstdFlags)
+	log.Println("input host,port,user,pwd")
+	fmt.Scanf("%s %s %s %s", &host, &port, &user, &pwd)
+	if host == "" && port == "" && user == "" && pwd == "" {
+		host = "119.23.182.117"
+		port = "5672"
+		user = "admin"
+		pwd = "123456"
+	}
+	//ch, err := GetChannel
+	rabbitmq.Init(host, port, user, pwd)
+	//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
+	receiveMq := rabbitmq.New(rabbitmq.Cfg.Uri, SingleQueueName)
+	//for {
+	receiveMq.Qos(1)
+	//接收消息时，指定
+	messages, ok, err := receiveMq.Get()
+	if err != nil {
+		t.Fatalf("Failed get: %v", err)
+	}
+
+	if !ok {
+		t.Fatalf("Get on a queued message did not find the message")
+	}
+	fmt.Println(">>>>>>ok>>>>>>", ok)
+	fmt.Println(">>>>>>messages>>>>>>", string(messages.Body))
+}