消息队列（RabbitMq）概要

1.本文档主要按照本人的理解介绍RabbitMq的功能、如何使用。
2.关于RabbitMq的各种使用场景以及与其他同类产品的横向、纵向对比请自行百度。
3.消息队列看起来貌似非常复杂，感觉很麻烦，其实通过本项目骨架封装之后，使用非常简单，开发者安装rabbitmq（类似安装mysql），配置好账号、密码、端口即可快速使用.
4.消息队列的两个核心角色：生产者(通常是一次性投递消息)，消费者(需要一直处于阻塞状态监听、接受、处理消息)。

rabbitmq涉及到的几个概念

1.Broker（消息代理者）：接收和分发消息的应用，RabbitMQ Server就是Message Broker。
2.Virtual host（虚拟主机）: 类似于mysql的数据库，当多个不同的用户使用同一个RabbitMQ server提供的服务时，可以划分出多个vhost，每个用户在自己的vhost创建exchange／queue等。
3.Connection（连接）: 生产者、消费者和broker之间的TCP连接。
4.Channel（通道）: 如果每一次访问RabbitMQ都建立一个Connection，在消息量大的时候建立TCP Connection的开销将是巨大的，效率也较低。Channel是在connection内部建立的逻辑连接Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销。
5.Exchange（交换机）: message到达broker的第一站，根据分发规则，分发消息到queue中去。常用的类型有：direct (定向消息)、topic (主题、指定键名) 、fanout (广播消息)。
6.Queue（队列）: 消息最终被送到这里等待consumer取走。一个message可以被同时拷贝到多个queue中。

快速安装步骤(基于docker)

1.比较详细的安装的参考地址：http://note.youdao.com/noteshare?id=3d8850a96ed288a0ae5c5421206b0f4e&sub=62EAE38FE217410E8D70859A152BCF8F
2.安装rabbitMq可以理解为安装一个mysql,默认创建的账号可以理解为 root，可以直接操作rabbitmq.
3.为了项目更安全，可以登录后台地址(http://IP:15672),自行为项目创建虚拟主机(类似mysql的数据库)、账号，最后将账号允许的操作虚拟进行绑定即可.

RabbitMq常用的几种模式

在线代码预览

消息队列（RabbitMq） - 图1

1.hello_world模式(最基本模式)， 特点如下：

1 一个生产者（producer）、一个消费者（consumer）通过队列（queue）进行 一对一 的数据传输。
2.使用非常简单，适合简单业务场景使用，相关的场景模型图：
消息队列（RabbitMq） - 图2

2.WorkQueue模式（在消费者之间按照竞争力分配任务）， 特点如下：

1 生产者（producer）、多个消费者（consumer）通过队列（queue）进行一对多、多对多的数据传输。
2.生产者（producer）将消息发布到交换机（exchange）的某个队列（queue），多个消费者（consumer）其中只要有一个消费（取走）了消息，那么其他消费者（consumer）将不会重复获得。
3.消费者支持设置更多的参数，使配置强的消费者可以多处理消息，配置低的可以少处理消息，做到尽其所能，资源最大化利用。
消息队列（RabbitMq） - 图3

3.publish/subscribe模式（同时向许多消费者发送消息）， 特点如下：

1 生产者（producer）、多个消费者（consumer）通过队列（queue）进行一对多、多对多的数据传输。
2.生产者（producer）将消息发布到交换机（exchange）的某个队列（queue），多个消费者（consumer）处理消息。
3.该模式也叫作广播（broadcast）、扇形（fanout）、发布/订阅模式，消费者（consumer）可以通过配置，接收来自生产者（consumer）发送的全部消息；或者每种消费者只接收指定队列的消息，将生产者发送的消息进行分类（按照不同的队列）处理。
消息队列（RabbitMq） - 图4

4.routing模式（有选择性地接收消息）， 特点如下：

1 生产者（producer）、多个消费者（consumer）通过队列（queue）进行一对多、多对多的数据传输。
2.生产者（producer）将消息发布到交换机（exchange）已经绑定好路由键的某个队列（queue），多个消费者（consumer）可以通过绑定的路由键获取消息、处理消息。
3.该模式下，消息的分类应该应该明确、种类数量不是非常多，那么就可以指定路由键（key）、绑定的到交换器的队列实现消息精准投递。
消息队列（RabbitMq） - 图5

5.topics模式（基于主题接收消息）， 特点如下：

1 该模式就是routing模式的加强版，由原来的路由键精确匹配模式升级现在的模糊匹配模式。
2.语法层面主要表现为灵活的匹配规则：
2.1 # 表示匹配一个或多个任意字符；
2.2 *表示匹配一个字符；
2.3 .（点）本身无实际意义，不表示任何匹配规则，主要用于将关键词分隔开，它的左边或右边可以写匹配规则，例如：abc.# 表示匹配 abc张三 、abc你好 等；#.abc.# 表示匹配路由键中含有abc的字符；
3.注意：匹配语法中如果没有 .（点），那么匹配规则是无效的，例如：orange#，可能本意是匹配orange任意字符，实际上除了匹配 orange#本身之外，什么也匹配不到。
消息队列（RabbitMq） - 图6

6.RPC模式（请求、回复）， 特点如下：

1 严格地说，该模式和消息队列没有什么关系，通常是微服务场景才会使用远程过程调用（RPC），本功能建议自行学习或者选择专业的微服务框架使用，解决实际问题，本文档不做介绍。
消息队列（RabbitMq） - 图7

RabbitMq快速使用指南

1.建议使用docker 快速安装使用即可，安装步骤请自行搜索。
2.详细使用指南参见单元测试demo代: rabbitmq全量单元测试
3.六种场景模型我们封装了统一的使用规范。

1.hello_world、work_queue、publish_subscribe 场景模型使用：

相关配置参见：config/config.yaml, rbbitmq 部分

1.1 启动一个消费者，通过回调函数在阻塞模式进行消息处理
  1. consumer, err := HelloWorld.CreateConsumer()
  2.     if err != nil {
  3.         fmt.Printf("HelloWorld单元测试未通过。%s\n", err.Error())
  4.         os.Exit(1)
  5.     }
  7.     // 连接关闭的回调，主要是记录错误，进行后续更进一步处理，不要尝试在这里编写重连逻辑
  8.     // 本项目已经封装了完善的消费者端重连逻辑，触发这里的代码说明重连已经超过了最大重试次数
  9.     consumer.OnConnectionError(func(err *amqp.Error) {
  10.         log.Fatal(MyErrors.ErrorsRabbitMqReconnectFail + "\n" + err.Error())
  11.     })
  13.     // 进入阻塞状态，处理消息
  14.     consumer.Received(func(received_data string) {
  15.         fmt.Printf("HelloWorld回调函数处理消息：--->%s\n", received_data)
  16.     })

1.2 调用生产者投递一个或者多个消息，投递通常都是一次性的。
  1.     // 这里创建场景模型的时候通过不同的模型名称创建即可，主要有：hello_world、work_queue、publish_subscribe 
  2.     hello_producer, _ := hello_world.CreateProducer()
  3.     var res bool
  4.     for i := 0; i < 10; i++ {
  5.         str := fmt.Sprintf("%d_hello_world开始发送消息测试", (i + 1))
  6.         // 参数解释：
  7.         // 参数一： 需要发送的消息
  8.         // 参数二：延迟毫秒，只有延迟模式才有效。
  9.         res = hello_producer.Send(str, 0)
  10.         //time.Sleep(time.Second * 1)
  11.     }
  13.     hello_producer.Close() // 消息投递结束，必须关闭连接
  14.     // 简单判断一下最后一次发送结果
  15.     if res {
  16.         fmt.Printf("消息发送OK")
  17.     } else {
  18.         fmt.Printf("消息发送 失败")
  19.     }

2.routing、topics 场景模型使用：

routing模式属于路由键的严格匹配模式。
topics模式比routing模式更灵活，两者使用、功能几乎完全一致。该模式完全可以代替routing模式，因此这里仅介绍 topics模式。
注意：生产者设置键的规则必须是：关键词A.关键词B.关键词C等，即关键词之间必须使用.（点）隔开，消费者端只需要将.（点）左边或右边的关键词使用#代替即可。

2.1 启动多个消费者，处于阻塞模式进行消息接受、处理。
  1.     // 启动第一个消费者，这里使用协程的目的主要是保证第一个启动后不阻塞，否则就会导致第二个消费者无法启动
  2.     go func(){
  3.         consumer, err := Topics.CreateConsumer()
  5.         if err != nil {
  6.             t.Errorf("Routing单元测试未通过。%s\n", err.Error())
  7.             os.Exit(1)
  8.         }
  10.     // 连接关闭的回调，主要是记录错误，进行后续更进一步处理，不要尝试在这里编写重连逻辑
  11.     // 本项目已经封装了完善的消费者端重连逻辑，触发这里的代码说明重连已经超过了最大重试次数
  12.         consumer.OnConnectionError(func(err *amqp.Error) {
  13.             log.Fatal(MyErrors.ErrorsRabbitMqReconnectFail + "\n" + err.Error())
  14.         })
  16.         // 通过route_key 模糊匹配队列路由键的消息来处理
  17.         consumer.Received("#.even", func(received_data string) {
  18.             fmt.Printf("模糊匹配偶数键：--->%s\n", received_data)
  19.         })
  20.     }()
  22.         // 启动第二个消费者，这里没有使用协程，在消息处理环节程序就会阻塞等待，处理消息
  23.         consumer, err := Topics.CreateConsumer()
  25.         if err != nil {
  26.             t.Errorf("Routing单元测试未通过。%s\n", err.Error())
  27.             os.Exit(1)
  28.         }
  30.         consumer.OnConnectionError(func(err *amqp.Error) {
  31.         // 连接关闭的回调，主要是记录错误，进行后续更进一步处理，不要尝试在这里编写重连逻辑
  32.         // 本项目已经封装了完善的消费者端重连逻辑，触发这里的代码说明重连已经超过了最大重试次数
  33.             log.Fatal(MyErrors.ErrorsRabbitMqReconnectFail + "\n" + err.Error())
  34.         })
  36.         // 通过route_key 模糊匹配队列路由键的消息来处理
  37.         consumer.Received("#.odd", func(received_data string) {
  39.             fmt.Printf("模糊匹配奇数键：--->%s\n", received_data)
  40.         })

2.2 调用生产者投递一个或者多个消息
  2.     producer, _ := Topics.CreateProducer()
  3.     var res bool
  4.     var key string
  5.     for i := 1; i <= 10; i++ {
  7.         //  将 偶数 和  奇数 分发到不同的key，消费者端，启动两个也各自处理偶数和奇数
  8.         if i%2 == 0 {
  9.             key = "key.even" //  偶数键
  10.         } else {
  11.             key = "key.odd" //  奇数键
  12.         }
  13.         str_data := fmt.Sprintf("%d_Routing_%s, 开始发送消息测试", i, key)
  15.         // 参数解释：
  16.         // 参数一： key 键名
  17.         // 参数二： 需要发送的消息
  18.         // 参数三：延迟毫秒，只有延迟模式才有效。
  19.         res = producer.Send(key, str_data, 0)
  20.         //time.Sleep(time.Second * 1)
  21.     }
  23.     producer.Close() // 消息投递结束，必须关闭连接
  25.     // 简单判断一下最后一次发送结果
  26.     if res {
  27.         fmt.Printf("消息发送OK")
  28.     } else {
  29.         fmt.Printf("消息发送 失败")
  30.     }
  31.     //Output: 消息发送OK

关于消费者端如何启动、监听问题


1.与项目骨架捆绑启动

开发完成消费者代码，在程序启动处bootstrap/init.go ，通过调用你的 rabbitmq 客户端初始化命令启动，该模式相当于与本项目骨架捆绑启动。
示例代码：

  1. 1.项目启动入口调用 rabbitmq 客户端/消费者端 的初始化命令
  2. // 入口调用 rabbitmq 客户端的初始化命令
  3. func init() {
  5. // 省略其他代码
  7. // 初始化消息推送rabbitMq
  8. go service_demo.CreateRabbitMqFactory().StartHandleMsg()
  10. //省略其他代码
  12. }
  1. rabbitmq 的消费者端初始化示例代码，上文中出现的 service_demo.CreateRabbitMqFactory().StartHandleMsg() 中的 service_demo代码如下： ```go
  2. func CreateRabbitMqFactory() *rmqWsHandleMsg { return &rmqWsHandleMsg{} }

type rmqWsHandleMsg struct{}

func (r *rmqWsHandleMsg) StartHandleMsg() { // 根据实际业务创建一个也定类型的 rabbitmq 客户端（消费者端） consumer, err := work_queue.CreateConsumer() if err != nil { variable.ZapLog.Error(“创建RabbiMq 消费者端出错”, zap.Error(err)) return }

// 发生错误做好日志记录、以及后续通知等 consumer.OnConnectionError(func(err *amqp.Error) {

  1. variable.ZapLog.Error(my_errors.ErrorsRabbitMqReconnectFail + "\n" + err.Error())

})

// 该函数会进入阻塞状态，处理 rabbitmq 接受到的消息： receivedData consumer.Received(func(receivedData string) { fmt.Println(“消费者端接受到的消息：”,receivedData)

  1. })

} ```

2.独立启动

本项目骨架引入了cobra包，在 cli模式结合 commond目录创建相关功能分类、入口文件，调用相关的消费者初始化命令，独立编译、启动。
本质上就是在 cli模式调用一下初始化 消费者端 的代码，进入阻塞监听模式即可。