RabbitMQ学习

1、概述

用于进程通信的中间件。

优势：劣势：

1、应用解耦：提高了系统容错性和可维护性 1、系统依赖越多不能保证MQ的高可用

2、异步提速：提升用户体验和系统吞吐量 2、复杂度提高

3、削峰填谷：提高系统稳定性 3、一致性问题

2、rabbit mq安装

1. 安装依赖环境

在线安装依赖环境：

yum install build-essential openssl openssl-devel unixODBC unixODBC-devel make gcc gcc-c++ kernel-devel m4 ncurses-devel tk tc xz

2. 安装Erlang

上传

erlang-18.3-1.el7.centos.x86_64.rpm
socat-1.7.3.2-5.el7.lux.x86_64.rpm
rabbitmq-server-3.6.5-1.noarch.rpm

# 安装
rpm -ivh erlang-18.3-1.el7.centos.x86_64.rpm

如果出现如下错误

说明gblic 版本太低。我们可以查看当前机器的gblic 版本

strings /lib64/libc.so.6 | grep GLIBC

当前最高版本2.12，需要2.15.所以需要升级glibc

使用yum更新安装依赖

sudo yum install zlib-devel bzip2-devel openssl-devel ncurses-devel sqlite-devel readline-devel tk-devel gcc make -y

下载rpm包

wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepo-el6/epel-6-x86_64/glibc-2.17-55.fc20/glibc-utils-2.17-55.el6.x86_64.rpm &
wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepo-el6/epel-6-x86_64/glibc-2.17-55.fc20/glibc-static-2.17-55.el6.x86_64.rpm &
wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepo-el6/epel-6-x86_64/glibc-2.17-55.fc20/glibc-2.17-55.el6.x86_64.rpm &
wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepo-el6/epel-6-x86_64/glibc-2.17-55.fc20/glibc-common-2.17-55.el6.x86_64.rpm &
wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepo-el6/epel-6-x86_64/glibc-2.17-55.fc20/glibc-devel-2.17-55.el6.x86_64.rpm &
wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepo-el6/epel-6-x86_64/glibc-2.17-55.fc20/glibc-headers-2.17-55.el6.x86_64.rpm &
wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepo-el6/epel-6-x86_64/glibc-2.17-55.fc20/nscd-2.17-55.el6.x86_64.rpm &

安装rpm包

sudo rpm -Uvh *-2.17-55.el6.x86_64.rpm --force --nodeps

安装完毕后再查看glibc版本,发现glibc版本已经到2.17了
```
strings /lib64/libc.so.6 | grep GLIBC
```

3. 安装RabbitMQ

# 安装
rpm -ivh socat-1.7.3.2-5.el7.lux.x86_64.rpm# 安装
rpm -ivh rabbitmq-server-3.6.5-1.noarch.rpm

4. 开启管理界面及配置

# 开启管理界面
rabbitmq-plugins enable rabbitmq_management
# 修改默认配置信息
vim /usr/lib/rabbitmq/lib/rabbitmq_server-3.6.5/ebin/rabbit.app
# 比如修改密码、配置等等，例如：loopback_users 中的 <<"guest">>,只保留guest

5. 启动

service rabbitmq-server start # 启动服务  或者 systemctl start rabbitmq-server
service rabbitmq-server stop # 停止服务
service rabbitmq-server restart # 重启服务

设置配置文件

cd /usr/share/doc/rabbitmq-server-3.6.5/cp rabbitmq.config.example /etc/rabbitmq/rabbitmq.config

6. 配置虚拟主机及用户

6.1. 用户角色

RabbitMQ在安装好后，可以访问http://ip地址:15672 ；其自带了guest/guest的用户名和密码；如果需要创建自定义用户；那么也可以登录管理界面后，如下操作：

角色说明：

1、超级管理员(administrator)

可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。

2、监控者(monitoring)

可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)

3、策略制定者(policymaker)

可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。

4、普通管理者(management)

仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。

5、其他

无法登陆管理控制台，通常就是普通的生产者和消费者。

6.2. Virtual Hosts配置

像mysql拥有数据库的概念并且可以指定用户对库和表等操作的权限。RabbitMQ也有类似的权限管理；在RabbitMQ中可以虚拟消息服务器Virtual Host，每个Virtual Hosts相当于一个相对独立的RabbitMQ服务器，每个VirtualHost之间是相互隔离的。exchange、queue、message不能互通。相当于mysql的db。Virtual Name一般以/开头。

6.2.1. 创建Virtual Hosts

6.2.2. 设置Virtual Hosts权限

3、快速搭建（简单模式）

依赖：<dependencies><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.6.0</version></dependency></dependencies>

//                                    生产者try{//1、建立连接工厂ConnectionFactory factory=new ConnectionFactory();//2、设置参数factory.setHost("192.168.10.130");  //默认是localhostfactory.setPort(5672);  //默认也是5672factory.setVirtualHost("/admin"); //默认是/factory.setUsername("admin"); //默认是guestfactory.setPassword("123456"); //默认是guest//3、创建连接Connection connection=factory.newConnection();//4、创建chanelChannel channel=connection.createChannel();//5、创建队列queue/** String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments* 1、queue:队列名称* 2、durable：队列是否写入磁盘持久化* 3、exclusive：是否独占;connection关闭时是否关闭队列* 4、autodelete：自动删除* 5、arguments一些参数信息*** *///如果有名字叫hello_word就会创建没有则不会channel.queueDeclare("hello_word",true,false,false,null);//6、发送消息/** (String exchange, String routingKey, boolean mandatory, BasicProperties props, byte[] body* 1、exchange:交换机，简单模式下为""* 2、routingkey：路由名称,没有交换机就必须和队列名一致* 3、props：配置信息* 4、b0dy；字节信息** */String body="hello rabbitmq";channel.basicPublish("","hello_word",null,body.getBytes());//7、释放资源channel.close();connection.close();}catch (Exception e){e.printStackTrace();}

//                                   消费者try{//1、建立连接工厂ConnectionFactory factory=new ConnectionFactory();//2、设置参数factory.setHost("192.168.10.130");  //默认是localhostfactory.setPort(5672);  //默认也是5672factory.setVirtualHost("/admin"); //默认是/factory.setUsername("admin"); //默认是guestfactory.setPassword("123456"); //默认是guest//3、创建连接Connection connection=factory.newConnection();//4、创建chanelChannel channel=connection.createChannel();//5、创建队列queue/** String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments* 1、queue:队列名称* 2、durable：队列是否写入磁盘持久化* 3、exclusive：是否独占;connection关闭时是否关闭队列* 4、autodelete：自动删除* 5、arguments一些参数信息*** *///如果有名字叫hello_word就会创建没有则不会channel.queueDeclare("hello_word",true,false,false,null);//6、发送消息/** (String exchange, String routingKey, boolean mandatory, BasicProperties props, byte[] body* 1、exchange:交换机，简单模式下为""* 2、routingkey：路由名称,没有交换机就必须和队列名一致* 3、props：配置信息* 4、b0dy；字节信息** */
//            String body="hello rabbitmq";
//            channel.basicPublish("","hello_word",null,body.getBytes());/** (String queue, DeliverCallback deliverCallback, ConsumerShutdownSignalCallback shutdownSignalCallback**1、queue队列名称* 2、* */Consumer consumer=new DefaultConsumer(channel){//回调方法，当收到消息后会自动执行该方法/** consumertag ： 标识* envelop：获取一些信息，交换机，路由key* properties：配置信息* body : 真实数据* */@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(consumerTag+"     -"+envelope+"     -"+properties+"     -"+new String(body));}};channel.basicConsume("hello_word",true,consumer);}catch (Exception e){e.printStackTrace();}

4、工作模式

4.1、workqueue 工作队列

实现：同简单模式，只不过多了一个消费者，同一个消息消费者之间是竞争关系只能有一个能消费。

4.2、订阅模式

生产者：try{//1、建立连接工厂ConnectionFactory factory=new ConnectionFactory();//2、设置参数factory.setHost("192.168.10.130");  //默认是localhostfactory.setPort(5672);  //默认也是5672factory.setVirtualHost("/admin"); //默认是/factory.setUsername("admin"); //默认是guestfactory.setPassword("123456"); //默认是guest//3、创建连接Connection connection=factory.newConnection();//4、创建chanelChannel channel=connection.createChannel();//5、创建交换机/** String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)*1、exchange:交换机名称*2、type：交换机的类型* DIRECT("direct"):定向的方式FANOUT("fanout")：广播TOPIC("topic")：通配符HEADERS("headers"):参数匹配*3、durable：是否持久化* 4、autoDelete：自动删除* 5、internal：内部使用* 6、arguments：参数列表* */String ex1="test_fanout";channel.exchangeDeclare(ex1, BuiltinExchangeType.FANOUT,true,false,false,null);//6、创建队列String q1="test_fanout_queue1";String q2="test_fanout_queue2";channel.queueDeclare(q1,true,false,false,null);channel.queueDeclare(q2,true,false,false,null);//7、绑定队列和交换机/** String queue, String exchange, String routingKey* routingkey=路由键，fanout是*             * */channel.queueBind(q1,ex1,"");channel.queueBind(q2,ex1,"");//8、发送消息String body="日志信息：XXXX";channel.basicPublish(ex1,"",null,body.getBytes());//9、释放资源channel.close();connection.close();}catch (Exception e){e.printStackTrace();}

//消费者try{//1、建立连接工厂ConnectionFactory factory=new ConnectionFactory();//2、设置参数factory.setHost("192.168.10.130");  //默认是localhostfactory.setPort(5672);  //默认也是5672factory.setVirtualHost("/admin"); //默认是/factory.setUsername("admin"); //默认是guestfactory.setPassword("123456"); //默认是guest//3、创建连接Connection connection=factory.newConnection();//4、创建chanelChannel channel=connection.createChannel();String q1="test_fanout_queue1";//5、创建队列queue/** String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments* 1、queue:队列名称* 2、durable：队列是否写入磁盘持久化* 3、exclusive：是否独占;connection关闭时是否关闭队列* 4、autodelete：自动删除* 5、arguments一些参数信息*** *///如果有名字叫hello_word就会创建没有则不会
//            channel.queueDeclare("hello_word",true,false,false,null);//6、发送消息/** (String exchange, String routingKey, boolean mandatory, BasicProperties props, byte[] body* 1、exchange:交换机，简单模式下为""* 2、routingkey：路由名称,没有交换机就必须和队列名一致* 3、props：配置信息* 4、b0dy；字节信息** */
//            String body="hello rabbitmq";
//            channel.basicPublish("","hello_word",null,body.getBytes());/** (String queue, DeliverCallback deliverCallback, ConsumerShutdownSignalCallback shutdownSignalCallback**1、queue队列名称* 2、* */Consumer consumer=new DefaultConsumer(channel){//回调方法，当收到消息后会自动执行该方法/** consumertag ： 标识* envelop：获取一些信息，交换机，路由key* properties：配置信息* body : 真实数据* */@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(new String(body));}};channel.basicConsume("test_fanout_queue1",true,consumer);}catch (Exception e){e.printStackTrace();}

4.3、路由模式

生产者：try{//1、建立连接工厂ConnectionFactory factory=new ConnectionFactory();//2、设置参数factory.setHost("192.168.10.130");  //默认是localhostfactory.setPort(5672);  //默认也是5672factory.setVirtualHost("/admin"); //默认是/factory.setUsername("admin"); //默认是guestfactory.setPassword("123456"); //默认是guest//3、创建连接Connection connection=factory.newConnection();//4、创建chanelChannel channel=connection.createChannel();//5、创建交换机/** String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)*1、exchange:交换机名称*2、type：交换机的类型* DIRECT("direct"):定向的方式FANOUT("fanout")：广播TOPIC("topic")：通配符HEADERS("headers"):参数匹配*3、durable：是否持久化* 4、autoDelete：自动删除* 5、internal：内部使用* 6、arguments：参数列表* */String ex1="test_diret";channel.exchangeDeclare(ex1, BuiltinExchangeType.DIRECT,true,false,false,null);//6、创建队列String q1="test_direct_queue1";String q2="test_direct_queue2";channel.queueDeclare(q1,true,false,false,null);channel.queueDeclare(q2,true,false,false,null);//7、绑定队列和交换机/** String queue, String exchange, String routingKey* routingkey=路由键，fanout是*             * */channel.queueBind(q1,ex1,"error");channel.queueBind(q2,ex1,"error");channel.queueBind(q2,ex1,"info");channel.queueBind(q2,ex1,"warning");//8、发送消息String body="日志信息：XXXX";channel.basicPublish(ex1,"info",null,body.getBytes());//9、释放资源channel.close();connection.close();}catch (Exception e){e.printStackTrace();}

消费者：try{//1、建立连接工厂ConnectionFactory factory=new ConnectionFactory();//2、设置参数factory.setHost("192.168.10.130");  //默认是localhostfactory.setPort(5672);  //默认也是5672factory.setVirtualHost("/admin"); //默认是/factory.setUsername("admin"); //默认是guestfactory.setPassword("123456"); //默认是guest//3、创建连接Connection connection=factory.newConnection();//4、创建chanelChannel channel=connection.createChannel();String q2="test_direct_queue2";//5、创建队列queue/** String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments* 1、queue:队列名称* 2、durable：队列是否写入磁盘持久化* 3、exclusive：是否独占;connection关闭时是否关闭队列* 4、autodelete：自动删除* 5、arguments一些参数信息*** *///如果有名字叫hello_word就会创建没有则不会
//            channel.queueDeclare("hello_word",true,false,false,null);//6、发送消息/** (String exchange, String routingKey, boolean mandatory, BasicProperties props, byte[] body* 1、exchange:交换机，简单模式下为""* 2、routingkey：路由名称,没有交换机就必须和队列名一致* 3、props：配置信息* 4、b0dy；字节信息** */
//            String body="hello rabbitmq";
//            channel.basicPublish("","hello_word",null,body.getBytes());/** (String queue, DeliverCallback deliverCallback, ConsumerShutdownSignalCallback shutdownSignalCallback**1、queue队列名称* 2、* */Consumer consumer=new DefaultConsumer(channel){//回调方法，当收到消息后会自动执行该方法/** consumertag ： 标识* envelop：获取一些信息，交换机，路由key* properties：配置信息* body : 真实数据* */@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(new String(body));}};channel.basicConsume(q2,true,consumer);}catch (Exception e){e.printStackTrace();}

4.4、通配符模式

生产者：try{//1、建立连接工厂ConnectionFactory factory=new ConnectionFactory();//2、设置参数factory.setHost("192.168.10.130");  //默认是localhostfactory.setPort(5672);  //默认也是5672factory.setVirtualHost("/admin"); //默认是/factory.setUsername("admin"); //默认是guestfactory.setPassword("123456"); //默认是guest//3、创建连接Connection connection=factory.newConnection();//4、创建chanelChannel channel=connection.createChannel();//5、创建交换机/** String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)*1、exchange:交换机名称*2、type：交换机的类型* DIRECT("direct"):定向的方式FANOUT("fanout")：广播TOPIC("topic")：通配符HEADERS("headers"):参数匹配*3、durable：是否持久化* 4、autoDelete：自动删除* 5、internal：内部使用* 6、arguments：参数列表* */String ex1="test_topic";channel.exchangeDeclare(ex1, BuiltinExchangeType.TOPIC,true,false,false,null);//6、创建队列String q1="test_topic_queue1";String q2="test_topic_queue2";channel.queueDeclare(q1,true,false,false,null);channel.queueDeclare(q2,true,false,false,null);//7、绑定队列和交换机/** String queue, String exchange, String routingKey* routingkey=路由键，fanout是*             * */channel.queueBind(q1,ex1,"#.error");channel.queueBind(q1,ex1,"order.#");channel.queueBind(q2,ex1,"*.*");//8、发送消息String body="日志信息：XXXX";channel.basicPublish(ex1,"xxxx.error",null,body.getBytes());channel.basicPublish(ex1,"dasd.sada",null,body.getBytes());//9、释放资源channel.close();connection.close();}catch (Exception e){e.printStackTrace();}

5、spring整合rabbitmq

5.1、生产者

properties 配置：
----------------------------------------------------------------------------------------
rabbitmq.host=192.168.10.130
rabbitmq.port=5672
rabbitmq.username=admin
rabbitmq.password=123456
rabbitmq.virtual-host=/admin

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:context="http://www.springframework.org/schema/context"xmlns:rabbit="http://www.springframework.org/schema/rabbit"xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/contexthttps://www.springframework.org/schema/context/spring-context.xsdhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit.xsd"><!--加载配置文件--><context:property-placeholder location="classpath:rabbitmq.properties"/><!-- 定义rabbitmq connectionFactory --><rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"port="${rabbitmq.port}"username="${rabbitmq.username}"password="${rabbitmq.password}"virtual-host="${rabbitmq.virtual-host}"/><!--定义管理交换机、队列--><rabbit:admin connection-factory="connectionFactory"/><!--定义持久化队列，不存在则自动创建；不绑定到交换机则绑定到默认交换机默认交换机类型为direct，名字为：""，路由键为队列的名称--><rabbit:queue id="spring_queue" name="spring_queue" auto-declare="true"/><!-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~广播；所有队列都能收到消息~~~~~~~~~~~~~~~~~~~~~~~~~~~~ --><!--定义广播交换机中的持久化队列，不存在则自动创建--><rabbit:queue id="spring_fanout_queue_1" name="spring_fanout_queue_1" auto-declare="true"/><!--定义广播交换机中的持久化队列，不存在则自动创建--><rabbit:queue id="spring_fanout_queue_2" name="spring_fanout_queue_2" auto-declare="true"/><!--定义广播类型交换机；并绑定上述两个队列--><rabbit:fanout-exchange id="spring_fanout_exchange" name="spring_fanout_exchange" auto-declare="true"><rabbit:bindings><rabbit:binding queue="spring_fanout_queue_1"/><rabbit:binding queue="spring_fanout_queue_2"/></rabbit:bindings></rabbit:fanout-exchange><!-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~通配符；*匹配一个单词，#匹配多个单词 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~ --><!--定义广播交换机中的持久化队列，不存在则自动创建--><rabbit:queue id="spring_topic_queue_star" name="spring_topic_queue_star" auto-declare="true"/><!--定义广播交换机中的持久化队列，不存在则自动创建--><rabbit:queue id="spring_topic_queue_well" name="spring_topic_queue_well" auto-declare="true"/><!--定义广播交换机中的持久化队列，不存在则自动创建--><rabbit:queue id="spring_topic_queue_well2" name="spring_topic_queue_well2" auto-declare="true"/><rabbit:topic-exchange id="spring_topic_exchange" name="spring_topic_exchange" auto-declare="true"><rabbit:bindings><rabbit:binding pattern="heima.*" queue="spring_topic_queue_star"/><rabbit:binding pattern="heima.#" queue="spring_topic_queue_well"/><rabbit:binding pattern="itcast.#" queue="spring_topic_queue_well2"/></rabbit:bindings></rabbit:topic-exchange><!--定义rabbitTemplate对象操作可以在代码中方便发送消息--><rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>
</beans>

测试
-----------------------------------------------------------------------------------------
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations ="classpath:spring-rabbitmq-producer.xml")
public class Test {@Autowiredprivate RabbitTemplate rabbitTemplate;@org.junit.Testpublic void Test1(){rabbitTemplate.convertAndSend("spring_queue","hello word spring!");}@org.junit.Testpublic void Test2(){rabbitTemplate.convertAndSend("spring_fanout_exchange","","hello fanout");}}

5.2、消费者

properties 配置：
----------------------------------------------------------------------------------------
rabbitmq.host=192.168.10.130
rabbitmq.port=5672
rabbitmq.username=admin
rabbitmq.password=123456
rabbitmq.virtual-host=/admin

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:context="http://www.springframework.org/schema/context"xmlns:rabbit="http://www.springframework.org/schema/rabbit"xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/contexthttps://www.springframework.org/schema/context/spring-context.xsdhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit.xsd"><!--加载配置文件--><context:property-placeholder location="classpath:rabbitmq.properties"/><!-- 定义rabbitmq connectionFactory --><rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"port="${rabbitmq.port}"username="${rabbitmq.username}"password="${rabbitmq.password}"virtual-host="${rabbitmq.virtual-host}"/><bean id="springQueueListener" class="com.itheima.rabbitmq.listener.SpringQueueListener"/><bean id="fanoutListener1" class="com.itheima.rabbitmq.listener.FanoutListener1"/><bean id="fanoutListener2" class="com.itheima.rabbitmq.listener.FanoutListener2"/><bean id="topicListenerStar" class="com.itheima.rabbitmq.listener.TopicListenerStar"/><bean id="topicListenerWell" class="com.itheima.rabbitmq.listener.TopicListenerWell"/><bean id="topicListenerWell2" class="com.itheima.rabbitmq.listener.TopicListenerWell2"/><rabbit:listener-container connection-factory="connectionFactory" auto-declare="true"><rabbit:listener ref="springQueueListener" queue-names="spring_queue"/><rabbit:listener ref="fanoutListener1" queue-names="spring_fanout_queue_1"/><rabbit:listener ref="fanoutListener2" queue-names="spring_fanout_queue_2"/><rabbit:listener ref="topicListenerStar" queue-names="spring_topic_queue_star"/><rabbit:listener ref="topicListenerWell" queue-names="spring_topic_queue_well"/><rabbit:listener ref="topicListenerWell2" queue-names="spring_topic_queue_well2"/></rabbit:listener-container>
</beans>

监听
-----------------------------------------------------------------------------------------  package com.itheima.rabbitmq.listener;import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;public class SpringQueueListener implements MessageListener {@Overridepublic void onMessage(Message message) {System.out.println(new String(message.getBody()));}
}

测试
-----------------------------------------------------------------------------------------
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-consumer.xml")
public class Test {@org.junit.Testpublic void test1(){boolean flag=true;while (flag=true){}}
}

6、SpringBoot整合rabbitmq

6.1、依赖和配置

 <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>spring:rabbitmq:host: 192.168.10.130username: adminpassword: 123456virtual-host: /adminport: 5672

6.2、消费者创建交换机和建立队列以及测试

@Configuration
public class rabbitmqconfig {//1、交换机@Bean("bootExchange")public Exchange bootExchange(){return ExchangeBuilder.topicExchange("Exchange_Name").durable(true).build();}//2、queue@Bean("bootQueue")public Queue bootQueue(){return QueueBuilder.durable("Queue_name").build();}//3、bind@Beanpublic Binding bind(@Qualifier("bootExchange")Exchange exchange,@Qualifier("bootQueue") Queue queue){return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();}
}
-----------------------------------------------------------------------------------------
@SpringBootTest
@RunWith(SpringRunner.class)
public class test {@Autowiredprivate RabbitTemplate rabbitTemplate;@org.junit.Testpublic void Test1(){rabbitTemplate.convertAndSend("Exchange_Name","boot.333","hello springboot");}
}

6.3、消费者创建监听

@Component
public class RabbitMqListener {@RabbitListener(queues = "Queue_name")public void RabbitListener(Message message){System.out.println(message);}
}

7、高级特性

7.1、消息的可靠投递（作用消息到交换机之间）

1、开启确认模式

<!-- 定义rabbitmq connectionFactory --><rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"port="${rabbitmq.port}"username="${rabbitmq.username}"password="${rabbitmq.password}"virtual-host="${rabbitmq.virtual-host}"publisher-confirms="true"    --开启确认模式/>

2、使用

 @org.junit.Testpublic void Test3(){//1、开启确认模式//2、闯进一个comfirm回调函数/** 1、correlationData配置信息* 2、ack exchange是否成功收到消息 t收到f未收到* 3、causer  失败的原因** */rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String causer) {System.out.println("confirm执行了");if (ack){System.out.println("成功了"+causer);}else {System.out.println("失败"+causer);}}});rabbitTemplate.convertAndSend("test_exchange_confirm","confirm","hello confirm");}

7.2、回退模式（交换机路由到queue失败时回退）

1、开启回退模式

<rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"port="${rabbitmq.port}"username="${rabbitmq.username}"password="${rabbitmq.password}"virtual-host="${rabbitmq.virtual-host}"publisher-confirms="true"publisher-returns="true"  --开启回退模式/>

2、编写代码

 @org.junit.Testpublic void Test4(){//1、开启回退模式//2、闯进一个comfirm回调函数/** 1、message 消息* 2、replayCode 失败代码* 3、repayTest  失败信息* 4、exchange 交换机* 5、routingKey 路由键* *///设置交换机处理消息失败模式rabbitTemplate.setMandatory(true);rabbitTemplate.setReturnCallback((Message message, int replayCode, String repayTest, String exchange, String routingKey)->{System.out.println("Return 执行了");System.out.println(message);System.out.println(replayCode);System.out.println(repayTest);System.out.println(exchange);System.out.println(routingKey);});rabbitTemplate.convertAndSend("test_exchange_confirm","confirm11","hello confirm");}

7.3、Consumer ACK

消费者段：
auto-declare="true" acknowledge="manual"  开启自动接收后删除和手动接收消息

@Component
public class AckListener  implements ChannelAwareMessageListener {@Overridepublic void onMessage(Message message, Channel channel) throws Exception {System.out.println(new String(message.getBody()));/*long deliveryTag, boolean multiple, boolean requeue** 1、信息标签* 2、是否确认收到直到包含上述的标签* 3、是否重回队列*** */long  s= 13123123;try {channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);}catch (Exception e){channel.basicNack(message.getMessageProperties().getDeliveryTag(),true,true);}}
}

前提：acknowledge="manual"  prefetch="1"  //开启手动提交个每次获取值

@Component(value = "qosListener")
public class QosListener implements ChannelAwareMessageListener {@Overridepublic void onMessage(Message message, Channel channel) throws Exception {Thread.sleep(1000);System.out.println(new String(message.getBody()));channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);
//        try {//            channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);
//        }catch (Exception e){//            channel.basicNack(message.getMessageProperties().getDeliveryTag(),true,true);
//        }}
}

7.4、消费端限流（限制每次手动确认前接收信息数）

参数prfetch=“”，前提开启手动确认

    <rabbit:listener-container connection-factory="connectionFactory" auto-declare="true"           acknowledge="manual"  prefetch="1" >
<!--        <rabbit:listener ref="ackListener" queue-names="test_queue_confirm"/>--><rabbit:listener ref="qosListener" queue-names="test_queue_confirm"/></rabbit:listener-container>

7.5、TTL

1、对整个队列设置过期时间

    <rabbit:queue id="test_queue_ttl" name="test_queue_ttl" auto-declare="true"><rabbit:queue-arguments>//声明queue并设置过期参数<entry key="x-message-ttl" value-type="java.lang.Integer" value="10000"/></rabbit:queue-arguments></rabbit:queue><rabbit:topic-exchange id="test_exchange_ttl" name="test_exchange_ttl" auto-declare="true" ><rabbit:bindings><rabbit:binding pattern="ttl.#" queue="test_queue_ttl"/></rabbit:bindings></rabbit:topic-exchange>

2、对一条消息设置过期时间

    @org.junit.Testpublic void Test6(){for (int i = 0; i <3 ; i++) {rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.ddd", "hello ttl", new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {message.getMessageProperties().setExpiration("5000");return message;}});}}

注意：单独设置个和整体都设置时采取最短的；有ttl的消息前如果存在没有ttl则会等到没有ttl的消息消失才会消失。

8、死信队列

8.1、消息在什么情况下会变成死信

1、队列消息长度到达限制
2、消费者拒收消息，手动接受消息并且拒接收受消息并没有返回队列
3、原队列对消息设置过期时间，消息达到时间未消费

8.2、绑定死信交换机

<entry key="x-dead-letter-exchange" value="test_exchange_dead"/>  //设置死信交换机
<entry key="x-dead-letter-routing-key" value="dead.xxx"/>         //设置死信路由
-----------------------------------------------------------------------------------------
<!--死信队列--><rabbit:queue id="test_queue_dead" name="test_queue_dead" auto-declare="true"></rabbit:queue><rabbit:queue id="test_queue_ttl" name="test_queue_ttl" auto-declare="true"><rabbit:queue-arguments>
<!--  <entry key="x-message-ttl" value-type="java.lang.Integer" value="3000"/>--><entry key="x-dead-letter-exchange" value="test_exchange_dead"/><entry key="x-dead-letter-routing-key" value="dead.xxx"/><entry key="x-max-length" value="10" value-type="java.lang.Integer" /></rabbit:queue-arguments></rabbit:queue>
<!--    TTL exchange--><rabbit:topic-exchange id="test_exchange_ttl" name="test_exchange_ttl" auto-declare="true" ><rabbit:bindings><rabbit:binding pattern="ttl.#" queue="test_queue_ttl"/></rabbit:bindings></rabbit:topic-exchange><!--    死信 exchange--><rabbit:topic-exchange id="test_exchange_dead" name="test_exchange_dead" auto-declare="true" ><rabbit:bindings><rabbit:binding pattern="dead.#" queue="test_queue_dead"/></rabbit:bindings></rabbit:topic-exchange>

9、延迟队列

定义：发布消息后不会立即可以被消费，在超过锁定的时间限制后才可进行消费。

场景：1、订单回滚

原理：TTL+死信队列

10、日志与监控

日志路径： /var/log/rabbitmq

rabbitMq控制台命令

11、消息追踪

1、fireHouse

2、rabbitmq_tracing

rabbitmq-plugins list    //查看rabbit插件情况
前面有e*表示以及启用
启用插件
rabbitmq-plugins enable 插件名

12、RabbitMq应用问题

1、消息可靠性保障

2、消息幂等性的保障

13、集群搭建

1、单机多实例部署

由于某些因素的限制，有时候你不得不在一台机器上去搭建一个rabbitmq集群，这个有点类似zookeeper的单机版。真实生成环境还是要配成多机集群的。有关怎么配置多机集群的可以参考其他的资料，这里主要论述如何在单机中配置多个rabbitmq实例。

主要参考官方文档：https://www.rabbitmq.com/clustering.html

首先确保RabbitMQ运行没有问题

[root@super ~]# rabbitmqctl status
Status of node rabbit@super ...
[{pid,10232},{running_applications,[{rabbitmq_management,"RabbitMQ Management Console","3.6.5"},{rabbitmq_web_dispatch,"RabbitMQ Web Dispatcher","3.6.5"},{webmachine,"webmachine","1.10.3"},{mochiweb,"MochiMedia Web Server","2.13.1"},{rabbitmq_management_agent,"RabbitMQ Management Agent","3.6.5"},{rabbit,"RabbitMQ","3.6.5"},{os_mon,"CPO  CXC 138 46","2.4"},{syntax_tools,"Syntax tools","1.7"},{inets,"INETS  CXC 138 49","6.2"},{amqp_client,"RabbitMQ AMQP Client","3.6.5"},{rabbit_common,[],"3.6.5"},{ssl,"Erlang/OTP SSL application","7.3"},{public_key,"Public key infrastructure","1.1.1"},{asn1,"The Erlang ASN1 compiler version 4.0.2","4.0.2"},{ranch,"Socket acceptor pool for TCP protocols.","1.2.1"},{mnesia,"MNESIA  CXC 138 12","4.13.3"},{compiler,"ERTS  CXC 138 10","6.0.3"},{crypto,"CRYPTO","3.6.3"},{xmerl,"XML parser","1.3.10"},{sasl,"SASL  CXC 138 11","2.7"},{stdlib,"ERTS  CXC 138 10","2.8"},{kernel,"ERTS  CXC 138 10","4.2"}]},{os,{unix,linux}},{erlang_version,"Erlang/OTP 18 [erts-7.3] [source] [64-bit] [async-threads:64] [hipe] [kernel-poll:true]\n"},{memory,[{total,56066752},{connection_readers,0},{connection_writers,0},{connection_channels,0},{connection_other,2680},{queue_procs,268248},{queue_slave_procs,0},{plugins,1131936},{other_proc,18144280},{mnesia,125304},{mgmt_db,921312},{msg_index,69440},{other_ets,1413664},{binary,755736},{code,27824046},{atom,1000601},{other_system,4409505}]},{alarms,[]},{listeners,[{clustering,25672,"::"},{amqp,5672,"::"}]},{vm_memory_high_watermark,0.4},{vm_memory_limit,411294105},{disk_free_limit,50000000},{disk_free,13270233088},{file_descriptors,[{total_limit,924},{total_used,6},{sockets_limit,829},{sockets_used,0}]},{processes,[{limit,1048576},{used,262}]},{run_queue,0},{uptime,43651},{kernel,{net_ticktime,60}}]

停止rabbitmq服务

[root@super sbin]# service rabbitmq-server stop
Stopping rabbitmq-server: rabbitmq-server.

启动第一个节点：

[root@super sbin]# RABBITMQ_NODE_PORT=5673 RABBITMQ_NODENAME=rabbit1 rabbitmq-server startRabbitMQ 3.6.5. Copyright (C) 2007-2016 Pivotal Software, Inc.##  ##      Licensed under the MPL.  See http://www.rabbitmq.com/##  ############  Logs: /var/log/rabbitmq/rabbit1.log######  ##        /var/log/rabbitmq/rabbit1-sasl.log##########Starting broker...completed with 6 plugins.

启动第二个节点：

web管理插件端口占用,所以还要指定其web插件占用的端口号。

[root@super ~]# RABBITMQ_NODE_PORT=5674 RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15674}]" RABBITMQ_NODENAME=rabbit2 rabbitmq-server startRabbitMQ 3.6.5. Copyright (C) 2007-2016 Pivotal Software, Inc.##  ##      Licensed under the MPL.  See http://www.rabbitmq.com/##  ############  Logs: /var/log/rabbitmq/rabbit2.log######  ##        /var/log/rabbitmq/rabbit2-sasl.log##########Starting broker...completed with 6 plugins.

结束命令：

rabbitmqctl -n rabbit1 stop
rabbitmqctl -n rabbit2 stop

rabbit1操作作为主节点：

[root@super ~]# rabbitmqctl -n rabbit1 stop_app
Stopping node rabbit1@super ...
[root@super ~]# rabbitmqctl -n rabbit1 reset
Resetting node rabbit1@super ...
[root@super ~]# rabbitmqctl -n rabbit1 start_app
Starting node rabbit1@super ...
[root@super ~]#

rabbit2操作为从节点：

[root@super ~]# rabbitmqctl -n rabbit2 stop_app
Stopping node rabbit2@super ...
[root@super ~]# rabbitmqctl -n rabbit2 reset
Resetting node rabbit2@super ...
[root@super ~]# rabbitmqctl -n rabbit2 join_cluster rabbit1@'super' ###''内是主机名换成自己的
Clustering node rabbit2@super with rabbit1@super ...
[root@super ~]# rabbitmqctl -n rabbit2 start_app
Starting node rabbit2@super ...

查看集群状态：

[root@super ~]# rabbitmqctl cluster_status -n rabbit1
Cluster status of node rabbit1@super ...
[{nodes,[{disc,[rabbit1@super,rabbit2@super]}]},{running_nodes,[rabbit2@super,rabbit1@super]},{cluster_name,<<"rabbit1@super">>},{partitions,[]},{alarms,[{rabbit2@super,[]},{rabbit1@super,[]}]}]

web监控：

2、集群管理

rabbitmqctl join_cluster {cluster_node} [–ram]
将节点加入指定集群中。在这个命令执行前需要停止RabbitMQ应用并重置节点。

rabbitmqctl cluster_status
显示集群的状态。

rabbitmqctl change_cluster_node_type {disc|ram}
修改集群节点的类型。在这个命令执行前需要停止RabbitMQ应用。

rabbitmqctl forget_cluster_node [–offline]
将节点从集群中删除，允许离线执行。

rabbitmqctl update_cluster_nodes {clusternode}

在集群中的节点应用启动前咨询clusternode节点的最新信息，并更新相应的集群信息。这个和join_cluster不同，它不加入集群。考虑这样一种情况，节点A和节点B都在集群中，当节点A离线了，节点C又和节点B组成了一个集群，然后节点B又离开了集群，当A醒来的时候，它会尝试联系节点B，但是这样会失败，因为节点B已经不在集群中了。

rabbitmqctl cancel_sync_queue [-p vhost] {queue}
取消队列queue同步镜像的操作。

rabbitmqctl set_cluster_name {name}
设置集群名称。集群名称在客户端连接时会通报给客户端。Federation和Shovel插件也会有用到集群名称的地方。集群名称默认是集群中第一个节点的名称，通过这个命令可以重新设置。

3、 RabbitMQ镜像集群配置

上面已经完成RabbitMQ默认集群模式，但并不保证队列的高可用性，尽管交换机、绑定这些可以复制到集群里的任何一个节点，但是队列内容不会复制。虽然该模式解决一项目组节点压力，但队列节点宕机直接导致该队列无法应用，只能等待重启，所以要想在队列节点宕机或故障也能正常应用，就要复制队列内容到集群里的每个节点，必须要创建镜像队列。

镜像队列是基于普通的集群模式的，然后再添加一些策略，所以你还是得先配置普通集群，然后才能设置镜像队列，我们就以上面的集群接着做。

设置的镜像队列可以通过开启的网页的管理端Admin->Policies，也可以通过命令。

rabbitmqctl set_policy my_ha “^” ‘{“ha-mode”:“all”}’

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-EDtFZxgP-1661862392857)(RabbitMQ学习.assets/1566072300852.png)]

Name:策略名称

Pattern：匹配的规则，如果是匹配所有的队列，是^.

Definition:使用ha-mode模式中的all，也就是同步所有匹配的队列。问号链接帮助文档。

4、负载均衡-HAProxy

HAProxy提供高可用性、负载均衡以及基于TCP和HTTP应用的代理，支持虚拟主机，它是免费、快速并且可靠的一种解决方案,包括Twitter，Reddit，StackOverflow，GitHub在内的多家知名互联网公司在使用。HAProxy实现了一种事件驱动、单一进程模型，此模型支持非常大的并发连接数。

4.1、安装HAProxy

//下载依赖包
yum install gcc vim wget
//上传haproxy源码包
//解压
tar -zxvf haproxy-1.6.5.tar.gz -C /usr/local
//进入目录、进行编译、安装
cd /usr/local/haproxy-1.6.5
make TARGET=linux31 PREFIX=/usr/local/haproxy
make install PREFIX=/usr/local/haproxy
mkdir /etc/haproxy
//赋权
groupadd -r -g 149 haproxy
useradd -g haproxy -r -s /sbin/nologin -u 149 haproxy
//创建haproxy配置文件
mkdir /etc/haproxy
vim /etc/haproxy/haproxy.cfg

4.2、配置HAProxy

配置文件路径：/etc/haproxy/haproxy.cfg

#logging options
globallog 127.0.0.1 local0 infomaxconn 5120chroot /usr/local/haproxyuid 99gid 99daemonquietnbproc 20pidfile /var/run/haproxy.piddefaultslog globalmode tcpoption tcplogoption dontlognullretries 3option redispatchmaxconn 2000contimeout 5sclitimeout 60ssrvtimeout 15s
#front-end IP for consumers and producterslisten rabbitmq_clusterbind 0.0.0.0:5672mode tcp#balance url_param userid#balance url_param session_id check_post 64#balance hdr(User-Agent)#balance hdr(host)#balance hdr(Host) use_domain_only#balance rdp-cookie#balance leastconn#balance source //ipbalance roundrobinserver node1 127.0.0.1:5673 check inter 5000 rise 2 fall 2server node2 127.0.0.1:5674 check inter 5000 rise 2 fall 2listen statsbind 172.16.98.133:8100mode httpoption httplogstats enablestats uri /rabbitmq-statsstats refresh 5s

启动HAproxy负载

/usr/local/haproxy/sbin/haproxy -f /etc/haproxy/haproxy.cfg
//查看haproxy进程状态
ps -ef | grep haproxy访问如下地址对mq节点进行监控
http://172.16.98.133:8100/rabbitmq-stats

代码中访问mq集群地址，则变为访问haproxy地址:5672

proxy
mkdir /etc/haproxy
//赋权
groupadd -r -g 149 haproxy
useradd -g haproxy -r -s /sbin/nologin -u 149 haproxy
//创建haproxy配置文件
mkdir /etc/haproxy
vim /etc/haproxy/haproxy.cfg

##### 4.2、 配置HAProxy配置文件路径：/etc/haproxy/haproxy.cfg```shell
#logging options
globallog 127.0.0.1 local0 infomaxconn 5120chroot /usr/local/haproxyuid 99gid 99daemonquietnbproc 20pidfile /var/run/haproxy.piddefaultslog globalmode tcpoption tcplogoption dontlognullretries 3option redispatchmaxconn 2000contimeout 5sclitimeout 60ssrvtimeout 15s
#front-end IP for consumers and producterslisten rabbitmq_clusterbind 0.0.0.0:5672mode tcp#balance url_param userid#balance url_param session_id check_post 64#balance hdr(User-Agent)#balance hdr(host)#balance hdr(Host) use_domain_only#balance rdp-cookie#balance leastconn#balance source //ipbalance roundrobinserver node1 127.0.0.1:5673 check inter 5000 rise 2 fall 2server node2 127.0.0.1:5674 check inter 5000 rise 2 fall 2listen statsbind 172.16.98.133:8100mode httpoption httplogstats enablestats uri /rabbitmq-statsstats refresh 5s

启动HAproxy负载

/usr/local/haproxy/sbin/haproxy -f /etc/haproxy/haproxy.cfg
//查看haproxy进程状态
ps -ef | grep haproxy访问如下地址对mq节点进行监控
http://172.16.98.133:8100/rabbitmq-stats

代码中访问mq集群地址，则变为访问haproxy地址:5672

RabbitMQ学习相关推荐

RabbitMQ学习系列二：.net 环境下 C#代码使用 RabbitMQ 消息队列
上一篇已经讲了Rabbitmq如何在Windows平台安装,不懂请移步:RabbitMQ学习系列一:windows下安装RabbitMQ服务一.理论: .net环境下,C#代码调用RabbitMQ消 ...
RabbitMQ学习总结第一篇：理论篇
目录 RabbitMQ学习总结第一篇:理论篇 RabbitMQ学习总结第二篇:快速入门HelloWorld RabbitMQ学习总结第三篇:工作队列Work Queue RabbitMQ学习总结 ...
RabbitMQ学习笔记四：RabbitMQ命令（附疑难问题解决）
RabbitMQ学习笔记四:RabbitMQ命令(附疑难问题解决) 参考文章: (1)RabbitMQ学习笔记四:RabbitMQ命令(附疑难问题解决) (2)https://www.cnblogs. ...
rabbitmq 学习-9- RpcClient发送消息和同步接收消息原理
rabbitmq 学习-9- RpcClient发送消息和同步接收消息原理
RabbitMQ学习之集群消息可靠性测试
之前介绍过关于消息发送和接收的可靠性:RabbitMQ学习之消息可靠性及特性下面主要介绍一下集群环境下,rabbitmq实例宕机的情况下,消息的可靠性.验证rabbitmq版本[3.4.1]. ...
官网英文版学习——RabbitMQ学习笔记（二）RabbitMQ安装
一.安装RabbitMQ的依赖Erlang 要进行RabbitMQ学习,首先需要进行RabbitMQ服务的安装,安装我们可以根据官网指导进行http://www.rabbitmq.com/downlo ...
RabbitMQ学习笔记（3）----RabbitMQ Worker的使用
1. Woker队列结构图这里表示一个生产者生产了消息发送到队列中,但是确有两个消费者在消费同一个队列中的消息. 2. 创建一个生产者 Producer如下: package com.wangx.r ...
RabbitMQ 学习笔记
RabbitMQ 学习笔记 RabbitMQ 学习笔记 1. 中间件 1.1 什么是中间件 1.2 为什么要使用消息中间件 1.3 中间件特点 1.4 在项目中什么时候使用中间件技术 2. 中间件技术 ...
Rabbitmq学习笔记(尚硅谷2021)
Rabbitmq学习笔记 (尚硅谷) 1.MQ 的概念 1.1 什么是 MQ? 1.2 为什么要用 MQ? 削峰解耦异步 1.3 MQ 的分类 ActiveMQ Kafka RocketMQ Ra ...

RabbitMQ学习

RabbitMQ学习

1、概述

2、rabbit mq安装

1. 安装依赖环境

2. 安装Erlang

3. 安装RabbitMQ

4. 开启管理界面及配置

5. 启动

6. 配置虚拟主机及用户

6.1. 用户角色

6.2. Virtual Hosts配置

6.2.1. 创建Virtual Hosts

6.2.2. 设置Virtual Hosts权限

3、快速搭建（简单模式）

4、工作模式

4.1、workqueue 工作队列

4.2、订阅模式

4.3、路由模式

4.4、通配符模式

5、spring整合rabbitmq

5.1、生产者

5.2、消费者

6、SpringBoot整合rabbitmq

6.1、依赖和配置

6.2、消费者创建交换机和建立队列以及测试

6.3、消费者创建监听

7、高级特性

7.1、消息的可靠投递（作用消息到交换机之间）

7.2、回退模式 （交换机路由到queue失败时回退）

7.3、Consumer ACK

7.4、消费端限流（限制每次手动确认前接收信息数）

7.5、TTL

8、死信队列

8.1、消息在什么情况下会变成死信

8.2、绑定死信交换机

9、延迟队列

10、日志与监控

11、消息追踪

12、RabbitMq应用问题

1、消息可靠性保障

2、消息幂等性的保障

13、集群搭建

1、 单机多实例部署

2、 集群管理

3、 RabbitMQ镜像集群配置

4、 负载均衡-HAProxy

4.1、 安装HAProxy

4.2、 配置HAProxy

RabbitMQ学习相关推荐

最新文章

热门文章

7.2、回退模式（交换机路由到queue失败时回退）

1、单机多实例部署

2、集群管理

4、负载均衡-HAProxy

4.1、安装HAProxy

4.2、配置HAProxy