From 1a48fa95f6ee967edd5bd0f4c5695df190ae0641 Mon Sep 17 00:00:00 2001 From: DengBiao <2319963317@qq.com> Date: Fri, 30 Sep 2022 12:01:14 +0800 Subject: [PATCH] update --- README.md | 57 +++++++++++++++++++++++++++++++++ main.go | 47 +++++++++++++++++++++++++++ rabbit/pool.go | 14 +++++++++ rabbitmq/tools.go | 11 +++++-- test/fanout_exchange_test.go | 61 ++++++++++++++++++++++++++++-------- test/single_queue_test.go | 53 +++++++++++++++++++++++++------ 6 files changed, 219 insertions(+), 24 deletions(-) create mode 100644 main.go diff --git a/README.md b/README.md index e69de29..06dfca5 100644 --- a/README.md +++ b/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中的消费者。 \ No newline at end of file diff --git a/main.go b/main.go new file mode 100644 index 0000000..1de51d0 --- /dev/null +++ b/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 +} diff --git a/rabbit/pool.go b/rabbit/pool.go index 7c2ec4c..909e4ef 100644 --- a/rabbit/pool.go +++ b/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) { diff --git a/rabbitmq/tools.go b/rabbitmq/tools.go index 6701055..e7358bc 100644 --- a/rabbitmq/tools.go +++ b/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() diff --git a/test/fanout_exchange_test.go b/test/fanout_exchange_test.go index cacd5aa..e6a7162 100644 --- a/test/fanout_exchange_test.go +++ b/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) { diff --git a/test/single_queue_test.go b/test/single_queue_test.go index 69e18fa..70c0420 100644 --- a/test/single_queue_test.go +++ b/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)) +}