阶段七模块二 RabbitMQ
内容输出来源：拉钩教育Java就业训练营
面试题：避免消息堆积
- workqueue，多个消费者监听同一个队列
- 接收到消息后，通过线程池，异步消费
1 RabbitMQ
1.1 MQ（Message Queue）消息队列
消息队列中间件，是分布式系统中的重要组件
主要解决，异步处理，应用解耦，流量削峰等问题
从而实现高性能，高可用，可伸缩和最终一致性的架构
1.1.1 异步处理
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-7BLezibG-1618660866150)(RabbitMQ.assets/RabbitMQ详解.jpg)]
1.1.2 应用解耦
订单系统需要通知库存系统，如果库存系统异常，则订单调用库存失败，导致下单失败
订单系统和库存系统耦合度太高
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-6hEK1Xvs-1618660866152)(RabbitMQ.assets/RabbitMQ详解-1618448182873.jpg)]
下单的时候，库存系统不能正常运行，也不会影响下单，因为下单后，订单系统写入消息队列就不再关心其他的后续操作了，实现了订单系统和库存系统的应用解耦；
1.1.3 流量削峰
抢购，秒杀等业务，针对高并发的场景
因为流量过大，暴增会导致应用挂掉，为解决这个问题，在前端加入消息队列
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-lzNpU3SR-1618660866153)(RabbitMQ.assets/RabbitMQ详解-1618448372074.jpg)]
1.2 RabbitMQ各组件
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-TQMV618e-1618660866155)(RabbitMQ.assets/RabbitMQ详解-1618448854663.jpg)]
- Broker：消息队列服务器实体
-
Virtual Host：虚拟主机
- 标识一批交换机、消息队列和相关对象，形成的整体
- 虚拟主机是共享相同的身份认证和加密环境的独立服务器域
- 每个vhost本质上就是一个mini版的RabbitMQ服务器，拥有自己的队列、交换器、绑定和权限机制
- vhost是AMQP概念的基础，RabbitMQ默认的vhost是 /，必须在链接时指定
-
Exchange：交换器（路由）
- 用来接收生产者发送的消息并将这些消息路由给服务器中的队列
-
Queue：消息队列
- 用来保存消息直到发送给消费者。
- 它是消息的容器，也是消息的终点。
- 一个消息可投入一个或多个队列。
- 消息一直在队列里面，等待消费者连接到这个队列将其取走。
- Banding：绑定，用于消息队列和交换机之间的关联。
-
Channel：通道（信道）
- 多路复用连接中的一条独立的双向数据流通道。
- 信道是建立在真实的TCP连接内的 虚拟链接
- AMQP命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，都是通过信道完成的
- 因为对于操作系统来说，建立和销毁TCP连接都是非常昂贵的开销，所以引入了信道的概念，用来复用TCP连接。
- Connection：网络连接，比如一个TCP连接。
- Publisher：消息的生产者，也是一个向交换器发布消息的客户端应用程序。
- Consumer：消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。
-
Message：消息
- 消息是不具名的，它是由消息头和消息体组成。
- 消息体是不透明的，而消息头则是由一系列的可选属性组成，这些属性包括routing-key(路由键)、priority(优先级)、delivery-mode(消息可能需要持久性存储[消息的路由模式])等。
2 RabbitMQ安装及使用
2.1 安装及启动
2.1.1 安装
rpm -ivh erlang-21.3.8.16-1.el7.x86_64.rpm
rpm -ivh socat-1.7.3.2-5.el7.lux.x86_64.rpm
rpm -ivh rabbitmq-server-3.8.6-1.el7.noarch.rpm
2.1.2 启动后台管理插件
rabbitmq-plugins enable rabbitmq_management
2.1.3 启动RabbitMq
systemctl start rabbitmq-server.service
systemctl status rabbitmq-server.service
systemctl restart rabbitmq-server.service
systemctl stop rabbitmq-server.service
2.1.4 查看进程
ps -ef | grep rabbitmq
2.1.5 测试
关闭防火墙
systemctl stop firewalld
浏览器输入：http://ip:15672
1.创建账号
rabbitmqctl add_user laosun 123456
2.设置用户角色
rabbitmqctl set_user_tags laosun administrator
3.设置用户权限
rabbitmqctl set_permissions -p "/" laosun ".*"
".*" ".*"
4.查看当前用户和角色
rabbitmqctl list_users
5.更改密码
rabbitmqctl change_password wx 123123
管理界面端口：
5672：RabbitMQ提供给编程语言客户端链接的端口
15672：RabbitMQ管理界面的端口
25672：RabbitMQ集群的端口
2.2 java与RabbitMQ连接
<dependencies>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.7.3</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.25</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.9</version>
</dependency>
</dependencies>
日志依赖log4j
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target=System.out
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %m%n
log4j.appender.file=org.apache.log4j.FileAppender
log4j.appender.file.File=rebbitmq.log
log4j.appender.file.layout=org.apache.log4j.PatternLayout
log4j.appender.file.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %l %m%n
log4j.rootLogger=debug, stdout,file
package util;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class ConnectionUtil {
public static Connection getConnection() throws Exception{
//1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.在工厂对象中设置MQ的连接信息（ip,port,vhost,username,password）
factory.setHost("192.168.163.128");
factory.setPort(5672);
factory.setVirtualHost("/lagou");
factory.setUsername("wx");
factory.setPassword("123456");
//3.通过工厂获得与MQ的连接
Connection connection = factory.newConnection();
return connection;
}
public static void main(String[] args) throws Exception{
Connection connection = getConnection();
System.out.println("connection = " + connection);
connection.close();
}
}
2.3 RabbitMQ模式
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-lvs5Pd2c-1618660866156)(RabbitMQ.assets/RabbitMQ详解-1618451692977.jpg)]
5种消息模型，大体分为两类：
1和2属于点对点
3、4、5属于发布订阅模式（一对多）
点对点模式：P2P（point to point）模式包含三个角色：
消息队列（queue），发送者（sender），接收者（receiver）
每个消息发送到一个特定的队列中，接收者从中获得消息
队列中保留这些消息，直到他们被消费或超时
特点：
每个消息只有一个消费者，一旦消费，消息就不在队列中了
发送者和接收者之间没有依赖性，发送者发送完成，不管接收者是否运行，都不会影响消息发送到队列中
接收者成功接收消息之后需向对象应答成功
发布订阅模式：publish（Pub）/subscribe（Sub）
pub/sub模式包含三个角色：交换机（exchange），发布者（publisher），订阅者（subcriber）
多个发布者将消息发送交换机，系统将这些消息传递给多个订阅者
特点：
每个消息可以有多个订阅者
发布者和订阅者之间在时间上有依赖，对于某个交换机的订阅者，必须创建一个订阅后，才能消费发布者的消息
为了消费消息，订阅者必须保持运行状态
2.3.1 简单模式
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-u65S4aFv-1618660866158)(RabbitMQ.assets/RabbitMQ详解-1618453698752.jpg)]
生产者
public class Sender {
public static void main(String[] args) throws Exception {
String msg = "wx：Hello,RabbitMQ!";
// 1.获得连接
Connection connection = ConnectionUtil.getConnection();
// 2.在连接中创建通道（信道）
Channel channel = connection.createChannel();
// 3.创建消息队列(1,2,3,4,5)
/*
参数1:队列的名称
参数2:队列中的数据是否持久化
参数3:是否排外（是否支持扩展，当前队列只能自己用，不能给别人用）
参数4:是否自动删除（当队列的连接数为0时，队列会销毁，不管队列是否还存保存数据）
参数5:队列参数（没有参数为null）
*/
channel.queueDeclare("queue1",false,false,false,null);
// 4.向指定的队列发送消息(1,2,3,4)
/*
参数1:交换机名称，当前是简单模式，也就是P2P模式，没有交换机，所以名称为""
参数2:目标队列的名称
参数3:设置消息的属性（没有属性则为null）
参数4:消息的内容(只接收字节数组)
*/
channel.basicPublish("","queue1",null,msg.getBytes());
System.out.println("发送：" + msg);
// 5.释放资源
channel.close();
connection.close();
}
}
消费者
public class Recer {
public static void main(String[] args) throws Exception {
// 1.获得连接
Connection connection = ConnectionUtil.getConnection();
// 2.获得通道（信道）
Channel channel = connection.createChannel();
// 3.从信道中获得消息
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override //交付处理（收件人信息，包裹上的快递标签，协议的配置，消息）
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
// body就是从队列中获取的消息
String s = new String(body);
System.out.println("接收 = " + s);
}
};
// 4.监听队列 true:自动消息确认
channel.basicConsume("queue1", true,consumer);
}
}
消息确认机制ACK
消息一旦被消费，消息就会立刻从队列中移除
如果消费者接收消息后，还没执行操作就抛异常宕机导致消费失败，但是RabbitMQ无从得知，这样消息就丢失了
RabbitMQ有一个ACK机制，当消费者获取消息后，会向RabbitMQ发送回执ACK，告知消息已经被接收
回执ACK分为两种情况：
自动ACK：消息接收后，消费者立刻自动发送ACK（快递放在快递柜）
手动ACK：消息接收后，不会发送ACK，需要手动调用（快递必须本人签收）
如果消息不太重要，丢失也没有影响，自动ACK会比较方便
如果消息非常重要，最好消费完成手动ACK，如果自动ACK消费后，RabbitMQ就会把消息从队列中删除，如果此时消费者抛异常宕机，那么消息就永久丢失了
// false：手动消息确认
channel.basicConsume("queue1", false, consumer);
实例：
public class RecerByACK {
public static void main(String[] args) throws Exception {
// 1.获得连接
Connection connection = ConnectionUtil.getConnection();
// 2.获得通道（信道）
final Channel channel = connection.createChannel();
// 3.从信道中获得消息
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override //交付处理（收件人信息，包裹上的快递标签，协议的配置，消息）
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
// body就是从队列中获取的消息
String s = new String(body);
System.out.println("接收 = " + s);
// 手动确认（收件人信息，是否同时确认多个消息）
channel.basicAck(envelope.getDeliveryTag(),false);
}
};
// 4.监听队列，false:手动消息确认
channel.basicConsume("queue1", false,consumer);
}
}
2.3.2 工作队列模式
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-YOR9Kv0b-1618660866158)(RabbitMQ.assets/RabbitMQ详解-1618453725057.jpg)]
当我们运行许多消费者程序时，消息队列中的任务会被众多消费者共
享，但其中某一个消息只会被一个消费者获取
生产者
public class MessageSender {
public static void main(String[] args) throws Exception{
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 声明队列（此处为生产者，创建队列）注明出餐口位置，通知大家来排队
channel.queueDeclare("test_work_queue",false,false,false,null);
for(int i = 1;i<=100;i++) {
String msg = "羊肉串 --> "+i;
channel.basicPublish("", "test_work_queue", null, msg.getBytes());
System.out.println("师傅烤好：" + msg);
}
channel.close();
connection.close();
}
}
1号消费者
public class MessageReceiver1 {
static int i = 1; // 记录执行次数
public static void main(String[] args) throws IOException, TimeoutException{
Connection connection = ConnectionUtil.getConnection();
final Channel channel = connection.createChannel();
// 声明队列（此处为消费者，不是声明创建队列，而是获取，二者代码相同）出餐口排队
channel.queueDeclare("test_work_queue",false,false,false,null);
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = new String(body);
System.out.println("【顾客1】吃掉 " + msg+" ! 共吃【"+i++ +"】串");
// 撸一会，有延迟
try {
Thread.sleep(200);
}catch (InterruptedException e){
}
channel.basicAck(envelope.getDeliveryTag(),false);
}
};
channel.basicConsume("test_work_queue", false, consumer);
}
}
2号消费者
public class MessageReceiver2 {
static int i = 1; // 记录执行次数
public static void main(String[] args) throws IOException, TimeoutException{
Connection connection = ConnectionUtil.getConnection();
final Channel channel = connection.createChannel();
// 声明队列（此处为消费者，不是声明创建队列，而且获取，二者代码相同）出餐口排队
channel.queueDeclare("test_work_queue",false,false,false,null);
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = new String(body);
System.out.println("【顾客2】吃掉 " + msg+" ! 共吃【"+i++ +"】串");
// 撸一会，有延迟
try {
Thread.sleep(200);
}catch (InterruptedException e){
}
channel.basicAck(envelope.getDeliveryTag(),false);
}
};
channel.basicConsume("test_work_queue", false, consumer);
}
}
先运行1、2号消费者，在运行生产者
两个消费者的消费速度（线程休眠时间）不一致，但消费数量相同，各消费50个。
解决这一问题
// 声明队列（此处为消费者，不是声明创建队列，而且获取，二者代码相同）出餐口排队
channel.queueDeclare("test_work_queue",false,false,false,null);
在1、2号消费者中上述声明队列代码之后添加
// 可以理解为：快递一个一个送，送完一个再送下一个，速度快的送件就多
channel.basicQos(1);
2.3.3 发布订阅模式
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-uP6DNuWa-1618660866159)(RabbitMQ.assets/RabbitMQ详解-1618455117555.jpg)]
P生产者发送信息给X路由，X将信息转发给绑定X的队列
X队列将信息通过信道发送给消费者，从而进行消费
必须先创建路由
路由在生产者程序中创建
因为路由没有存储消息的能力，当生产者将信息发送给路由后，消费者还没有运行，所以没有队列，路由并不知道将信息发送给谁
生产者
public class Sender {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 声明路由(路由名，路由类型)
// fanout：不处理路由键（只需要将队列绑定到路由上，发送到路由的消息都会被转发到与该路由绑定的所有队列上）
channel.exchangeDeclare("test_exchange_fanout", "fanout");
String msg = "hello，大家好！";
channel.basicPublish("test_exchange_fanout", "", null, msg.getBytes());
System.out.println("生产者：" + msg);
channel.close();
connection.close();
}
}
1号消费者
public class Recer1 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 声明队列
channel.queueDeclare("test_exchange_fanout_queue_1",false,false,false,null);
// 绑定路由（关注）
/*
参数1：队列名
参数2：交换器名称
参数3：路由key（暂时无用，""即可）
*/
channel.queueBind("test_exchange_fanout_queue_1",
"test_exchange_fanout", "");
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String s = new String(body);
System.out.println("【消费者1】 = " + s);
}
};
// 4.监听队列 true:自动消息确认
channel.basicConsume("test_exchange_fanout_queue_1", true,consumer);
}
}
2号消费者同1号代码一致，但需要更改队列名称为2.
2.3.4 路由模式
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4CVJrXbk-1618660866159)(RabbitMQ.assets/RabbitMQ详解-1618456013569.jpg)]
路由会根据类型进行定向分发消息给不同的队列，如图所示
生产者
public class Sender {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 声明路由(路由名，路由类型)
// direct：根据路由键进行定向分发消息
channel.exchangeDeclare("test_exchange_direct", "direct");
String msg = "用户注册，【userid=S101】";
channel.basicPublish("test_exchange_direct", "insert", null,msg.getBytes());
System.out.println("[用户系统]：" + msg);
channel.close();
connection.close();
}
}
1号消费者
public class Recer1 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 声明队列
channel.queueDeclare("test_exchange_direct_queue_1",false,false,false,null);
// 绑定路由（如果路由键的类型是 添加，删除，修改 的话，绑定到这个队列1上）
channel.queueBind("test_exchange_direct_queue_1",
"test_exchange_direct", "insert");
channel.queueBind("test_exchange_direct_queue_1",
"test_exchange_direct", "update");
channel.queueBind("test_exchange_direct_queue_1",
"test_exchange_direct", "delete");
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String s = new String(body);
System.out.println("【消费者1】 = " + s);
}
};
// 4.监听队列 true:自动消息确认
channel.basicConsume("test_exchange_direct_queue_1", true,consumer);
}
}
2号消费者
public class Recer2 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 声明队列
channel.queueDeclare("test_exchange_direct_queue_2",false,false,false,null);
// 绑定路由（如果路由键的类型是 查询 的话，绑定到这个队列2上）
channel.queueBind("test_exchange_direct_queue_2",
"test_exchange_direct", "select");
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String s = new String(body);
System.out.println("【消费者2】 = " + s);
}
};
// 4.监听队列 true:自动消息确认
channel.basicConsume("test_exchange_direct_queue_2", true,consumer);
}
}
- 记住运行程序的顺序，先运行一次sender（创建路由器）
- 有了路由器之后，在创建两个Recer1和Recer2，进行队列绑定
- 再次运行sender，发出消息
2.3.5 通配符模式
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ZHH6sTdY-1618660866160)(RabbitMQ.assets/RabbitMQ详解-1618456698593.jpg)]
和路由模式基本一样，区别在于，路由键支持模糊查询
*：只能匹配一个词（正好一个词，多一个不行，少一个也不行）
#：匹配0个或更多个词
生产者
public class Sender {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 声明路由(路由名，路由类型)
// topic：模糊匹配的定向分发
channel.exchangeDeclare("test_exchange_topic", "topic");
String msg = "商品降价";
channel.basicPublish("test_exchange_topic", "product.price", null,
msg.getBytes());
System.out.println("[用户系统]：" + msg);
channel.close();
connection.close();
}
}
1号消费者
public class Recer1 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 声明队列
channel.queueDeclare("test_exchange_topic_queue_1",false,false,false,null);
// 绑定路由（绑定 用户相关 的消息）
channel.queueBind("test_exchange_topic_queue_1", "test_exchange_topic","user.#");
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String s = new String(body);
System.out.println("【消费者1】 = " + s);
}
};
// 4.监听队列 true:自动消息确认
channel.basicConsume("test_exchange_topic_queue_1", true,consumer);
}
}
2号消费者
public class Recer2 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 声明队列
channel.queueDeclare("test_exchange_topic_queue_2",false,false,false,null);
// 绑定路由（绑定 商品和订单相关 的消息）
channel.queueBind("test_exchange_topic_queue_2", "test_exchange_topic","product.#");
channel.queueBind("test_exchange_topic_queue_2", "test_exchange_topic","order.#");
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String s = new String(body);
System.out.println("【消费者2】 = " + s);
}
};
// 4.监听队列 true:自动消息确认
channel.basicConsume("test_exchange_topic_queue_2", true,consumer);
}
}
2.4 持久化
想要将消息持久化，那么 路由和队列都要持久化 才可以
生产者
public class Sender {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 声明路由(路由名，路由类型，持久化)
channel.exchangeDeclare("test_exchange_topic", "topic",true);
String msg = "商品降价";
// 发送消息(第三个参数作用是让消息持久化)
channel.basicPublish("test_exchange_topic", "product.price",
MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
System.out.println("[用户系统]：" + msg);
channel.close();
connection.close();
}
}
消费者
public class Recer1 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 声明队列( 第二个参数为true：支持持久化)
channel.queueDeclare("test_exchange_topic_queue_1",true,false,false,null);
channel.queueBind("test_exchange_topic_queue_1", "test_exchange_topic",
"user.#");
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String s = new String(body);
System.out.println("【消费者1】 = " + s);
}
};
channel.basicConsume("test_exchange_topic_queue_1", true,consumer);
}
}
2.5 Spring整合RabbitMQ
五种消息模型，在企业中应用最广泛的就是最后一种：定向匹配topic
生产端工程
pom.xml
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>2.0.1.RELEASE</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.25</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.9</version>
</dependency>
spring-rabbitmq-producer.xml
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
<!-- 1.配置连接 -->
<rabbit:connection-factory
id="connectionFactory"
host="192.168.163.128"
port="5672"
username="wx"
password="123456"
virtual-host="/lagou"
/>
<!-- 2.配置队列 -->
<rabbit:queue name="test_spring_queue_1"/>
<!-- 3.配置rabbitAdmin：主要用于在Java代码中对理队和队列进行管理，用于创建、绑定、删
除队列与交换机，发送消息等 -->
<rabbit:admin connection-factory="connectionFactory"/>
<!-- 4.配置topic类型exchange；队列绑定到交换机 -->
<rabbit:topic-exchange name="spring_topic_exchange">
<rabbit:bindings>
<rabbit:binding queue="test_spring_queue_1" pattern="msg.#"/>
</rabbit:bindings>
</rabbit:topic-exchange>
<!-- 5. 配置消息对象json转换类 -->
<bean id="jsonMessageConverter"
class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter"/
>
<!-- 6. 配置RabbitTemplate（消息生产者） -->
<rabbit:template id="rabbitTemplate"
connection-factory="connectionFactory"
exchange="spring_topic_exchange"
message-converter="jsonMessageConverter"
/>
</beans>
发送消息
public class Sender {
public static void main(String[] args) {
// 1.创建spring容器
ClassPathXmlApplicationContext context = new
ClassPathXmlApplicationContext("spring/spring-rabbitmq-producer.xml");
// 2.从容器中获取对象
RabbitTemplate template = context.getBean(RabbitTemplate.class);
// 3.发送消息
Map<String, String> map = new HashMap();
map.put("name", "大佬孙");
map.put("email", "[email protected]");
template.convertAndSend("msg.user", map);
context.close();
}
}
消费端工程
pox.xml同生产端一致
spring-rabbitmq-consumer.xml
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd">
<!-- 1. 配置连接 -->
<rabbit:connection-factory
id="connectionFactory"
host="192.168.163.128"
port="5672"
username="wx"
password="123456"
virtual-host="/lagou"
/>
<!-- 2. 配置队列 -->
<rabbit:queue name="test_spring_queue_1"/>
<!-- 3.配置rabbitAdmin -->
<rabbit:admin connection-factory="connectionFactory"/>
<!-- 4.springIOC注解扫描包-->
<context:component-scan base-package="listener"/>
<!-- 5.配置监听 -->
<rabbit:listener-container connection-factory="connectionFactory">
<rabbit:listener ref="consumerListener" queuenames="test_spring_queue_1" />
</rabbit:listener-container>
</beans>
MessageListener接口用于spring容器接收到消息后处理消息
如果需要使用自己定义的类型 来实现 处理消息时，必须实现该接口，并重写onMessage()方法
当spring容器接收消息后，会自动交由onMessage进行处理
消费者
@Component
public class ConsumerListener implements MessageListener {
// jackson提供序列化和反序列中使用最多的类，用来转换json的
private static final ObjectMapper MAPPER = new ObjectMapper();
public void onMessage(Message message) {
try {
// 将message对象转换成json
JsonNode jsonNode = MAPPER.readTree(message.getBody());
String name = jsonNode.get("name").asText();
String email = jsonNode.get("email").asText();
System.out.println("从队列中获取：【"+name+"的邮箱是:"+email+"】");
} catch (Exception e){
e.printStackTrace();
}
}
}
启动项目
public class TestRunner {
public static void main(String[] args) throws Exception {
// 获得容器
ClassPathXmlApplicationContext context = new
ClassPathXmlApplicationContext("spring/spring-rabbitmq-consumer.xml");
// 让程序一直运行，别终止
System.in.read();
}
}
2.6 消息成功确认机制
在发送的消息出现错误是，我们无法保证消息能够成功投递。理想情况下应为消息发送后出错则将出错前的消息返回，并不发送，此时需要考虑用事务机制。
2.6.1 事务机制
AMQP协议提供的一种保证消息成功投递的方式，通过信道开启 transactional 模式
并利用信道 的三个方法来实现以事务方式 发送消息，若发送失败，通过异常处理回滚事务，确保消息成功投递
channel.txSelect()： 开启事务
channel.txCommit() ：提交事务
channel.txRollback() ：回滚事务
生产者
public class Sender {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare("test_transaction", "topic");
channel.txSelect(); // 开启事务
try {
channel.basicPublish("test_transaction", "product.price", null, "商品
降价1".getBytes());
// System.out.println(1/0); // 模拟异常！
channel.basicPublish("test_transaction", "product.price", null, "商品
降价2".getBytes());
System.out.println("消息全部发出！");
channel.txCommit(); // 事务提交
}catch (Exception e){
System.out.println("由于系统异常，消息全部撤回！");
channel.txRollback(); // 事务回滚
e.printStackTrace();
}finally {
channel.close();
connection.close();
}
}
}
消费者
public class Recer {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
channel.queueDeclare("test_transaction_queue",false,false,false,null);
channel.queueBind("test_transaction_queue", "test_transaction",
"product.#");
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String s = new String(body);
System.out.println("【消费者】 = " + s);
}
};
channel.basicConsume("test_transaction_queue", true,consumer);
}
}
2.6.2 Confirm发布确认机制
RabbitMQ为了保证消息的成功投递，采用通过AMQP协议层面为我们提供事务机制的方案，但是采用事务会大大降低消息的吞吐量
事务效率为什么会这么低呢？试想一下：10条消息，前9条成功，如果第10条失败，那么9条消息要全部撤销回滚。太太太浪费
而confirm模式则采用补发第10条的措施来完成10条消息的送达
spring-rabbitmq-producer.xml
在前文中的xml文件中进行修改，添加生产者确认机制
<!--1.配置连接，启动生产者确认机制: publisher-confirms="true"-->
<rabbit:connection-factory id="connectionFactory"
host="192.168.163.128"
port="5672"
username="wx"
password="123456"
virtual-host="/lagou"
publisher-confirms="true"
/>
<!--6.配置rabbitmq的模版,添加确认回调处理类:confirmcallback="
msgSendConfirmCallback"-->
<rabbit:template id="rabbitTemplate"
connection-factory="connectionFactory"
exchange="spring_topic_exchange"
message-converter="jsonMessageConverter"
confirm-callback="msgSendConfirmCallback"/>
<!--7.确认机制处理类-->
<bean id="msgSendConfirmCallback" class="confirm.MsgSendConfirmCallback"/>
消息确认处理类
判断消息是否能够成功投递的类
/**
* @BelongsProject: spring-rabbitmq-producer
* @Author: GuoAn.Sun
* @Description: 确认机制
*/
@Component
public class MsgSendConfirmCallback implements
RabbitTemplate.ConfirmCallback {
public void confirm(CorrelationData correlationData, boolean b, String s) {
if (b){
System.out.println("消息确认成功！！");
} else {
System.out.println("消息确认失败。。。");
// 如果本条消息一定要发送到队列中，例如下订单消息，我们可以采用消息补发
// 采用递归（固定次数，不可无限）或 redis+定时任务
}
}
}
log4j.properties
能够打印程序运行的日志（可查询错误信息），但是否存在不影响消息传递的功能实现。
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target=System.out
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %m%n
log4j.appender.file=org.apache.log4j.FileAppender
log4j.appender.file.File=rabbitmq.log
log4j.appender.file.layout=org.apache.log4j.PatternLayout
log4j.appender.file.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %l%m%n
log4j.rootLogger=debug, stdout,file
发送消息
public class Sender {
public static void main(String[] args) {
ClassPathXmlApplicationContext context = new
ClassPathXmlApplicationContext("spring/spring-rabbitmq-producer.xml");
RabbitTemplate rabbitTemplate =
context.getBean(RabbitTemplate.class);
Map<String,String> map = new HashMap<String, String>();
map.put("name", "吕布");
map.put("email", "[email protected]");
// 第一个参数是路由名称，
// 不写，则使用spring容器中创建的路由
// 乱写一个，因为路由名错误导致报错，则进入消息确认失败流程
rabbitTemplate.convertAndSend("x","msg.user",map);
System.out.println("ok");
context.close();
}
}
2.7 消费端限流
当数据量特别大的时候，我们对生产端限流肯定是不科学的，因为有时候并发量就是特别大，有时候并发量又特别少，这是用户的行为，我们是无法约束的
所以我们应该对消费端限流，用于保持消费端的稳定
RabbitMQ 提供了一种 Qos （Quality of Service，服务质量）服务质量保证功能
即在非自动确认消息的前提下，如果一定数目的消息未被确认前，不再进行消费新的消息
消费者进行限流处理
<!--5.配置监听-->
<!-- prefetch="3" 一次性消费的消息数量。会告诉 RabbitMQ 不要同时给一个消费者推送多于
N 个消息，一旦有 N 个消息还没有ack，则该 consumer 将阻塞，直到消息被ack-->
<!-- acknowledge-mode: manual 手动确认-->
<rabbit:listener-container connection-factory="connectionFactory"
prefetch="3" acknowledge="manual">
<rabbit:listener ref="consumerListener" queuenames="test_spring_queue_1" />
</rabbit:listener-container>
// AbstractAdaptableMessageListener用于在spring容器接收到消息后用于处理消息的抽象基类
@Component
public class ConsumerListener extends AbstractAdaptableMessageListener {
// jackson提供序列化和反序列中使用最多的类，用来转换json的
private static final ObjectMapper MAPPER = new ObjectMapper();
public void onMessage(Message message, Channel channel) throws Exception
{
try {
// String str = new String(message.getBody());
// 将message对象转换成json
JsonNode jsonNode = MAPPER.readTree(message.getBody());
String name = jsonNode.get("name").asText();
String email = jsonNode.get("email").asText();
System.out.println("从队列中获取：【"+name+"的邮箱是:"+email+"】");
long deliveryTag =
message.getMessageProperties().getDeliveryTag();
//确认收到(参数1,参数2)
/*
参数1：RabbitMQ 向该 Channel 投递的这条消息的唯一标识 ID，是一个单调递增的正整数，delivery_tag 的范围仅限于 Channel
参数2：为了减少网络流量，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
*/
channel.basicAck(deliveryTag , true);
Thread.sleep(3000);
System.out.println("休息三秒然后再接收消息");
} catch (Exception e){
e.printStackTrace();
}
}
}
2.8 过期时间TTL
Time To Live：生存时间、还能活多久，单位毫秒
在这个周期内，消息可以被消费者正常消费，超过这个时间，则自动删除（其实是被称为deadmessage并投入到死信队列，无法消费该消息）
RabbitMQ可以对消息和队列设置TTL
通过队列设置，队列中所有消息都有相同的过期时间
对消息单独设置，每条消息的TTL可以不同（更颗粒化）
2.8.1 设置队列TTL
spring-rabbitmq-producer.xml
5s后自动删除
<!--2.重新配置一个队列，同时，对队列中的消息设置过期时间-->
<rabbit:queue name="test_spring_queue_ttl" auto-declare="true">
<rabbit:queue-arguments>
<entry key="x-message-ttl" value-type="long" value="5000"></entry>
</rabbit:queue-arguments>
</rabbit:queue>
2.8.2 设置消息TTL
设置某条消息的ttl，只需要在创建发送消息时指定即可
<!--2.配置队列-->
<rabbit:queue name="test_spring_queue_ttl_2">
/**
* @BelongsProject: spring-rabbitmq-producer
* @Author: GuoAn.Sun
* @Description: 生产者
*/
public class Sender2 {
public static void main(String[] args) {
ClassPathXmlApplicationContext context = new
ClassPathXmlApplicationContext("spring/spring-rabbitmq-producer.xml");
RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);
// 创建消息配置对象
MessageProperties messageProperties = new MessageProperties();
// 设置消息过期时间
messageProperties.setExpiration("6000");
// 创建消息
Message message = new Message("6秒后自动删除".getBytes(),
messageProperties);
// 发送消息
rabbitTemplate.convertAndSend("msg.user", message);
System.out.println("消息已发出...");
context.close();
}
}
如果同时设置了queue和message的TTL值，则二者中较小的才会起作用
2.9 死信队列
DLX（Dead Letter Exchanges）死信交换机/死信邮箱，当消息在队列中由于某些原因没有被及时消费而变成死信（dead message）后，这些消息就会被分发到DLX交换机中，而绑定DLX交换机的队列，称之为：“死信队列”
消息没有被及时消费的原因：
消息被拒绝（basic.reject/ basic.nack）并且不再重新投递 requeue=false
消息超时未消费达到最大队列长度
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-xSZkAxiL-1618660866161)(RabbitMQ.assets/RabbitMQ详解-1618546937827.jpg)]
spring-rabbitmq-producer-dlx.xml
<!--1.配置连接-->
<rabbit:connection-factory id="connectionFactory"
host="192.168.163.128"
port="5672"
username="wx"
password="123456"
virtual-host="/lagou"/>
<!--3.配置rabbitAdmin:主要用于在java代码中对队列的管理，用来创建，绑定，删除队列与交
换机，发送消息等-->
<rabbit:admin connection-factory="connectionFactory"/>
<!--6.配置rabbitmq的模版-->
<rabbit:template id="rabbitTemplate"
connection-factory="connectionFactory"
exchange="my_exchange"/>
<!--死信队列-->
<rabbit:queue name="dlx_queue"/>
<!--定向死信交换机-->
<rabbit:direct-exchange name="dlx_exchange" >
<rabbit:bindings>
<rabbit:binding key="dlx_ttl" queue="dlx_queue"></rabbit:binding>
<rabbit:binding key="dlx_max" queue="dlx_queue"></rabbit:binding>
</rabbit:bindings>
</rabbit:direct-exchange>
<!--测试超时的队列-->
<rabbit:queue name="test_ttl_queue">
<rabbit:queue-arguments>
<!--队列ttl为6秒-->
<entry key="x-message-ttl" value-type="long" value="6000"/>
<!--超时 消息 投递给 死信交换机-->
<entry key="x-dead-letter-exchange" value="dlx_exchange"/>
</rabbit:queue-arguments>
</rabbit:queue>
<!--测试超长度的队列-->
<rabbit:queue name="test_max_queue">
<rabbit:queue-arguments>
<!--队列ttl为6秒-->
<entry key="x-max-length" value-type="long" value="2"/>
<!--超时 消息 投递给 死信交换机-->
<entry key="x-dead-letter-exchange" value="dlx_exchange"/>
</rabbit:queue-arguments>
</rabbit:queue>
<!--定向测试交换机-->
<rabbit:direct-exchange name="my_exchange" >
<rabbit:bindings>
<rabbit:binding key="dlx_ttl" queue="test_ttl_queue">
</rabbit:binding>
<rabbit:binding key="dlx_max" queue="test_max_queue">
</rabbit:binding>
</rabbit:bindings>
</rabbit:direct-exchange>
发送消息测试
public class SenderDLX {
public static void main(String[] args) {
ClassPathXmlApplicationContext context = new
ClassPathXmlApplicationContext("spring/spring-rabbitmq-producer-dlx.xml");
RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);
// rabbitTemplate.convertAndSend("dlx_ttl", "测试超时".getBytes());
rabbitTemplate.convertAndSend("dlx_max", "测试长度1".getBytes());
rabbitTemplate.convertAndSend("dlx_max", "测试长度2".getBytes());
rabbitTemplate.convertAndSend("dlx_max", "测试长度3".getBytes());
System.out.println("消息已发出...");
context.close();
}
}
2.10 延迟队列
延迟队列：TTL + 死信队列的合体
死信队列只是一种特殊的队列，里面的消息仍然可以消费
spring-rabbitmq-consumer.xml
<!-- 监听死信队列 -->
<rabbit:listener-container connection-factory="connectionFactory" prefetch="3"
acknowledge="manual">
<rabbit:listener ref="consumerListener" queue-names="dlx_queue" />
</rabbit:listener-container>
3 RabbitMQ集群
rabbitmq有3种模式，但集群模式是2种
单一模式：即单机情况不做集群，就单独运行一个rabbitmq而已（并不推荐使用）
普通模式：默认模式，以两个节点（A、B）为例来进行说明
当消息进入A节点的Queue后，consumer从B节点消费时，RabbitMQ会在A和B之间创建临时通道进行消息传输，把A中的消息实体取出并经过通过交给B发送给consumer
当A故障后，B就无法取到A节点中未消费的消息实体
如果做了消息持久化，那么得等A节点恢复，然后才可被消费
如果没有持久化的话，就会产生消息丢失的现象
镜像模式：非常经典的 mirror 镜像模式，保证 100% 数据不丢失。
高可靠性解决方案，主要就是实现数据的同步，一般来讲是 2 - 3 个节点实现数据同步
对于 100% 数据可靠性解决方案，一般是采用 3 个节点。
在实际工作中也是用得最多的，并且实现非常的简单，一般互联网大厂都会构建这种镜像集群模式
3.1 集群搭建
前置条件：准备两台linux，并安装好rabbitmq
集群步骤：
-
修改 /etc/hosts 映射文件
1号服务器：
127.0.0.1 A localhost localhost.localdomain localhost4 localhost4.localdomain4 ::1 A localhost localhost.localdomain localhost6 localhost6.localdomain6 192.168.204.141 A 192.168.204.142 B2号服务器：
127.0.0.1 B localhost localhost.localdomain localhost4 localhost4.localdomain4 ::1 B localhost localhost.localdomain localhost6 localhost6.localdomain6 192.168.204.141 A 192.168.204.142 B -
相互通信，cookie必须保持一致，同步 rabbitmq的cookie 文件：跨服务器拷贝 .erlang.cookie（隐藏文件，使用 ls -all 显示）
[[email protected] opt]# scp /var/lib/rabbitmq/.erlang.cookie 192.168.204.142:/var/lib/rabbitmq修改cookie文件，要重启服务器，reboot
-
停止防火墙，启动rabbitmq服务
[[email protected] ~]# systemctl stop firewalld [[email protected] ~]# systemctl start rabbitmq-server -
加入集群节点
[[email protected] ~]# rabbitmqctl stop_app [[email protected] ~]# rabbitmqctl join_cluster [email protected] [[email protected] ~]# rabbitmqctl start_app -
查看节点状态
[[email protected] ~]# rabbitmqctl cluster_status -
查看管理端
搭建集群结构之后，之前创建的交换机、队列、用户都属于单一结构，在新的集群环境中是不能用的
所以在新的集群中重新手动添加用户即可（任意节点添加，所有节点共享）
[[email protected] ~]# rabbitmqctl add_user laosun 123123 [[email protected] ~]# rabbitmqctl set_user_tags laosun administrator [[email protected] ~]# rabbitmqctl set_permissions -p "/" laosun ".*" ".*" ".*"注意：当节点脱离集群还原成单一结构后，交换机，队列和用户等数据 都会重新回来
3.2 镜像模式
将所有队列设置为镜像队列，即队列会被复制到各个节点，各个节点状态一致
语法：set_policy {name} {pattern} {definition}
name：策略名，可自定义
pattern：队列的匹配模式（正则表达式）
“^” 可以使用正则表达式，比如"^queue_" 表示对队列名称以“queue_”开头的所有队列进行镜像，而"^"表示匹配所有的队列
definition：镜像定义，包括三个部分ha-mode, ha-params, ha-sync-mode
ha-mode：（High Available，高可用）模式，指明镜像队列的模式，有效值为all/exactly/nodes，当前策略模式为 all，即复制到所有节点，包含新增节点
all：表示在集群中所有的节点上进行镜像
exactly：表示在指定个数的节点上进行镜像，节点的个数由ha-params指定
nodes：表示在指定的节点上进行镜像，节点名称通过ha-params指定
ha-params：ha-mode模式需要用到的参数
ha-sync-mode：进行队列中消息的同步方式，有效值为automatic和manual
[[email protected] ~]# rabbitmqctl set_policy xall "^" '{"ha-mode":"all"}'
3.3 HAProxy实现镜像队列的负载均衡
3.3.1 HAProxy
HA（High Available，高可用），Proxy（代理）
HAProxy是一款提供高可用性，负载均衡，并且基于TCP和HTTP应用的代理软件
HAProxy可以支持数以万计的并发连接
HAProxy可以简单又安全的整合进架构中，同时还保护web服务器不被暴露到网络上
3.3.2 HAProxy与Nginx
OSI：（Open System Interconnection：开放式系统互联 是把网络通信的工作分为7层,分别是物理层,数据链路层,网络层,传输层,会话层,表示层和应用层）
Nginx的优点：
工作在OSI第7层，可以针对http应用做一些分流的策略
Nginx对网络的依赖非常小，理论上能ping通就就能进行负载功能，屹立至今的绝对优势
Nginx安装和配置比较简单，测试起来比较方便；
Nginx不仅仅是一款优秀的负载均衡器/反向代理软件，它同时也是功能强大的Web应用服务器
HAProxy的优点：
工作在网络4层和7层，支持TCP与Http协议，
它仅仅就只是一款负载均衡软件；单纯从效率上来讲HAProxy更会比Nginx有更出色的负载均
衡速度，在并发处理上也是优于Nginx的
支持8种负载均衡策略 ，支持心跳检测
性能上HA胜，功能性和便利性上Nginx胜
对于Http协议，Haproxy处理效率比Nginx高。所以，没有特殊要求的时候或者一般场景，建议使用Haproxy来做Http协议负载
但如果是Web应用，那么建议使用Nginx
``
注意：当节点脱离集群还原成单一结构后，交换机，队列和用户等数据 都会重新回来
3.2 镜像模式
将所有队列设置为镜像队列，即队列会被复制到各个节点，各个节点状态一致
语法：set_policy {name} {pattern} {definition}
name：策略名，可自定义
pattern：队列的匹配模式（正则表达式）
“^” 可以使用正则表达式，比如"^queue_" 表示对队列名称以“queue_”开头的所有队列进行镜像，而"^"表示匹配所有的队列
definition：镜像定义，包括三个部分ha-mode, ha-params, ha-sync-mode
ha-mode：（High Available，高可用）模式，指明镜像队列的模式，有效值为all/exactly/nodes，当前策略模式为 all，即复制到所有节点，包含新增节点
all：表示在集群中所有的节点上进行镜像
exactly：表示在指定个数的节点上进行镜像，节点的个数由ha-params指定
nodes：表示在指定的节点上进行镜像，节点名称通过ha-params指定
ha-params：ha-mode模式需要用到的参数
ha-sync-mode：进行队列中消息的同步方式，有效值为automatic和manual
[[email protected] ~]# rabbitmqctl set_policy xall "^" '{"ha-mode":"all"}'
3.3 HAProxy实现镜像队列的负载均衡
3.3.1 HAProxy
HA（High Available，高可用），Proxy（代理）
HAProxy是一款提供高可用性，负载均衡，并且基于TCP和HTTP应用的代理软件
HAProxy可以支持数以万计的并发连接
HAProxy可以简单又安全的整合进架构中，同时还保护web服务器不被暴露到网络上
3.3.2 HAProxy与Nginx
OSI：（Open System Interconnection：开放式系统互联 是把网络通信的工作分为7层,分别是物理层,数据链路层,网络层,传输层,会话层,表示层和应用层）
Nginx的优点：
工作在OSI第7层，可以针对http应用做一些分流的策略
Nginx对网络的依赖非常小，理论上能ping通就就能进行负载功能，屹立至今的绝对优势
Nginx安装和配置比较简单，测试起来比较方便；
Nginx不仅仅是一款优秀的负载均衡器/反向代理软件，它同时也是功能强大的Web应用服务器
HAProxy的优点：
工作在网络4层和7层，支持TCP与Http协议，
它仅仅就只是一款负载均衡软件；单纯从效率上来讲HAProxy更会比Nginx有更出色的负载均
衡速度，在并发处理上也是优于Nginx的
支持8种负载均衡策略 ，支持心跳检测
性能上HA胜，功能性和便利性上Nginx胜
对于Http协议，Haproxy处理效率比Nginx高。所以，没有特殊要求的时候或者一般场景，建议使用Haproxy来做Http协议负载
但如果是Web应用，那么建议使用Nginx