设为首页收藏本站
开启辅助访问

RabbitMQ使用教程(超详细)

 您所在的位置：网站首页 › rabbitmq开发 › RabbitMQ使用教程(超详细)

RabbitMQ使用教程(超详细)

#RabbitMQ使用教程(超详细)| 来源: 网络整理| 查看: 265

推荐springCloud教程： https://blog.csdn.net/hellozpc/article/details/83692496

推荐Springboot2.0教程： https://blog.csdn.net/hellozpc/article/details/82531834

文章目录 RabbitMQ实战教程1.什么是MQ2.RabbitMQ2.1.RabbitMQ的简介2.2.官网2.3.MQ的其他产品2.4.学习5种队列2.5.安装文档 3.搭建RabbitMQ环境3.1.下载3.2.windows下安装3.3.Linux下安装3.4.安装的注意事项3.5.安装完成后操作 4.添加用户4.1.添加admin用户4.2.用户角色4.3.创建Virtual Hosts4.4.管理界面中的功能 5.学习五种队列5.1.导入my-rabbitmq项目5.2.简单队列5.3.Work模式5.4.Work模式的“能者多劳”5.5.消息的确认模式5.6.订阅模式5.7.路由模式5.8.主题模式（通配符模式） 6.Spring-Rabbit6.1.Spring项目6.2.简介6.3.使用6.4.持久化交换机和队列 7.Spring集成RabbitMQ一个完整案例7.1.在A系统中发送消息到交换机7.2.在B系统接收消息7.3.在C系统中接收消息 8.Springboot集成RabbitMQ8.1.简单队列8.2.多对多使用（Work模式）8.3.Topic Exchange（主题模式）8.4.Fanout Exchange（订阅模式） 9.总结

RabbitMQ实战教程 1.什么是MQ 消息队列（Message Queue，简称MQ），从字面意思上看，本质是个队列，FIFO先入先出，只不过队列中存放的内容是message而已。 其主要用途：不同进程Process/线程Thread之间通信。

为什么会产生消息队列？有几个原因：

不同进程（process）之间传递消息时，两个进程之间耦合程度过高，改动一个进程，引发必须修改另一个进程，为了隔离这两个进程，在两进程间抽离出一层（一个模块），所有两进程之间传递的消息，都必须通过消息队列来传递，单独修改某一个进程，不会影响另一个；

不同进程（process）之间传递消息时，为了实现标准化，将消息的格式规范化了，并且，某一个进程接受的消息太多，一下子无法处理完，并且也有先后顺序，必须对收到的消息进行排队，因此诞生了事实上的消息队列；

关于消息队列的详细介绍请参阅： 《Java帝国之消息队列》 《一个故事告诉你什么是消息队列》 《到底什么时候该使用MQ》

MQ框架非常之多，比较流行的有RabbitMq、ActiveMq、ZeroMq、kafka，以及阿里开源的RocketMQ。本文主要介绍RabbitMq。

本教程pdf及代码下载地址： 代码：https://download.csdn.net/download/zpcandzhj/10585077 教程：https://download.csdn.net/download/zpcandzhj/10585092

2.RabbitMQ 2.1.RabbitMQ的简介

这里写图片描述 开发语言：Erlang – 面向并发的编程语言。

这里写图片描述

2.1.1.AMQP AMQP是消息队列的一个协议。

这里写图片描述

2.2.官网

这里写图片描述

2.3.MQ的其他产品

这里写图片描述

2.4.学习5种队列

这里写图片描述

2.5.安装文档

这里写图片描述

3.搭建RabbitMQ环境 3.1.下载

下载地址：http://www.rabbitmq.com/download.html

3.2.windows下安装

3.2.1.安装Erlang 下载：http://www.erlang.org/download/otp_win64_17.3.exe 安装： 这里写图片描述 这里写图片描述 这里写图片描述 这里写图片描述 这里写图片描述 安装完成。

3.2.2.安装RabbitMQ 这里写图片描述 这里写图片描述 这里写图片描述 安装完成。

开始菜单里出现如下选项： 这里写图片描述

启动、停止、重新安装等。

3.2.3.启用管理工具 1、双击这里写图片描述 2、进入C:\Program Files (x86)\RabbitMQ Server\rabbitmq_server-3.4.1\sbin输入命令： rabbitmq-plugins enable rabbitmq_management 这里写图片描述

这样就启动了管理工具，可以试一下命令： 停止：net stop RabbitMQ 启动：net start RabbitMQ

3、在浏览器中输入地址查看：http://127.0.0.1:15672/ 这里写图片描述 4、使用默认账号登录：guest/ guest

3.3.Linux下安装

3.3.1.安装Erlang 3.3.2.添加yum支持 cd /usr/local/src/ mkdir rabbitmq cd rabbitmq

wget http://packages.erlang-solutions.com/erlang-solutions-1.0-1.noarch.rpm rpm -Uvh erlang-solutions-1.0-1.noarch.rpm

rpm --import http://packages.erlang-solutions.com/rpm/erlang_solutions.asc

使用yum安装： sudo yum install erlang 这里写图片描述

3.3.3.安装RabbitMQ 上传rabbitmq-server-3.4.1-1.noarch.rpm文件到/usr/local/src/rabbitmq/ 安装： rpm -ivh rabbitmq-server-3.4.1-1.noarch.rpm

3.3.4.启动、停止 service rabbitmq-server start service rabbitmq-server stop service rabbitmq-server restart 3.3.5.设置开机启动 chkconfig rabbitmq-server on 3.3.6.设置配置文件 cd /etc/rabbitmq cp /usr/share/doc/rabbitmq-server-3.4.1/rabbitmq.config.example /etc/rabbitmq/

mv rabbitmq.config.example rabbitmq.config 3.3.7.开启用户远程访问 vi /etc/rabbitmq/rabbitmq.config 这里写图片描述 注意要去掉后面的逗号。 3.3.8.开启web界面管理工具 rabbitmq-plugins enable rabbitmq_management service rabbitmq-server restart 3.3.9.防火墙开放15672端口 /sbin/iptables -I INPUT -p tcp --dport 15672 -j ACCEPT /etc/rc.d/init.d/iptables save

3.4.安装的注意事项

1、推荐使用默认的安装路径 2、系统用户名必须是英文 Win10改名字非常麻烦，具体方法百度 这里写图片描述 3、计算机名必须是英文 这里写图片描述 4、系统的用户必须是管理员

如果安装失败应该如何解决： 1、重装系统 – 不推荐 2、将RabbitMQ安装到linux虚拟机中 a)推荐 3、使用别人安装好的RabbitMQ服务 a)只要给你开通一个账户即可。 b)使用公用的RabbitMQ服务，在192.168.50.22 c)推荐

常见错误： 这里写图片描述

3.5.安装完成后操作

1、系统服务中有RabbitMQ服务，停止、启动、重启 这里写图片描述 2、打开命令行工具 这里写图片描述 如果找不到命令行工具,直接cd到相应目录： 这里写图片描述 输入命令rabbitmq-plugins enable rabbitmq_management启用管理插件 这里写图片描述 查看管理页面 这里写图片描述 通过默认账户 guest/guest 登录 如果能够登录，说明安装成功。 这里写图片描述

4.添加用户 4.1.添加admin用户

这里写图片描述

4.2.用户角色

1、超级管理员(administrator) 可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。 2、监控者(monitoring) 可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等) 3、策略制定者(policymaker) 可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。 4、普通管理者(management) 仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。 5、其他 无法登陆管理控制台，通常就是普通的生产者和消费者。

4.3.创建Virtual Hosts

这里写图片描述

选中Admin用户，设置权限： 这里写图片描述 看到权限已加： 这里写图片描述

4.4.管理界面中的功能

这里写图片描述

这里写图片描述

5.学习五种队列

这里写图片描述

5.1.导入my-rabbitmq项目

项目下载地址： https://download.csdn.net/download/zpcandzhj/10585077 这里写图片描述

5.2.简单队列

5.2.1.图示 这里写图片描述

P：消息的生产者 C：消息的消费者 红色：队列

生产者将消息发送到队列，消费者从队列中获取消息。 5.2.2.导入RabbitMQ的客户端依赖

com.rabbitmq amqp-client 3.4.1

5.2.3.获取MQ的连接

package com.zpc.rabbitmq.util; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.Connection; public class ConnectionUtil { public static Connection getConnection() throws Exception { //定义连接工厂 ConnectionFactory factory = new ConnectionFactory(); //设置服务地址 factory.setHost("localhost"); //端口 factory.setPort(5672); //设置账号信息，用户名、密码、vhost factory.setVirtualHost("testhost"); factory.setUsername("admin"); factory.setPassword("admin"); // 通过工程获取连接 Connection connection = factory.newConnection(); return connection; } }

5.2.4.生产者发送消息到队列

package com.zpc.rabbitmq.simple; import com.zpc.rabbitmq.util.ConnectionUtil; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; public class Send { private final static String QUEUE_NAME = "q_test_01"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); // 从连接中创建通道 Channel channel = connection.createChannel(); // 声明（创建）队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 消息内容 String message = "Hello World!"; channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); //关闭通道和连接 channel.close(); connection.close(); } }

5.2.5.管理工具中查看消息 这里写图片描述

点击上面的队列名称，查询具体的队列中的信息： 这里写图片描述 5.2.6.消费者从队列中获取消息

package com.zpc.rabbitmq.simple; import com.zpc.rabbitmq.util.ConnectionUtil; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.QueueingConsumer; public class Recv { private final static String QUEUE_NAME = "q_test_01"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); // 从连接中创建通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列 channel.basicConsume(QUEUE_NAME, true, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); } } } 5.3.Work模式

这里写图片描述

5.3.1.图示 这里写图片描述

一个生产者、2个消费者。

一个消息只能被一个消费者获取。 5.3.2.消费者1

package com.zpc.rabbitmq.work; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.QueueingConsumer; import com.zpc.rabbitmq.util.ConnectionUtil; public class Recv { private final static String QUEUE_NAME = "test_queue_work"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 同一时刻服务器只会发一条消息给消费者 //channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列，false表示手动返回完成状态，true表示自动 channel.basicConsume(QUEUE_NAME, true, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [y] Received '" + message + "'"); //休眠 Thread.sleep(10); // 返回确认状态，注释掉表示使用自动确认模式 //channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } }

5.3.3.消费者2

package com.zpc.rabbitmq.work; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.QueueingConsumer; import com.zpc.rabbitmq.util.ConnectionUtil; public class Recv2 { private final static String QUEUE_NAME = "test_queue_work"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 同一时刻服务器只会发一条消息给消费者 //channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列，false表示手动返回完成状态，true表示自动 channel.basicConsume(QUEUE_NAME, true, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); // 休眠1秒 Thread.sleep(1000); //下面这行注释掉表示使用自动确认模式 //channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } }

5.3.4.生产者 向队列中发送100条消息。

package com.zpc.rabbitmq.work; import com.zpc.rabbitmq.util.ConnectionUtil; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; public class Send { private final static String QUEUE_NAME = "test_queue_work"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); for (int i = 0; i private final static String EXCHANGE_NAME = "test_exchange_fanout"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明exchange channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); // 消息内容 String message = "Hello World!"; channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); channel.close(); connection.close(); } }

注意：消息发送到没有队列绑定的交换机时，消息将丢失，因为，交换机没有存储消息的能力，消息只能存在在队列中。 5.6.3.消费者1（看作是前台系统）

package com.zpc.rabbitmq.subscribe; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.QueueingConsumer; import com.zpc.rabbitmq.util.ConnectionUtil; public class Recv { private final static String QUEUE_NAME = "test_queue_work1"; private final static String EXCHANGE_NAME = "test_exchange_fanout"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); // 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列，手动返回完成 channel.basicConsume(QUEUE_NAME, false, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [Recv] Received '" + message + "'"); Thread.sleep(10); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } }

5.6.4.消费者2（看作是搜索系统）

package com.zpc.rabbitmq.subscribe; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.QueueingConsumer; import com.zpc.rabbitmq.util.ConnectionUtil; public class Recv2 { private final static String QUEUE_NAME = "test_queue_work2"; private final static String EXCHANGE_NAME = "test_exchange_fanout"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); // 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列，手动返回完成 channel.basicConsume(QUEUE_NAME, false, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [Recv2] Received '" + message + "'"); Thread.sleep(10); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } }

5.6.5.测试 测试结果： 同一个消息被多个消费者获取。一个消费者队列可以有多个消费者实例，只有其中一个消费者实例会消费到消息。

在管理工具中查看队列和交换机的绑定关系：

这里写图片描述

5.7.路由模式

这里写图片描述 5.7.1.图示 这里写图片描述

5.7.2.生产者 这里写图片描述 5.7.3.消费者1(假设是前台系统) 这里写图片描述 5.7.4.消费2（假设是搜索系统） 这里写图片描述

5.8.主题模式（通配符模式）

这里写图片描述

这里写图片描述

5.8.1.图示 这里写图片描述 同一个消息被多个消费者获取。一个消费者队列可以有多个消费者实例，只有其中一个消费者实例会消费到消息。

5.8.2.生产者

package com.zpc.rabbitmq.topic; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.zpc.rabbitmq.util.ConnectionUtil; public class Send { private final static String EXCHANGE_NAME = "test_exchange_topic"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明exchange channel.exchangeDeclare(EXCHANGE_NAME, "topic"); // 消息内容 String message = "Hello World!!"; channel.basicPublish(EXCHANGE_NAME, "routekey.1", null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); channel.close(); connection.close(); } }

5.8.3.消费者1（前台系统）

package com.zpc.rabbitmq.topic; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.QueueingConsumer; import com.zpc.rabbitmq.util.ConnectionUtil; public class Recv { private final static String QUEUE_NAME = "test_queue_topic_work_1"; private final static String EXCHANGE_NAME = "test_exchange_topic"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "routekey.*"); // 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列，手动返回完成 channel.basicConsume(QUEUE_NAME, false, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [Recv_x] Received '" + message + "'"); Thread.sleep(10); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } }

5.8.4.消费者2（搜索系统）

package com.zpc.rabbitmq.topic; import com.zpc.rabbitmq.util.ConnectionUtil; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.QueueingConsumer; public class Recv2 { private final static String QUEUE_NAME = "test_queue_topic_work_2"; private final static String EXCHANGE_NAME = "test_exchange_topic"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.*"); // 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列，手动返回完成 channel.basicConsume(QUEUE_NAME, false, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [Recv2_x] Received '" + message + "'"); Thread.sleep(10); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } } 6.Spring-Rabbit 6.1.Spring项目

http://spring.io/projects

这里写图片描述

6.2.简介

这里写图片描述 这里写图片描述

6.3.使用

6.3.1.消费者

package com.zpc.rabbitmq.spring; /** * 消费者 * * @author Evan */ public class Foo { //具体执行业务的方法 public void listen(String foo) { System.out.println("\n消费者： " + foo + "\n"); } }

6.3.2.生产者

package com.zpc.rabbitmq.spring; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; public class SpringMain { public static void main(final String... args) throws Exception { AbstractApplicationContext ctx = new ClassPathXmlApplicationContext( "classpath:spring/rabbitmq-context.xml"); //RabbitMQ模板 RabbitTemplate template = ctx.getBean(RabbitTemplate.class); //发送消息 template.convertAndSend("Hello, 鸟鹏!"); Thread.sleep(1000);// 休眠1秒 ctx.destroy(); //容器销毁 } }

6.3.3.配置文件 1、定义连接工厂

2、定义模板（可以指定交换机或队列）

3、定义队列、交换机、以及完成队列和交换机的绑定

4、定义监听

5、定义管理，用于管理队列、交换机等：

完整配置文件rabbitmq-context.xml

6.4.持久化交换机和队列

这里写图片描述

持久化：将交换机或队列的数据保存到磁盘，服务器宕机或重启之后依然存在。 非持久化：将交换机或队列的数据保存到内存，服务器宕机或重启之后将不存在。

非持久化的性能高于持久化。

如何选择持久化？非持久化？ – 看需求。

7.Spring集成RabbitMQ一个完整案例

创建三个系统A,B,C A作为生产者，B、C作为消费者(B,C作为web项目启动) 项目下载地址：https://download.csdn.net/download/zpcandzhj/10585077

7.1.在A系统中发送消息到交换机

7.1.1.导入依赖

4.0.0 com.zpc myrabbitA 0.0.1-SNAPSHOT jar myrabbit org.springframework.amqp spring-rabbit 1.4.0.RELEASE com.alibaba fastjson 1.2.47

7.1.2.队列和交换机的绑定关系 实现： 1、在配置文件中将队列和交换机完成绑定 2、可以在管理界面中完成绑定 a)绑定关系如果发生变化，需要修改配置文件，并且服务需要重启 b)管理更加灵活 c)更容易对绑定关系的权限管理，流程管理 本例选择第2种方式 7.1.3.配置 rabbitmq-context.xml

7.1.4.消息内容 方案： 1、消息内容使用对象做json序列化发送 a)数据大 b)有些数据其他人是可能用不到的 2、发送特定的业务字段，如id、操作类型

7.1.5.实现 生产者MsgSender.java：

package com.zpc.myrabbit; import com.alibaba.fastjson.JSON; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import java.text.SimpleDateFormat; import java.util.Date; import java.util.HashMap; import java.util.Map; /** * 消息生产者 */ public class MsgSender { public static void main(String[] args) throws Exception { AbstractApplicationContext ctx = new ClassPathXmlApplicationContext( "classpath:spring/rabbitmq-context.xml"); //RabbitMQ模板 RabbitTemplate template = ctx.getBean(RabbitTemplate.class); String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());//24小时制 //发送消息 Map msg = new HashMap(); msg.put("type", "1"); msg.put("date", date); template.convertAndSend("type2", JSON.toJSONString(msg)); Thread.sleep(1000);// 休眠1秒 ctx.destroy(); //容器销毁 } } 7.2.在B系统接收消息

7.2.1.导入依赖

4.0.0 com.zpc myrabbitB 0.0.1-SNAPSHOT war myrabbit 4.1.3.RELEASE 1.2.46 com.rabbitmq amqp-client 3.4.1 org.springframework.amqp spring-rabbit 1.4.0.RELEASE org.springframework spring-webmvc ${spring.version} com.alibaba fastjson 1.2.47 ${project.artifactId} org.apache.tomcat.maven tomcat7-maven-plugin /testRabbit UTF-8 8081

7.2.2.配置

7.2.3.具体处理逻辑

public class Listener { //具体执行业务的方法 public void listen(String msg) { System.out.println("\n消费者B开始处理消息： " + msg + "\n"); } }

7.2.4.在界面管理工具中完成绑定关系 选中定义好的交换机(exchange) 这里写图片描述 1）direct 这里写图片描述 2）fanout 这里写图片描述 3）topic 这里写图片描述

7.3.在C系统中接收消息

（和B系统配置差不多，无非是Q名和Q对应的处理逻辑变了）

7.3.1.配置

7.3.2.处理业务逻辑

public class Listener { //具体执行业务的方法 public void listen(String msg) { System.out.println("\n消费者C开始处理消息： " + msg + "\n"); } }

7.3.3.在管理工具中绑定队列和交换机 见7.2.4

7.3.4.测试 分别启动B,C两个web应用，然后运行A的MsgSender主方法发送消息，分别测试fanout、direct、topic三种类型

8.Springboot集成RabbitMQ springboot集成RabbitMQ非常简单，如果只是简单的使用配置非常少，springboot提供了spring-boot-starter-amqp对消息各种支持。 代码下载地址：https://download.csdn.net/download/zpcandzhj/10585077 8.1.简单队列

1、配置pom文件，主要是添加spring-boot-starter-amqp的支持

org.springframework.boot spring-boot-starter-amqp

2、配置application.properties文件 配置rabbitmq的安装地址、端口以及账户信息

spring.application.name=spirng-boot-rabbitmq spring.rabbitmq.host=127.0.0.1 spring.rabbitmq.port=5672 spring.rabbitmq.username=admin spring.rabbitmq.password=admin

3、配置队列

package com.zpc.rabbitmq; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RabbitConfig { @Bean public Queue queue() { return new Queue("q_hello"); } }

4、发送者

package com.zpc.rabbitmq; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import java.text.SimpleDateFormat; import java.util.Date; @Component public class HelloSender { @Autowired private AmqpTemplate rabbitTemplate; public void send() { String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());//24小时制 String context = "hello " + date; System.out.println("Sender : " + context); //简单对列的情况下routingKey即为Q名 this.rabbitTemplate.convertAndSend("q_hello", context); } }

5、接收者

package com.zpc.rabbitmq; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component @RabbitListener(queues = "q_hello") public class HelloReceiver { @RabbitHandler public void process(String hello) { System.out.println("Receiver : " + hello); } }

6、测试

package com.zpc.rabbitmq; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; @RunWith(SpringRunner.class) @SpringBootTest public class RabbitMqHelloTest { @Autowired private HelloSender helloSender; @Test public void hello() throws Exception { helloSender.send(); } } 8.2.多对多使用（Work模式）

注册两个Receiver:

package com.zpc.rabbitmq; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component @RabbitListener(queues = "q_hello") public class HelloReceiver2 { @RabbitHandler public void process(String hello) { System.out.println("Receiver2 : " + hello); } } @Test public void oneToMany() throws Exception { for (int i=0;i String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());//24小时制 String context = "hello " + i + " " + date; System.out.println("Sender : " + context); //简单对列的情况下routingKey即为Q名 this.rabbitTemplate.convertAndSend("q_hello", context); } 8.3.Topic Exchange（主题模式） topic 是RabbitMQ中最灵活的一种方式，可以根据routing_key自由的绑定不同的队列

首先对topic规则配置，这里使用两个队列(消费者)来演示。 1)配置队列，绑定交换机

package com.zpc.rabbitmq.topic; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.TopicExchange; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class TopicRabbitConfig { final static String message = "q_topic_message"; final static String messages = "q_topic_messages"; @Bean public Queue queueMessage() { return new Queue(TopicRabbitConfig.message); } @Bean public Queue queueMessages() { return new Queue(TopicRabbitConfig.messages); } /** * 声明一个Topic类型的交换机 * @return */ @Bean TopicExchange exchange() { return new TopicExchange("mybootexchange"); } /** * 绑定Q到交换机,并且指定routingKey * @param queueMessage * @param exchange * @return */ @Bean Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) { return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message"); } @Bean Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) { return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#"); } }

2)创建2个消费者 q_topic_message 和q_topic_messages

package com.zpc.rabbitmq.topic; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component @RabbitListener(queues = "q_topic_message") public class Receiver1 { @RabbitHandler public void process(String hello) { System.out.println("Receiver1 : " + hello); } } package com.zpc.rabbitmq.topic; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component @RabbitListener(queues = "q_topic_messages") public class Receiver2 { @RabbitHandler public void process(String hello) { System.out.println("Receiver2 : " + hello); } }

3)消息发送者（生产者）

package com.zpc.rabbitmq.topic; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @Component public class MsgSender { @Autowired private AmqpTemplate rabbitTemplate; public void send1() { String context = "hi, i am message 1"; System.out.println("Sender : " + context); this.rabbitTemplate.convertAndSend("mybootexchange", "topic.message", context); } public void send2() { String context = "hi, i am messages 2"; System.out.println("Sender : " + context); this.rabbitTemplate.convertAndSend("mybootexchange", "topic.messages", context); } }

send1方法会匹配到topic.#和topic.message，两个Receiver都可以收到消息，发送send2只有topic.#可以匹配所有只有Receiver2监听到消息。 4)测试

package com.zpc.rabbitmq.topic; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; @RunWith(SpringRunner.class) @SpringBootTest public class RabbitTopicTest { @Autowired private MsgSender msgSender; @Test public void send1() throws Exception { msgSender.send1(); } @Test public void send2() throws Exception { msgSender.send2(); } } 8.4.Fanout Exchange（订阅模式） Fanout 就是我们熟悉的广播模式或者订阅模式，给Fanout交换机发送消息，绑定了这个交换机的所有队列都收到这个消息。 1)配置队列，绑定交换机 package com.zpc.rabbitmq.fanout; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.FanoutExchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class FanoutRabbitConfig { @Bean public Queue aMessage() { return new Queue("q_fanout_A"); } @Bean public Queue bMessage() { return new Queue("q_fanout_B"); } @Bean public Queue cMessage() { return new Queue("q_fanout_C"); } @Bean FanoutExchange fanoutExchange() { return new FanoutExchange("mybootfanoutExchange"); } @Bean Binding bindingExchangeA(Queue aMessage, FanoutExchange fanoutExchange) { return BindingBuilder.bind(aMessage).to(fanoutExchange); } @Bean Binding bindingExchangeB(Queue bMessage, FanoutExchange fanoutExchange) { return BindingBuilder.bind(bMessage).to(fanoutExchange); } @Bean Binding bindingExchangeC(Queue cMessage, FanoutExchange fanoutExchange) { return BindingBuilder.bind(cMessage).to(fanoutExchange); } }

2）创建3个消费者

package com.zpc.rabbitmq.fanout; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component @RabbitListener(queues = "q_fanout_A") public class ReceiverA { @RabbitHandler public void process(String hello) { System.out.println("AReceiver : " + hello + "/n"); } } package com.zpc.rabbitmq.fanout; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component @RabbitListener(queues = "q_fanout_B") public class ReceiverB { @RabbitHandler public void process(String hello) { System.out.println("BReceiver : " + hello + "/n"); } } package com.zpc.rabbitmq.fanout; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component @RabbitListener(queues = "q_fanout_C") public class ReceiverC { @RabbitHandler public void process(String hello) { System.out.println("CReceiver : " + hello + "/n"); } }

3）生产者

package com.zpc.rabbitmq.fanout; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @Component public class MsgSenderFanout { @Autowired private AmqpTemplate rabbitTemplate; public void send() { String context = "hi, fanout msg "; System.out.println("Sender : " + context); this.rabbitTemplate.convertAndSend("mybootfanoutExchange","", context); } }

4）测试

package com.zpc.rabbitmq.fanout; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; @RunWith(SpringRunner.class) @SpringBootTest public class RabbitFanoutTest { @Autowired private MsgSenderFanout msgSender; @Test public void send1() throws Exception { msgSender.send(); } }

结果如下，三个消费者都收到消息： AReceiver : hi, fanout msg CReceiver : hi, fanout msg BReceiver : hi, fanout msg

9.总结

使用MQ实现商品数据的同步优势： 1、降低系统间耦合度 2、便于管理数据的同步（数据一致性）

推荐阅读 《RabbitMQ详解》 《大型网站技术架构：核心原理与案例分析》

推荐springCloud教程： https://blog.csdn.net/hellozpc/article/details/83692496

推荐Springboot2.0教程： https://blog.csdn.net/hellozpc/article/details/82531834



【本文地址】


今日新闻

推荐新闻

CopyRight 2018-2019 办公设备维修网 版权所有 豫ICP备15022753号-3