一、RabbitMQ 基础理解

RabbitMQ，是一个使用 erlang 编写的 AMQP（高级消息队列协议）的服务实现，简单来说，就是一个功能强大的消息队列服务。

概念理解：

• Producer: 消息发送者
• RabbitMQ:
• Vhost: 相当于分组，每个vhost下数据是隔离的
• Exchange: 路由器，接收消息，本根据RoutingKey分发消息
• headers：消息头类型 路由器，内部应用
• direct：精准匹配类型 路由器
• topic：主题匹配类型 路由器，支持正则 模糊匹配
• fanout：广播类型 路由器，RoutingKey无效
• RoutingKey: 路由规则
• Queue: 队列，用于存储消息（消息的目的地）
• Consumer: 消息消费者

持久化：

一个好的消息队列当然需要消息持久化功能，服务宕机，未消费消息不丢失，RabbitMQ持久化分为Exchange、Queue、Message

Exchange 和 Queue 持久化 指持久化Exchange、Queue 元数据，持久化的是自身，服务宕机，Exchange 和 Queue 自身就没有了

Message 持久化 顾名思义 把每一条消息体持久化，服务宕机，消息不丢失

Durable 持久、Transient 临时，Queue新建类似

分析理解:

便于更直观的理解，把 RabbitMQ 的消息流对比与Http Rest接口更家熟悉形象

www.xxx.com/webappPath/trade/getOrder -> getOrder(message) GET

RabbitMQ Server：同比 域名 www.xxx.com，只有通过域名才能到达 Server

Vhost：同比 /webappPath，一个域名可能指向多个app

Exchange：同比 /trade，trade/* 下有多个method，但是需要先到达这个Class

RoutingKey：同比 /getOrder，只有完成的 URL 才是有效的，才能确定到具体的方法 Queue：同比 getOrder(message) 消息的最终目的地

Exchange Type: 同比 GET，但是Rest MethodType是整个URL的Type，而不是 Queue

以上只是为了更好理解，千万不要混淆

Producer / Consumer 就很好理解了，基于AMQP协议链接RabbitMQ Server，发送消息 / 接收消息

二、RabbitMQ 消息确认策略分析

Confrim / Transaction 概念应用

RabbitMQ 提供了两种可靠性的确认策略 Confrim / Transaction，Producer Client仅分析Spring-Amqp，两种机制主要影响发送:

Confrim: 简单说就是直接传送消息 client > mq, 接收到 mq > ack, mq 在异步的完成 接下来的事情

Transaction: client 请求开启事务 > 发送message > client 提交事务，整个过程是同步的，mq必须完成消息持久化、消息同步等。

spring-amqp 提供的发送客户端 默认是Confrim 异步Ack模式，不用特殊配置，Transaction 需要在默认的基础上增加 RabbitMQ事务管理器

// 1.向Spring中注册RabbitMQ事务管理器

@Bean

public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory) {

return new RabbitTransactionManager(connectionFactory);

}

...

// 2.设置通道为Transaction类型

rabbitTemplate.setChannelTransacted(true);

...

// 3.对应的方法添加@Transactional

@Transactional

public void send(String exchange, String routingKey, Object object) {

rabbitTemplate.convertAndSend(exchange, routingKey, object);

}

// 这是只是举例，具体写法和其他说明，请具体看代码注释有更详细的说明和写法component.RabbitSender

Confrim / Transaction 测试分析

Consumer

消费的机制和发送差不多, 但流程变为 Consumer 处理消息，需要Ack MQ Server, Server 才会真正的删除消息，通常消费者不需要开启事务，当处理异常抛出，Ack无法发到Server到，消息就会回到队列中，继续重试，阻塞到直到消息被消费Ack掉，所说的消息阻塞

具体写法和其他说明，请具体看代码注释有更详细的说明和写法component.Receiver

三、RabbitMQ 配置

RabbitMQ 安装

RabbitMQ是基于Erlang运行的，首先选择RabbitMQ版本，确定需要的Erlang版本，然后安装Erlang，自行百度、谷歌、RabbitMQ官网或者Erlang官网都会有相应的资源、教程（ps: Erlang 版本请严格按照所选RabbitMQ版本要求的Erlang范围安装，否则会有各种不治之症）

本文以 Erlang20.03，RabbitMQ 3.7为例，RabbitMQ为linux 通用包，不同安装方式版本配置文件路径有差异，通用包好处，可移植性、控制性好

包目录结构：

./sbin/ rabbitmq 启动rabbitmq-server、插件rabbitmq-plugins、功能rabbitmqctl等脚本位置

./etc/rabbitmq/ rabbitmq 启动配置，包括随启动插件配置、环境配置、应用配置

RabbitMQ 配置文件

• rabbitmq-env.conf 环境配置 key = val 形式

# 指定节点的名字 默认 rabbit@${hostname}，如指定了节点名，需配置 host ip cluster1

NODENAME=rabbit@cluster1

# 指定端口 默认 5672

NODE_PORT=5672

# 配置持久目录

MNESIA_BASE=/mnt/data1/rabbitmq/store

# 配置日志目录 默认文件名字：${NODENAME}.log 可以用配置修改

LOG_BASE=/mnt/data1/rabbitmq/logs

• rabbitmq.conf 环境配置 key = val 形式

主要配置日志、默认用户信息、持久化相关等，没有定制化通常不用修改

# console log config 主要测试排查问题

log.console = false

log.console.level= debug

# log config

log.file= rabbit.log

log.file.level= info

log.file.rotation.date= $D0

# web port

management.listener.port= 15672

• enabled_plugins 配置相应插件的名字 server start plugin也会启动

[rabbitmq_management].

具体请看官方配置说明, 详细的讲解了rabbitmq-env.conf 和 rabbitmq.conf 配置 官方配置说明

RabbitMQ 常用命令：

# 后台启动本地服务

./rabbitmq-server –detached

# 开启/关闭 服务

./rabbitmqctl start_app {-n node_name}

./rabbitmqctl stop_app {-n node_name}

# 开启/关闭某个插件 (重启服务器后生效)

./rabbitmq-plugins enable xxx

./rabbitmq-plugins disable xxx

# 更改节点类型

./rabbitmqctl change_cluster_node_type {disc/ram} {-n node_name}

# 配置用户

./rabbitmqctl add_user username password

./rabbitmqctl change_password username newpassword

./rabbitmqctl delete_user username

./rabbitmqctl set_user_tags username administrator

Tag: none、management、policymaker、monitoring、administrator

./rabbitmqctl set_permissions -p /vhost1 username 'conf' 'write' 'read'

conf 一个正则表达式match哪些配置资源能够被该用户配置

write 一个正则表达式match哪些配置资源能够被该用户写入

read 一个正则表达式match哪些配置资源能够被该用户读取

四、高可用的集群搭建

基础概念

RabbitMQ 集群分为两种 普通集群 和 镜像集群，可以说 镜像集群 是 普通集群 的晋升版

普通集群：

以两个节点（rabbit01、rabbit02）为例来进行说明。

rabbit01和rabbit02两个节点仅有相同的元数据，即队列的结构，但消息实体只存在于其中一个节点rabbit01（或者rabbit02）中。

当消息进入rabbit01节点的Queue后，consumer从rabbit02节点消费时，RabbitMQ会临时在rabbit01、rabbit02间进行消息传输，把A中的消息实体取出并经过B发送给consumer。所以consumer应尽量连接每一个节点，从中取消息。即对于同一个逻辑队列，要在多个节点建立物理Queue。否则无论consumer连rabbit01或rabbit02，出口总在rabbit01，会产生瓶颈。当rabbit01节点故障后，rabbit02节点无法取到rabbit01节点中还未消费的消息实体。如果做了消息持久化，那么得等rabbit01节点恢复，然后才可被消费；如果没有持久化的话，就会产生消息丢失的现象。

镜像集群：

在普通集群的基础上，把需要的队列做成镜像队列，消息实体会主动在镜像节点间同步，而不是在客户端取数据时临时拉取，也就是说多少节点消息就会备份多少份。该模式带来的副作用也很明显，除了降低系统性能外，如果镜像队列数量过多，加之大量的消息进入，集群内部的网络带宽将会被这种同步通讯大大消耗掉。所以在对可靠性要求较高的场合中适用

由于镜像队列之间消息自动同步，且内部有选举master机制，即使master节点宕机也不会影响整个集群的使用，达到去中心化的目的，从而有效的防止消息丢失及服务不可用等问题

集群搭建

RabbitMQ 集群通信的验证机制是通过 erlang.cookie进行确认的，只有erlang.cookie一致的两个服务才能通信，创建cookie文件：

mkdir ~/.erlang.cookie

echo 'SJJARLYPVRPMWFVGKWZZ' > ~/.erlang.cookie

chmod 400 ~/.erlang.cookie

cluster1=10.0.0.1, cluster2=10.0.0.2 2台服务器为例，本地搭建需修改 tcp端口、web端口

写入hostes：

10.0.0.1 cluster1

10.0.0.2 cluster2

修改rabbitmq-env.conf：

# server1

NODENAME=rabbit@cluster1

...

# server2

NODENAME=rabbit@cluster2

启动server1：

./rabbitmq-server –detached

将server2加入到server形成集群：

./rabbitmqctl -n rabbit@cluster2 stop_app

# 重置元数据、集群配置等信息

./rabbitmqctl -n rabbit@cluster2 reset

# cluster2 加入到 cluster1 的集群中 --ram表示cluster2为RAM节点 默认为disc

./rabbitmqctl -n rabbit@cluster2 join_cluster rabbit@cluster1 --ram

./rabbitmqctl -n rabbit@cluster2 start_app

普通集群就搭建完成了，普通集群并不是高可用的，基于普通集群升级为镜像集群RabbitMQ HA方案

./rabbitmqctl set_policy <name> [-p <vhost>] <pattern> <definition> [--apply-to <apply-to>]

name: 策略名称

vhost: 指定vhost, 默认值 /

pattern: 需要镜像的正则

definition:

ha-mode: 指明镜像队列的模式，有效值为 all/exactly/nodes

all 表示在集群所有的节点上进行镜像，无需设置ha-params

exactly 表示在指定个数的节点上进行镜像，节点的个数由ha-params指定

nodes 表示在指定的节点上进行镜像，节点名称通过ha-params指定

ha-params: ha-mode 模式需要用到的参数

ha-sync-mode: 镜像队列中消息的同步方式，有效值为automatic，manually

apply-to: 可选值3个，默认all

exchanges 表示镜像 exchange (并不知道意义所在)

queues 表示镜像 queue

all 表示镜像 exchange和queue

eg:

./rabbitmqctl set_policy test "test" '{"ha-mode":"all","ha-sync-mode":"automatic"}'

测试: exchange = test, queue = test

case1: pattern=test, apply-to=exchanges -> 结果 exchange被镜像

case2: pattern=test, apply-to=queues -> 结果 queue被镜像

case3: pattern=test, apply-to=all -> 结果 queue被镜像

结论: 不知道exchange被镜像的意义所在，镜像queue才是关键

ps:

保证集群的高可用，至少要有1个disc节点

RabbitMQ Cluster 全部挂掉，RAM节点无法先启动，必须先启动disc节点

推荐 2 RAM 1 DISC 集群搭建方式

总结：

RabbitMQ高可用集群还是非常有必要的，高可用的代价就是性能的降低，对可靠性要求比较高的企业务还是值得的，据我测试2R1D镜像集群(非压测, 压测结果绝对更高)，达到1000QPS+还是没问题的，如果开启事务，保证同步发送应答，也可达500QPS+，绝对满足大多数可靠性要求高的业务。