RabbitMQ基础入门

学习目标

能够说出什么是消息中间件
能够安装 RabbitMQ
能够编写 RabbitMQ 的入门程序
能够说出 RabbitMQ 的5种模式特征
能够使用 Spring 整合 RabbitMQ

MQ 的基本概念

MQ 的概述

MQ全称 Message Queue（消息队列），是在消息的传输过程中保存消息的容器。多用于分布式系统之间进行通信。
在这里插入图片描述

在这里插入图片描述

小結

MQ，消息队列，存储消息的中间件
分布式系统通信两种方式：直接远程调用 和 借助第三方完成间接通信
发送方称为生产者，接收方称为消费者

MQ的优势和劣势

优势

应用解耦

在这里插入图片描述
系统的耦合性越高，容错性就越低，可维护性就越低。

使用 MQ 使得应用间解耦，提升容错性和可维护性。

异步提速

在这里插入图片描述
一个下单操作耗时：20 + 300 + 300 + 300 = 920ms

用户点击完下单按钮后，需要等待920ms才能得到下单响应，太慢！
在这里插入图片描述
用户点击完下单按钮后，只需等待25ms就能得到下单响应 (20 + 5 = 25ms)。

提升用户体验和系统吞吐量（单位时间内处理请求的数目）。

削峰填谷

在这里插入图片描述

使用了 MQ 之后，限制消费消息的速度为1000，这样一来，高峰期产生的数据势必会被积压在 MQ 中，高峰就被“削”掉了，但是因为消息积压，在高峰期过后的一段时间内，消费消息的速度还是会维持在1000，直到消费完积压的消息，这就叫做“填谷”。

使用MQ后，可以提高系统稳定性。

小结

应用解耦：提高系统容错性和可维护性
异步提速：提升用户体验和系统吞吐量
削峰填谷：提高系统稳定性

劣势

在这里插入图片描述

系统可用性降低

系统引入的外部依赖越多，系统稳定性越差。一旦 MQ 宕机，就会对业务造成影响。如何保证MQ的高可用？

系统复杂度提高

MQ 的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过 MQ 进行异步调用。如何保证消息没有被重复消费？怎么处理消息丢失情况？那么保证消息传递的顺序性？

一致性问题

A 系统处理完业务，通过 MQ 给B、C、D三个系统发消息数据，如果 B 系统、C 系统处理成功，D 系统处理失败。如何保证消息数据处理的一致性？

小结

既然 MQ 有优势也有劣势，那么使用 MQ 需要满足什么条件呢？

生产者不需要从消费者处获得反馈。引入消息队列之前的直接调用，其接口的返回值应该为空，这才让明明下层的动作还没做，上层却当成动作做完了继续往后走，即所谓异步成为了可能。
允许短暂的不一致性。
确实是用了有效果。即解耦、提速、削峰这些方面的收益，超过加入MQ，管理MQ这些成本。

常见的 MQ 产品

目前业界有很多的 MQ 产品，例如 RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、MetaMq等，也有直接使用 Redis 充当消息队列的案例，而这些消息队列产品，各有侧重，在实际选型时，需要结合自身需求及 MQ 产品特征，综合考虑。
在这里插入图片描述
由于 RabbitMQ 综合能力强劲，所以接下来的课程中，我们将主要学习 RabbitMQ。

RabbitMQ 简介

AMQP，即 Advanced Message Queuing Protocol（高级消息队列协议），是一个网络协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。2006年，AMQP 规范发布。类比HTTP。
在这里插入图片描述
2007年，Rabbit 技术公司基于 AMQP 标准开发的 RabbitMQ 1.0 发布。RabbitMQ 采用 Erlang 语言开发。Erlang 语言由 Ericson 设计，专门为开发高并发和分布式系统的一种语言，在电信领域使用广泛。

RabbitMQ 基础架构如下图：
在这里插入图片描述
RabbitMQ 中的相关概念：

Broker：接收和分发消息的应用，RabbitMQ Server就是 Message Broker
Virtual host：出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时，可以划分出多个vhost，每个用户在自己的 vhost 创建 exchange／queue 等
Connection：publisher／consumer 和 broker 之间的 TCP 连接
Channel：如果每一次访问 RabbitMQ 都建立一个 Connection，在消息量大的时候建立 TCP Connection的开销将是巨大的，效率也较低。Channel 是在 connection 内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的 channel 进行通讯，AMQP method 包含了channel id 帮助客户端和message broker 识别 channel，所以 channel 之间是完全隔离的。Channel 作为轻量级的 Connection 极大减少了操作系统建立 TCP connection 的开销
Exchange：message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到queue 中去。常用的类型有：direct (point-to-point), topic (publish-subscribe) and fanout (multicast)
Queue：消息最终被送到这里等待 consumer 取走
Binding：exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key。Binding 信息被保存到 exchange 中的查询表中，用于 message 的分发依据

RabbitMQ 提供了 6 种工作模式：

简单模式
work queues
Publish/Subscribe 发布与订阅模式
Routing 路由模式
Topics 主题模式
RPC 远程调用模式（远程调用，不太算 MQ；暂不作介绍）

官网对应模式介绍：https://www.rabbitmq.com/getstarted.html
在这里插入图片描述

JMS

JMS 即 Java 消息服务（JavaMessage Service）应用程序接口，是一个 Java 平台中关于面向消息中间件的API
JMS 是 JavaEE 规范中的一种，类比JDBC
很多消息中间件都实现了JMS规范，例如：ActiveMQ。RabbitMQ 官方没有提供 JMS 的实现包，但是开源社区有

小结

RabbitMQ 是基于 AMQP 协议使用 Erlang 语言开发的一款消息队列产品。
RabbitMQ提供了6种工作模式，我们学习5种。这是今天的重点。
AMQP 是协议，类比HTTP。
JMS 是 API 规范接口，类比 JDBC。

RabbitMQ 的安装和配置

RabbitMQ 官方地址：http://www.rabbitmq.com/
安装文档：RabbitMQ 的安装和配置

RabbitMQ 快速入门

入门程序（简单模式HelloWorld）

需求：使用简单模式完成消息传递

步骤：

1. 创建maven工程（生成者、消费者）

创建生产者工程和消费者工程，分别加入RabbitMQ java client的依赖。

itheima_rabbitmq_producer：生产者工程
在这里插入图片描述
itheima_rabbitmq_consumer：消费者工程

2. 分别在两个工程中添加依赖

<dependencies><!--rabbitmq java 客户端--><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.6.0</version></dependency>
</dependencies>

3. 编写生产者发送消息

在生产者工程中，编写消息生产者com.itheima.rabbitmq.simple.Producer_HelloWorld

package com.itheima.rabbitmq.simple;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 发送消息*/
public class Producer_HelloWorld {// 定义队列，如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建public static final String QUEUE_NAME = "hello_world";public static void main(String[] args) throws IOException, TimeoutException {// 1、创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2、设置参数// 主机地址：默认为 localhostconnectionFactory.setHost("192.168.198.128");// 连接端口：默认为 5672connectionFactory.setPort(5672);// 虚拟主机名称：默认为 /connectionFactory.setVirtualHost("/itcast");// 连接用户名：默认为 guestconnectionFactory.setUsername("heima");// 连接密码：默认为 guestconnectionFactory.setPassword("heima");// 3、创建连接Connection connection = connectionFactory.newConnection();// 4、创建 channelChannel channel = connection.createChannel();// 5、创建队列 queue/*queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)参数：1、queue：队列名称2、durable：是否持久化，当mq重启之后，还在3、exclusive：* 是否独占。只能有一个消费者监听这队列，* 当Connection关闭时，是否删除队列4、autoDelete：是否自动删除。当没有Consumer时，自动删除掉5、arguments：参数*/channel.queueDeclare(QUEUE_NAME, true, false, false, null);// 要发送的消息String body = "hello_world rabbitmq....";// 发送消息/*basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)参数：1、exchange：交换机名称，如果没有指定，简单模式下交换机会使用默认的 ""2、routingKey：路由key,简单模式可以传递队列名称3、props：配置信息4、body：发送消息数据*/channel.basicPublish("", QUEUE_NAME, null, body.getBytes());System.out.println("已发送消息：" + body);// 释放资源channel.close();connection.close();}
}

在执行上述的消息发送之后；可以登录rabbitMQ的管理控制台，可以发现队列和其消息：
在这里插入图片描述

4. 编写消费者接收消息

在消费者工程中，编写消息的消费者com.itheima.rabbitmq.simple.Consumer_HelloWorld

package com.itheima.rabbitmq.simple;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer_HelloWorld {public static void main(String[] args) throws IOException, TimeoutException {// 1、创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2、设置参数// 主机地址：默认为 localhostconnectionFactory.setHost("192.168.198.128");// 连接端口：默认为 5672connectionFactory.setPort(5672);// 虚拟主机名称：默认为 /connectionFactory.setVirtualHost("/itcast");// 连接用户名：默认为 guestconnectionFactory.setUsername("heima");// 连接密码：默认为 guestconnectionFactory.setPassword("heima");// 3、创建连接Connection connection = connectionFactory.newConnection();// 4、创建 channelChannel channel = connection.createChannel();// 5、创建队列/*queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)参数：1. queue：队列名称2. durable:是否持久化，当mq重启之后，还在3. exclusive：* 是否独占。只能有一个消费者监听这队列* 当Connection关闭时，是否删除队列4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉5. arguments：参数。*///如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建channel.queueDeclare("hello_world", true, false, false, null);// 6、接收消息DefaultConsumer consumer = new DefaultConsumer(channel) {/*回调方法，当收到消息后，会自动执行该方法参数：1、consumerTag：标识,消息者标签，在channel.basicConsume时候可以指定.2、envelope：获取一些信息，交换机，路由key...。消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)3、properties：配置信息4、body：数据*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("consumerTag：" + consumerTag);System.out.println("交换机为：" + envelope.getExchange());System.out.println("路由key为：" + envelope.getRoutingKey());System.out.println("消息id为：" + envelope.getDeliveryTag());System.out.println("properties：" + properties);System.out.println("接收到的消息：" + new String(body));}};/*basicConsume(String queue, boolean autoAck, Consumer callback)参数：1. queue：队列名称2. autoAck：是否自动确认。设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认3. callback：回调对象*/channel.basicConsume("hello_world", true, consumer);// 不关闭资源，应该一直监听消息// channel.close();// connection.close();}
}

小结

上述的入门案例中中其实使用的是如下的简单模式：
在这里插入图片描述
在上图的模型中，有以下概念：

P：生产者，也就是要发送消息的程序
C：消费者：消息的接收者，会一直等待消息到来
queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息

RabbitMQ工作模式

Work queues工作队列模式

模式说明

在这里插入图片描述
Work Queues：与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。

应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

代码编写

Work Queues 与入门程序的简单模式的代码几乎是一样的。可以完全复制，并多复制一个消费者进行多个消费者同时对消费消息的测试。

生产者代码

package com.itheima.rabbitmq.simple;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 发送多条消息*/
public class Producer_HelloWorld_1 {public static void main(String[] args) throws IOException, TimeoutException {// 1、创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2、设置参数// 主机地址：默认为 localhostconnectionFactory.setHost("192.168.198.128");// 连接端口：默认为 5672connectionFactory.setPort(5672);// 虚拟主机名称：默认为 /connectionFactory.setVirtualHost("/itcast");// 连接用户名：默认为 guestconnectionFactory.setUsername("heima");// 连接密码：默认为 guestconnectionFactory.setPassword("heima");// 3、创建连接Connection connection = connectionFactory.newConnection();// 4、创建 channelChannel channel = connection.createChannel();// 5、创建队列 queue/*queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)参数：1、queue：队列名称2、durable：是否持久化，当mq重启之后，还在3、exclusive：* 是否独占。只能有一个消费者监听这队列，* 当Connection关闭时，是否删除队列4、autoDelete：是否自动删除。当没有Consumer时，自动删除掉5、arguments：参数*/// 定义队列，如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建channel.queueDeclare("hello_world", true, false, false, null);// 发送消息/*basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)参数：1、exchange：交换机名称，如果没有指定，简单模式下交换机会使用默认的 ""2、routingKey：路由key,简单模式可以传递队列名称3、props：配置信息4、body：发送消息数据*/for (int i = 1; i < 10; i++) {// 要发送的消息String body = "hello_world rabbitmq...." + i;channel.basicPublish("", "hello_world", null, body.getBytes());System.out.println("已发送消息：" + body);}// 释放资源channel.close();connection.close();}
}

消费者1代码

package com.itheima.rabbitmq.simple;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer_HelloWorld_1 {public static void main(String[] args) throws IOException, TimeoutException {// 1、创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2、设置参数// 主机地址：默认为 localhostconnectionFactory.setHost("192.168.198.128");// 连接端口：默认为 5672connectionFactory.setPort(5672);// 虚拟主机名称：默认为 /connectionFactory.setVirtualHost("/itcast");// 连接用户名：默认为 guestconnectionFactory.setUsername("heima");// 连接密码：默认为 guestconnectionFactory.setPassword("heima");// 3、创建连接Connection connection = connectionFactory.newConnection();// 4、创建 channelChannel channel = connection.createChannel();// 一次只能接收并处理一个消息channel.basicQos(1);// 5、创建队列/*queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)参数：1. queue：队列名称2. durable:是否持久化，当mq重启之后，还在3. exclusive：* 是否独占。只能有一个消费者监听这队列* 当Connection关闭时，是否删除队列4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉5. arguments：参数。*///如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建channel.queueDeclare("hello_world", true, false, false, null);// 6、接收消息DefaultConsumer consumer = new DefaultConsumer(channel) {/*回调方法，当收到消息后，会自动执行该方法参数：1、consumerTag：标识,消息者标签，在channel.basicConsume时候可以指定.2、envelope：获取一些信息，交换机，路由key...。消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)3、properties：配置信息4、body：数据*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("consumerTag：" + consumerTag);System.out.println("交换机为：" + envelope.getExchange());System.out.println("路由key为：" + envelope.getRoutingKey());System.out.println("消息id为：" + envelope.getDeliveryTag());System.out.println("properties：" + properties);System.out.println("消费者一接收到的消息：" + new String(body));}};/*basicConsume(String queue, boolean autoAck, Consumer callback)参数：1. queue：队列名称2. autoAck：是否自动确认。设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认3. callback：回调对象*/channel.basicConsume("hello_world", true, consumer);// 不关闭资源，应该一直监听消息// channel.close();// connection.close();}
}

消费者2代码

package com.itheima.rabbitmq.simple;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer_HelloWorld_2 {public static void main(String[] args) throws IOException, TimeoutException {// 1、创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2、设置参数// 主机地址：默认为 localhostconnectionFactory.setHost("192.168.198.128");// 连接端口：默认为 5672connectionFactory.setPort(5672);// 虚拟主机名称：默认为 /connectionFactory.setVirtualHost("/itcast");// 连接用户名：默认为 guestconnectionFactory.setUsername("heima");// 连接密码：默认为 guestconnectionFactory.setPassword("heima");// 3、创建连接Connection connection = connectionFactory.newConnection();// 4、创建 channelChannel channel = connection.createChannel();// 一次只能接收并处理一个消息channel.basicQos(1);// 5、创建队列/*queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)参数：1. queue：队列名称2. durable:是否持久化，当mq重启之后，还在3. exclusive：* 是否独占。只能有一个消费者监听这队列* 当Connection关闭时，是否删除队列4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉5. arguments：参数。*///如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建channel.queueDeclare("hello_world", true, false, false, null);// 6、接收消息DefaultConsumer consumer = new DefaultConsumer(channel) {/*回调方法，当收到消息后，会自动执行该方法参数：1、consumerTag：标识,消息者标签，在channel.basicConsume时候可以指定.2、envelope：获取一些信息，交换机，路由key...。消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)3、properties：配置信息4、body：数据*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("consumerTag：" + consumerTag);System.out.println("交换机为：" + envelope.getExchange());System.out.println("路由key为：" + envelope.getRoutingKey());System.out.println("消息id为：" + envelope.getDeliveryTag());System.out.println("properties：" + properties);System.out.println("消费者二接收到的消息：" + new String(body));}};/*basicConsume(String queue, boolean autoAck, Consumer callback)参数：1. queue：队列名称2. autoAck：是否自动确认。设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认3. callback：回调对象*/channel.basicConsume("hello_world", true, consumer);// 不关闭资源，应该一直监听消息// channel.close();// connection.close();}
}

测试

启动两个消费者，然后再启动生产者发送消息；到IDEA的两个消费者对应的控制台查看是否竞争性的接收到消息。
在这里插入图片描述

小结

在一个队列中如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系。
Work Queues 对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。例如：短信服务部署多个，只需要有一个节点成功发送即可。

Publish/Subscribe发布与订阅模式

模式说明

在这里插入图片描述
前面2个案例中，只有3个角色：

P：生产者，也就是要发送消息的程序
C：消费者：消息的接受者，会一直等待消息到来。
queue：消息队列，图中红色部分

在订阅模型中，多了一个 Exchange 角色，而且过程略有变化：

P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
C：消费者，消息的接收者，会一直等待消息到来
Queue：消息队列，接收消息、缓存消息
Exchange：交换机（X）。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有常见以下3种类型：
- Fanout：广播，将消息交给所有绑定到交换机的队列
- Direct：定向，把消息交给符合指定routing key 的队列
- Topic：通配符，把消息交给符合routing pattern（路由模式）的队列

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与 Exchange 绑定，或者没有符合路由规则的队列，那么消息会丢失！

代码编写

生产者

package com.itheima.rabbitmq.ps;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 发布与订阅使用的交换机类型为：fanout*/
public class Producer_PubSub {//交换机名称public static final String FANOUT_EXCHAGE = "fanout_exchange";//队列名称public static final String FANOUT_QUEUE_1 = "fanout_queue_1";//队列名称public static final String FANOUT_QUEUE_2 = "fanout_queue_2";public static void main(String[] args) throws IOException, TimeoutException {// 1、创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2、设置参数// 主机地址：默认为 localhostconnectionFactory.setHost("192.168.198.128");// 连接端口：默认为 5672connectionFactory.setPort(5672);// 虚拟主机名称：默认为 /connectionFactory.setVirtualHost("/itcast");// 连接用户名：默认为 guestconnectionFactory.setUsername("heima");// 连接密码：默认为 guestconnectionFactory.setPassword("heima");// 3、创建连接Connection connection = connectionFactory.newConnection();// 4、创建 channelChannel channel = connection.createChannel();/*exchangeDeclare(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：内部使用。 一般false6、arguments：参数*/// 5、创建交换机channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);/*queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)参数：1、queue：队列名称2、durable：是否持久化，当mq重启之后，还在3、exclusive：* 是否独占。只能有一个消费者监听这队列，* 当Connection关闭时，是否删除队列4、autoDelete：是否自动删除。当没有Consumer时，自动删除掉5、arguments：参数*/channel.queueDeclare(FANOUT_QUEUE_1, true, false, false, null);channel.queueDeclare(FANOUT_QUEUE_2, true, false, false, null);// 7、绑定队列和交换机/*queueBind(String queue, String exchange, String routingKey)参数：1. queue：队列名称2. exchange：交换机名称3. routingKey：路由键，绑定规则如果交换机的类型为fanout ，routingKey设置为""*/channel.queueBind(FANOUT_QUEUE_1, FANOUT_EXCHAGE, "");channel.queueBind(FANOUT_QUEUE_2, FANOUT_EXCHAGE, "");// 8、发送消息/*basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)参数：1、exchange：交换机名称，如果没有指定，简单模式下交换机会使用默认的 ""2、routingKey：路由key,简单模式可以传递队列名称3、props：配置信息4、body：发送消息数据*/for (int i = 1; i <= 10; i++) {String body = "Hi rabbitmq...发布订阅模式--" + i;channel.basicPublish(FANOUT_EXCHAGE, "", null, body.getBytes());System.out.println("已发送消息：" + body);}// 9、释放资源channel.close();connection.close();}
}

消费者一

package com.itheima.rabbitmq.ps;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer_PubSub_1 {//交换机名称public static final String FANOUT_EXCHAGE = "fanout_exchange";//队列名称public static final String FANOUT_QUEUE_1 = "fanout_queue_1";//队列名称public static final String FANOUT_QUEUE_2 = "fanout_queue_2";public static void main(String[] args) throws IOException, TimeoutException {// 1、创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2、设置参数// 主机地址：默认为 localhostconnectionFactory.setHost("192.168.198.128");// 连接端口：默认为 5672connectionFactory.setPort(5672);// 虚拟主机名称：默认为 /connectionFactory.setVirtualHost("/itcast");// 连接用户名：默认为 guestconnectionFactory.setUsername("heima");// 连接密码：默认为 guestconnectionFactory.setPassword("heima");// 3、创建连接Connection connection = connectionFactory.newConnection();// 4、创建 channelChannel channel = connection.createChannel();// 声明交换机channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);// 声明（创建）队列/*** 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(FANOUT_QUEUE_1, true, false, false, null);// 队列绑定交换机channel.queueBind(FANOUT_QUEUE_1, FANOUT_EXCHAGE, "");// 接收消息Consumer consumer = new DefaultConsumer(channel) {/*回调方法，当收到消息后，会自动执行该方法1. consumerTag：标识2. envelope：获取一些信息，交换机，路由key...3. properties:配置信息4. body：数据*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// 路由keySystem.out.println("路由key为：" + envelope.getRoutingKey());// 交换机System.out.println("交换机为：" + envelope.getExchange());// 消息idSystem.out.println("消息id为：" + envelope.getDeliveryTag());// 收到的消息System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));}};// 监听消息/*** 参数1：队列名称* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认* 参数3：消息接收到后回调*/channel.basicConsume(FANOUT_QUEUE_1, true, consumer);// 关闭资源？ 不要}
}

消费者二

package com.itheima.rabbitmq.ps;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer_PubSub_2 {//交换机名称public static final String FANOUT_EXCHAGE = "fanout_exchange";//队列名称public static final String FANOUT_QUEUE_1 = "fanout_queue_1";//队列名称public static final String FANOUT_QUEUE_2 = "fanout_queue_2";public static void main(String[] args) throws IOException, TimeoutException {// 1、创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2、设置参数// 主机地址：默认为 localhostconnectionFactory.setHost("192.168.198.128");// 连接端口：默认为 5672connectionFactory.setPort(5672);// 虚拟主机名称：默认为 /connectionFactory.setVirtualHost("/itcast");// 连接用户名：默认为 guestconnectionFactory.setUsername("heima");// 连接密码：默认为 guestconnectionFactory.setPassword("heima");// 3、创建连接Connection connection = connectionFactory.newConnection();// 4、创建 channelChannel channel = connection.createChannel();// 声明交换机channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);// 声明（创建）队列/*** 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(FANOUT_QUEUE_2, true, false, false, null);// 队列绑定交换机channel.queueBind(FANOUT_QUEUE_2, FANOUT_EXCHAGE, "");// 接收消息Consumer consumer = new DefaultConsumer(channel) {/*回调方法，当收到消息后，会自动执行该方法1. consumerTag：标识2. envelope：获取一些信息，交换机，路由key...3. properties:配置信息4. body：数据*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// 路由keySystem.out.println("路由key为：" + envelope.getRoutingKey());// 交换机System.out.println("交换机为：" + envelope.getExchange());// 消息idSystem.out.println("消息id为：" + envelope.getDeliveryTag());// 收到的消息System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8"));}};// 监听消息/*** 参数1：队列名称* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认* 参数3：消息接收到后回调*/channel.basicConsume(FANOUT_QUEUE_2, true, consumer);// 关闭资源？ 不要}
}

测试

启动所有消费者，然后使用生产者发送消息；在每个消费者对应的控制台可以查看到生产者发送的所有消息；到达广播的效果。

在执行完测试代码后，其实到RabbitMQ的管理后台找到Exchanges选项卡，点击 fanout_exchange 的交换机，可以查看到如下的绑定：
在这里插入图片描述

小结

交换机需要与队列进行绑定，绑定之后；一个消息可以被多个消费者都收到。

发布订阅模式与工作队列模式的区别：

工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机
发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息(底层使用默认交换机)
发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑定到默认的交换机

Routing路由模式

模式说明

队列与交换机的绑定，不能是任意绑定了，而是要指定一个 RoutingKey（路由key）
消息的发送方在向 Exchange 发送消息时，也必须指定消息的 RoutingKey
Exchange 不再把消息交给每一个绑定的队列，而是根据消息的 Routing Key 进行判断，只有队列的Routingkey 与消息的 Routing key 完全一致，才会接收到消息

在这里插入图片描述
图解：

P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
X：Exchange（交换机），接收生产者的消息，然后把消息递交给与routing key完全匹配的队列
C1：消费者，其所在队列指定了需要routing key 为 error 的消息
C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

代码

在编码上与 Publish/Subscribe发布与订阅模式 的区别是交换机的类型为：Direct，还有队列绑定交换机的时候需要指定routing key。

生产者

package com.itheima.rabbitmq.routing;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 路由模式的交换机类型为：direct*/
public class Producer_Routing {//交换机名称public static final String DIRECT_EXCHAGE = "direct_exchange";//队列名称public static final String DIRECT_QUEUE_INSERT = "direct_queue_insert";//队列名称public static final String DIRECT_QUEUE_UPDATE = "direct_queue_update";public static void main(String[] args) throws IOException, TimeoutException {// 1、创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2、设置参数// 主机地址：默认为 localhostconnectionFactory.setHost("192.168.198.128");// 连接端口：默认为 5672connectionFactory.setPort(5672);// 虚拟主机名称：默认为 /connectionFactory.setVirtualHost("/itcast");// 连接用户名：默认为 guestconnectionFactory.setUsername("heima");// 连接密码：默认为 guestconnectionFactory.setPassword("heima");// 3、创建连接Connection connection = connectionFactory.newConnection();// 4、创建 channelChannel channel = connection.createChannel();/*exchangeDeclare(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：内部使用。 一般false6. arguments：参数*/// 5、创建交换机channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);/**参数1：队列名称参数2：是否定义持久化队列参数3：是否独占本次连接参数4：是否在不使用的时候自动删除队列参数5：队列其它参数*/// 6、创建队列channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null);channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null);/*queueBind(String queue, String exchange, String routingKey)参数：1. queue：队列名称2. exchange：交换机名称3. routingKey：路由键，绑定规则如果交换机的类型为fanout ，routingKey设置为""*/// 7、绑定队列和交换机channel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHAGE, "insert");channel.queueBind(DIRECT_QUEUE_UPDATE, DIRECT_EXCHAGE, "update");// 8、发送消息String body = "新增了商品。路由模式；routing key 为 insert ";/**参数1：交换机名称，如果没有指定则使用默认Default Exchage参数2：路由key,简单模式可以传递队列名称参数3：消息其它属性参数4：消息内容*/channel.basicPublish(DIRECT_EXCHAGE, "insert", null, body.getBytes());System.out.println("已发送消息：" + body);// 发送消息body = "修改了商品。路由模式；routing key 为 update";/**参数1：交换机名称，如果没有指定则使用默认Default Exchage参数2：路由key,简单模式可以传递队列名称参数3：消息其它属性参数4：消息内容*/channel.basicPublish(DIRECT_EXCHAGE, "update", null, body.getBytes());System.out.println("已发送消息：" + body);// 9、 释放资源channel.close();connection.close();}
}

消费者1

package com.itheima.rabbitmq.routing;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer_Routing_1 {//交换机名称public static final String DIRECT_EXCHAGE = "direct_exchange";//队列名称public static final String DIRECT_QUEUE_INSERT = "direct_queue_insert";public static void main(String[] args) throws IOException, TimeoutException {// 1、创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2、设置参数// 主机地址：默认为 localhostconnectionFactory.setHost("192.168.198.128");// 连接端口：默认为 5672connectionFactory.setPort(5672);// 虚拟主机名称：默认为 /connectionFactory.setVirtualHost("/itcast");// 连接用户名：默认为 guestconnectionFactory.setUsername("heima");// 连接密码：默认为 guestconnectionFactory.setPassword("heima");// 3、创建连接Connection connection = connectionFactory.newConnection();// 4、创建 channelChannel channel = connection.createChannel();// 声明交换机channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);/**参数1：队列名称参数2：是否定义持久化队列参数3：是否独占本次连接参数4：是否在不使用的时候自动删除队列参数5：队列其它参数*/// 6、创建队列channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null);// 队列绑定交换机channel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHAGE, "insert");// 接收消息Consumer consumer = new DefaultConsumer(channel) {/*回调方法，当收到消息后，会自动执行该方法1. consumerTag：标识2. envelope：获取一些信息，交换机，路由key...3. properties:配置信息4. body：数据*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// 路由keySystem.out.println("路由key为：" + envelope.getRoutingKey());// 交换机System.out.println("交换机为：" + envelope.getExchange());// 消息idSystem.out.println("消息id为：" + envelope.getDeliveryTag());// 收到的消息System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));}};/*** 参数1：队列名称* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认* 参数3：消息接收到后回调*/channel.basicConsume(DIRECT_QUEUE_INSERT, true, consumer);}
}

消费者2

package com.itheima.rabbitmq.routing;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer_Routing_2 {//交换机名称public static final String DIRECT_EXCHAGE = "direct_exchange";//队列名称public static final String DIRECT_QUEUE_UPDATE = "direct_queue_update";public static void main(String[] args) throws IOException, TimeoutException {// 1、创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2、设置参数// 主机地址：默认为 localhostconnectionFactory.setHost("192.168.198.128");// 连接端口：默认为 5672connectionFactory.setPort(5672);// 虚拟主机名称：默认为 /connectionFactory.setVirtualHost("/itcast");// 连接用户名：默认为 guestconnectionFactory.setUsername("heima");// 连接密码：默认为 guestconnectionFactory.setPassword("heima");// 3、创建连接Connection connection = connectionFactory.newConnection();// 4、创建 channelChannel channel = connection.createChannel();// 声明交换机channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);/**参数1：队列名称参数2：是否定义持久化队列参数3：是否独占本次连接参数4：是否在不使用的时候自动删除队列参数5：队列其它参数*/// 6、创建队列channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null);// 队列绑定交换机channel.queueBind(DIRECT_QUEUE_UPDATE, DIRECT_EXCHAGE, "update");// 接收消息Consumer consumer = new DefaultConsumer(channel) {/*回调方法，当收到消息后，会自动执行该方法1. consumerTag：标识2. envelope：获取一些信息，交换机，路由key...3. properties:配置信息4. body：数据*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// 路由keySystem.out.println("路由key为：" + envelope.getRoutingKey());// 交换机System.out.println("交换机为：" + envelope.getExchange());// 消息idSystem.out.println("消息id为：" + envelope.getDeliveryTag());// 收到的消息System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8"));}};/*** 参数1：队列名称* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认* 参数3：消息接收到后回调*/channel.basicConsume(DIRECT_QUEUE_UPDATE, true, consumer);}
}

测试

启动所有消费者，然后使用生产者发送消息；在消费者对应的控制台可以查看到生产者发送对应routing key对应队列的消息；到达按照需要接收的效果。

在执行完测试代码后，其实到RabbitMQ的管理后台找到Exchanges选项卡，点击 direct_exchange 的交换机，可以查看到如下的绑定：
在这里插入图片描述

小结

Routing模式要求队列在绑定交换机时要指定routing key，消息会转发到符合routing key的队列。

Topics通配符模式

模式说明

Topic类型与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！

Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

通配符规则：

#：匹配一个或多个词
*：匹配不多不少恰好1个词

举例：

item.#：能够匹配item.insert.abc 或者 item.insert
item.*：只能匹配item.insert

在这里插入图片描述

图解：

红色Queue：绑定的是usa.# ，因此凡是以 usa.开头的routing key 都会被匹配到
黄色Queue：绑定的是#.news ，因此凡是以 .news结尾的 routing key 都会被匹配

代码

生产者

使用topic类型的Exchange，发送消息的routing key有3种： item.insert、item.update、item.delete：

package com.itheima.rabbitmq.topic;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 通配符Topic的交换机类型为：topic*/
public class Producer_Topics {//交换机名称public static final String TOPIC_EXCHAGE = "topic_exchange";//队列名称public static final String TOPIC_QUEUE_1 = "topic_queue_1";//队列名称public static final String TOPIC_QUEUE_2 = "topic_queue_2";public static void main(String[] args) throws IOException, TimeoutException {// 1、创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2、设置参数// 主机地址：默认为 localhostconnectionFactory.setHost("192.168.198.128");// 连接端口：默认为 5672connectionFactory.setPort(5672);// 虚拟主机名称：默认为 /connectionFactory.setVirtualHost("/itcast");// 连接用户名：默认为 guestconnectionFactory.setUsername("heima");// 连接密码：默认为 guestconnectionFactory.setPassword("heima");// 3、创建连接Connection connection = connectionFactory.newConnection();// 4、创建 channelChannel channel = connection.createChannel();/*exchangeDeclare(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：内部使用。 一般false6. arguments：参数*/// 5、创建交换机channel.exchangeDeclare(TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);// 发送消息String body = "新增了商品。Topic模式；routing key 为 item.insert";channel.basicPublish(TOPIC_EXCHAGE, "item.insert", null, body.getBytes());System.out.println("已发送消息：" + body);// 发送消息body = "修改了商品。Topic模式；routing key 为 item.update";channel.basicPublish(TOPIC_EXCHAGE, "item.update", null, body.getBytes());System.out.println("已发送消息：" + body);// 发送消息body = "删除了商品。Topic模式；routing key 为 item.delete";channel.basicPublish(TOPIC_EXCHAGE, "item.delete", null, body.getBytes());System.out.println("已发送消息：" + body);// 关闭资源channel.close();connection.close();}
}

消费者1

接收两种类型的消息：更新商品和删除商品

package com.itheima.rabbitmq.topic;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer_Topic_1 {//交换机名称public static final String TOPIC_EXCHAGE = "topic_exchange";//队列名称public static final String TOPIC_QUEUE_1 = "topic_queue_1";//队列名称public static final String TOPIC_QUEUE_2 = "topic_queue_2";public static void main(String[] args) throws IOException, TimeoutException {// 1、创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2、设置参数// 主机地址：默认为 localhostconnectionFactory.setHost("192.168.198.128");// 连接端口：默认为 5672connectionFactory.setPort(5672);// 虚拟主机名称：默认为 /connectionFactory.setVirtualHost("/itcast");// 连接用户名：默认为 guestconnectionFactory.setUsername("heima");// 连接密码：默认为 guestconnectionFactory.setPassword("heima");// 3、创建连接Connection connection = connectionFactory.newConnection();// 4、创建 channelChannel channel = connection.createChannel();//声明交换机channel.exchangeDeclare(TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);// 声明（创建）队列/*** 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(TOPIC_QUEUE_1, true, false, false, null);// 队列绑定交换机channel.queueBind(TOPIC_QUEUE_1, TOPIC_EXCHAGE, "item.update");channel.queueBind(TOPIC_QUEUE_1, TOPIC_EXCHAGE, "item.delete");/*basicConsume(String queue, boolean autoAck, Consumer callback)参数：1. queue：队列名称2. autoAck：是否自动确认3. callback：回调对象*/// 接收消息Consumer consumer = new DefaultConsumer(channel) {/*回调方法，当收到消息后，会自动执行该方法1. consumerTag：标识2. envelope：获取一些信息，交换机，路由key...3. properties:配置信息4. body：数据*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// 路由keySystem.out.println("路由key为：" + envelope.getRoutingKey());// 交换机System.out.println("交换机为：" + envelope.getExchange());// 消息idSystem.out.println("消息id为：" + envelope.getDeliveryTag());// 收到的消息System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));}};//监听消息/*** 参数1：队列名称* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认* 参数3：消息接收到后回调*/channel.basicConsume(TOPIC_QUEUE_1, true, consumer);//关闭资源？不要}
}

消费者2

接收所有类型的消息：新增商品，更新商品和删除商品。

package com.itheima.rabbitmq.topic;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer_Topic_2 {//交换机名称public static final String TOPIC_EXCHAGE = "topic_exchange";//队列名称public static final String TOPIC_QUEUE_1 = "topic_queue_1";//队列名称public static final String TOPIC_QUEUE_2 = "topic_queue_2";public static void main(String[] args) throws IOException, TimeoutException {// 1、创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2、设置参数// 主机地址：默认为 localhostconnectionFactory.setHost("192.168.198.128");// 连接端口：默认为 5672connectionFactory.setPort(5672);// 虚拟主机名称：默认为 /connectionFactory.setVirtualHost("/itcast");// 连接用户名：默认为 guestconnectionFactory.setUsername("heima");// 连接密码：默认为 guestconnectionFactory.setPassword("heima");// 3、创建连接Connection connection = connectionFactory.newConnection();// 4、创建 channelChannel channel = connection.createChannel();//声明交换机channel.exchangeDeclare(TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);// 声明（创建）队列/*** 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(TOPIC_QUEUE_2, true, false, false, null);// 队列绑定交换机channel.queueBind(TOPIC_QUEUE_2, TOPIC_EXCHAGE, "item.*");/*basicConsume(String queue, boolean autoAck, Consumer callback)参数：1. queue：队列名称2. autoAck：是否自动确认3. callback：回调对象*/// 接收消息Consumer consumer = new DefaultConsumer(channel) {/*回调方法，当收到消息后，会自动执行该方法1. consumerTag：标识2. envelope：获取一些信息，交换机，路由key...3. properties:配置信息4. body：数据*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// 路由keySystem.out.println("路由key为：" + envelope.getRoutingKey());// 交换机System.out.println("交换机为：" + envelope.getExchange());// 消息idSystem.out.println("消息id为：" + envelope.getDeliveryTag());// 收到的消息System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8"));}};//监听消息/*** 参数1：队列名称* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认* 参数3：消息接收到后回调*/channel.basicConsume(TOPIC_QUEUE_2, true, consumer);//关闭资源？不要}
}

测试

启动所有消费者，然后使用生产者发送消息；在消费者对应的控制台可以查看到生产者发送对应routing key对应队列的消息；到达按照需要接收的效果；并且这些routing key可以使用通配符。

在执行完测试代码后，其实到RabbitMQ的管理后台找到Exchanges选项卡，点击 topic_exchange 的交换机，可以查看到如下的绑定：
在这里插入图片描述

小結

Topic主题模式可以实现 Publish/Subscribe发布与订阅模式 和 Routing路由模式 的功能；只是Topic在配置routing key 的时候可以使用通配符，显得更加灵活。

模式总结

1、简单模式 HelloWorld

一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）

2、工作队列模式 Work Queue

一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）

3、发布订阅模式 Publish/Subscribe

需要设置类型为fanout的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列

4、路由模式 Routing

需要设置类型为direct的交换机，交换机和队列进行绑定，并且指定routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

5、通配符模式 Topic

需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

Spring 整合RabbitMQ

搭建生产者工程

创建工程

生产者工程：itheima_spring_rabbitmq_producer

添加依赖

修改pom.xml文件内容为如下：

<dependencies><dependency><groupId>org.springframework</groupId><artifactId>spring-context</artifactId><version>5.1.7.RELEASE</version></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><version>2.1.8.RELEASE</version></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.12</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-test</artifactId><version>5.1.7.RELEASE</version></dependency>
</dependencies>

配置整合

1、创建rabbitmq.properties连接参数等配置文件；

rabbitmq.host=192.168.198.128
rabbitmq.port=5672
rabbitmq.username=heima
rabbitmq.password=heima
rabbitmq.virtual-host=/itcast

2、创建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: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，名字为：""，路由键为队列的名称--><!--id：bean的名称name：queue的名称auto-declare:自动创建auto-delete:自动删除。 最后一个消费者和该队列断开连接后，自动删除队列exclusive:是否独占durable：是否持久化--><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:direct-exchange name="aa" ><rabbit:bindings>&lt;!&ndash;direct 类型的交换机绑定队列  key ：路由key  queue：队列名称&ndash;&gt;<rabbit:binding queue="spring_queue" key="xxx"></rabbit:binding></rabbit:bindings></rabbit:direct-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>

发送消息

在test\java\com\itheima\test\ProducerTest.java 创建测试文件

package com.itheima.test;import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
public class ProducerTest {// 注入RabbitTemplateprivate RabbitTemplate rabbitTemplate;/*** 只发队列消息* 默认交换机类型为 direct* 交换机的名称为空，路由键为队列的名称*/@Testpublic void testQueue() {// 发送消息，路由键与队列同名rabbitTemplate.convertAndSend("spring_queue", "只发队列spring_queue的消息。");}/*** 发送广播* 交换机类型为 fanout* 绑定到该交换机的所有队列都能够收到消息*/@Testpublic void testFanout() {/*** 参数1：交换机名称* 参数2：路由键名（广播设置为空）* 参数3：发送的消息内容*/rabbitTemplate.convertAndSend("spring_fanout_exchange", "", "发送到spring_fanout_exchange交换机的广播消息");}/*** 通配符* 交换机类型为 topic* 匹配路由键的通配符，*表示一个单词，#表示多个单词* 绑定到该交换机的匹配队列能够收到对应消息*/@Testpublic void testTopic() {/*** 参数1：交换机名称* 参数2：路由键名* 参数3：发送的消息内容*/rabbitTemplate.convertAndSend("spring_topic_exchange", "heima.bj", "发送到spring_topic_exchange交换机heima.bj的消息");rabbitTemplate.convertAndSend("spring_topic_exchange", "heima.bj.1", "发送到spring_topic_exchange交换机heima.bj.1的消息");rabbitTemplate.convertAndSend("spring_topic_exchange", "heima.bj.2", "发送到spring_topic_exchange交换机heima.bj.2的消息");rabbitTemplate.convertAndSend("spring_topic_exchange", "itcast.cn", "发送到spring_topic_exchange交换机itcast.cn的消息");}
}

搭建消费者者工程

创建工程

消费者工程：itheima_spring_rabbitmq_consumer

添加依赖

修改pom.xml文件内容为如下：

<dependencies><dependency><groupId>org.springframework</groupId><artifactId>spring-context</artifactId><version>5.1.7.RELEASE</version></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><version>2.1.8.RELEASE</version></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.12</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-test</artifactId><version>5.1.7.RELEASE</version></dependency>
</dependencies>

配置整合

1、创建rabbitmq.properties连接参数等配置文件；

rabbitmq.host=192.168.198.128
rabbitmq.port=5672
rabbitmq.username=heima
rabbitmq.password=heima
rabbitmq.virtual-host=/itcast

2、创建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: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>

消息监听器

队列监听器

创建com.itheima.rabbitmq.listener.SpringQueueListener.java

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) {try {String msg = new String(message.getBody(), "utf-8");System.out.printf("接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",message.getMessageProperties().getReceivedExchange(),message.getMessageProperties().getReceivedRoutingKey(),message.getMessageProperties().getConsumerQueue(),msg);} catch (Exception e) {e.printStackTrace();}}
}

广播监听器1

创建com.itheima.rabbitmq.listener.FanoutListener1.java

package com.itheima.rabbitmq.listener;import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;public class FanoutListener1 implements MessageListener {@Overridepublic void onMessage(Message message) {try {String msg = new String(message.getBody(), "utf-8");System.out.printf("广播监听器1：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",message.getMessageProperties().getReceivedExchange(),message.getMessageProperties().getReceivedRoutingKey(),message.getMessageProperties().getConsumerQueue(),msg);} catch (Exception e) {e.printStackTrace();}}
}

广播监听器2

创建com.itheima.rabbitmq.listener.FanoutListener2.java

package com.itheima.rabbitmq.listener;import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;public class FanoutListener2 implements MessageListener {@Overridepublic void onMessage(Message message) {try {String msg = new String(message.getBody(), "utf-8");System.out.printf("广播监听器2：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",message.getMessageProperties().getReceivedExchange(),message.getMessageProperties().getReceivedRoutingKey(),message.getMessageProperties().getConsumerQueue(),msg);} catch (Exception e) {e.printStackTrace();}}
}

星号通配符监听器

创建com.itheima.rabbitmq.listener.TopicListenerStar.java

package com.itheima.rabbitmq.listener;import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;public class TopicListenerStar implements MessageListener {@Overridepublic void onMessage(Message message) {try {String msg = new String(message.getBody(), "utf-8");System.out.printf("通配符*监听器：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",message.getMessageProperties().getReceivedExchange(),message.getMessageProperties().getReceivedRoutingKey(),message.getMessageProperties().getConsumerQueue(),msg);} catch (Exception e) {e.printStackTrace();}}
}

井号通配符监听器

创建com.itheima.rabbitmq.listener.TopicListenerWell.java

package com.itheima.rabbitmq.listener;import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;public class TopicListenerWell implements MessageListener {@Overridepublic void onMessage(Message message) {try {String msg = new String(message.getBody(), "utf-8");System.out.printf("通配符#监听器：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",message.getMessageProperties().getReceivedExchange(),message.getMessageProperties().getReceivedRoutingKey(),message.getMessageProperties().getConsumerQueue(),msg);} catch (Exception e) {e.printStackTrace();}}
}

井号通配符监听器2

创建com.itheima.rabbitmq.listener.TopicListenerWell2.java

package com.itheima.rabbitmq.listener;import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;public class TopicListenerWell2 implements MessageListener {@Overridepublic void onMessage(Message message) {try {String msg = new String(message.getBody(), "utf-8");System.out.printf("通配符#监听器2：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",message.getMessageProperties().getReceivedExchange(),message.getMessageProperties().getReceivedRoutingKey(),message.getMessageProperties().getConsumerQueue(),msg);} catch (Exception e) {e.printStackTrace();}}
}

小结

使用 Spring 整合 RabbitMQ 将组件全部使用配置方式实现，简化编码
Spring 提供 RabbitTemplate 简化发送消息 API
使用监听机制简化消费者编码

Spring Boot整合RabbitMQ

简介

在Spring项目中，可以使用Spring-Rabbit去操作RabbitMQ，https://github.com/spring-projects/spring-amqp

尤其是在spring boot项目中只需要引入对应的amqp启动器依赖即可，方便的使用RabbitTemplate发送消息，使用注解接收消息。

一般在开发过程中：

生产者工程：

application.yml文件配置RabbitMQ相关信息；
在生产者工程中编写配置类，用于创建交换机和队列，并进行绑定
注入RabbitTemplate对象，通过RabbitTemplate对象发送消息到交换机

消费者工程：

application.yml文件配置RabbitMQ相关信息
创建消息处理类，用于接收队列中的消息并进行处理

搭建生产者工程

创建工程

创建生产者工程springboot_rabbitmq_producer

添加依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.itheima</groupId><artifactId>springboot_rabbitmq_consumer</artifactId><version>1.0-SNAPSHOT</version><!-- 1. 父工程依赖 --><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.1.4.RELEASE</version></parent><dependencies><!-- 2. rabbitmq--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId></dependency></dependencies></project>

启动类

在 com.itheima.rabbitmq包下创建启动类ProducerApplication

package com.itheima.rabbitmq;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class ProducerApplication {public static void main(String[] args) {SpringApplication.run(ProducerApplication.class);}
}

配置RabbitMQ

（1）配置文件

创建application.yml，内容如下：

# 配置RabbitMQ的基本信息  ip 端口 username  password..
spring:rabbitmq:host: 192.168.198.128 # ipport: 5672username: heimapassword: heimavirtual-host: /itcast

（2）绑定交换机和队列

创建RabbitMQ队列与交换机绑定的配置类com.itheima.rabbitmq.config.RabbitMQConfig

package com.itheima.rabbitmq.config;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitMQConfig {// 交换机名称public static final String ITEM_TOPIC_EXCHANGE = "item_topic_exchange";// 队列名称public static final String ITEM_QUEUE = "item_queue";// 声明交换机@Bean("topicExchange")public Exchange topicExchange() {return ExchangeBuilder.topicExchange(ITEM_TOPIC_EXCHANGE).durable(true).build();}// 声明队列@Bean("itemQueue")public Queue itemQueue() {return QueueBuilder.durable(ITEM_QUEUE).build();}// 3. 队列和交互机绑定关系 Binding/*1. 知道哪个队列2. 知道哪个交换机3. routing key*/@Beanpublic Binding itemQueueExchange(@Qualifier("itemQueue") Queue queue,@Qualifier("topicExchange") Exchange exchange) {return BindingBuilder.bind(queue).to(exchange).with("item.#").noargs();}
}

搭建消费者

创建工程

创建消费者工程springboot_rabbitmq_consumer

添加依赖

修改pom.xml文件内容为如下：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.itheima</groupId><artifactId>springboot_rabbitmq_consumer</artifactId><version>1.0-SNAPSHOT</version><!-- 1. 父工程依赖 --><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.1.4.RELEASE</version></parent><dependencies><!-- 2. rabbitmq--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId></dependency></dependencies></project>

启动类

package com.itheima.rabbitmq;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class ConsumerApplication {public static void main(String[] args) {SpringApplication.run(ConsumerApplication.class, args);}
}

配置RabbitMQ

创建application.yml，内容如下：

spring:rabbitmq:host: 192.168.198.128 #主机ipport: 5672 #端口username: heimapassword: heimavirtual-host: /itcast

消息监听处理类

编写消息监听器com.itheima.rabbitmq.listener.MyListener

package com.itheima.rabbitmq.listener;import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class MyListener {/*** 监听某个队列的消息** @param message 接收到的消息*/@RabbitListener(queues = "item_queue")public void myListener1(String message) {System.out.println("消费者接收到的消息为：" + message);}/*@RabbitListener(queues = "boot_queue")public void ListenerQueue(Message message){//System.out.println(message);System.out.println(new String(message.getBody()));}*/
}

测试

在生产者工程springboot-rabbitmq-producer中创建测试类，发送消息：

package com.itheima.rabbitmq;import com.itheima.rabbitmq.config.RabbitMQConfig;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
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 RabbitMQTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void test() {rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE, "item.insert", "商品新增，routing key 为item.insert");rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE, "item.update", "商品修改，routing key 为item.update");rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE, "item.delete", "商品删除，routing key 为item.delete");}
}

先运行上述测试程序（交换机和队列才能先被声明和绑定），然后启动消费者；在消费者工程springboot-rabbitmq-consumer中控制台查看是否接收到对应消息。

另外；也可以在RabbitMQ的管理控制台中查看到交换机与队列的绑定：
在这里插入图片描述

小结

SpringBoot提供了快速整合RabbitMQ的方式
基本信息再yml中配置，队列交互机以及绑定关系在配置类中使用Bean的方式配置
生产端直接注入RabbitTemplate完成消息发送
消费端直接使用@RabbitListener完成消息接收

代码仓库

查看全文
如若内容造成侵权/违法违规/事实不符，请联系编程学习网邮箱：809451989@qq.com进行投诉反馈，一经查实，立即删除！

渗透基础篇
服务器:就和我们的个人电脑差不多，也是一台主机，只不过是大型主机而且配置很高，一般就像网吧机房一样，一个冰箱高的柜子一样的有几层，里面放着主机虚拟机：就是一台服务器分割成多个服务器就这样理解，是一台主机虚拟分化出来的可以当正常主机使用VPS：就是更高级一点的虚…...
2024/4/18 21:18:36
idea编译改成eclipse后报错
idea编译改成eclipse后报错如下：Error:java: Internal error: java.lang.IllegalArgumentExceptionat org.eclipse.jdt.internal.compiler.tool.EclipseFileManager.handleOption(EclipseFileManager.java:758)at java.compiler@11.0.4/javax.tools.ForwardingJavaFileManage…...
2024/4/17 23:59:31
idea编译时报错“常量字符串过长”
依次点击 File >> Settings >> Build,Execution,Deployment >> Compiler >>Java Compiler 将Use compiler改为eclipse如果改完后build报错Error:java: Internal error: java.lang.IllegalArgumentExceptionat org.eclipse.jdt.internal.compiler.tool…...
2024/4/16 9:13:51
windows下一个好用的exel表格操作库c++
这个库是libxl,可以去官网下载，也可以用破解（仅供学习，不可用于商业，如用官方产生法律纠纷，概不负责）我这边是动态库libxl.dll 和libxl.lib 还有include 因为是动态库所以不管vs20xx还是debug/release/win32/x64都可以用，具体查看我的下载管理《libxl综合文件》或者…...
2024/4/16 9:14:22
STM32L475裸机例程学习红外接收实验
红外遥控实验红外遥控简介红外线遥控不具有像无线电遥控那样穿过障碍物去控制被控对象的能力，所以，在设计红外线遥控器时，不必要像无线电遥控器那样，每套(发射器和接收器)要有不同的遥控频率或编码(否则，就会隔墙控制或干扰邻居的家用电器)，所以同类产品的红外线遥控器…...
2024/4/16 9:13:26
吐血精简总结之操作系统
前言即将参加今年的秋招，校招比较考核学生的底层知识，开始撸操作系统、计算机组成原理、计算机网络以及Java的知识了。希望能在2020年本就不怎么友好的一年收获理想的Offer。目录前言操作系统一、基础1. 操作系统的定义（小鹏）2. 操作系统的特征特征一、并发并发和并行的区…...
2024/4/17 12:37:29
BUAA-OO-2019 第三单元总结
JML语言理论基础梳理及工具链注释结构 JML以javadoc注释的方式来表示规格，每行都以@起头。行注释：//@annotation 块注释：/* @ annotation @*/ JML表达式 JML的表达式是对Java表达式的扩展，新增了一些操作符和原子表达式。原子表达式 \result表达式：表示一个非 void 类型…...
2024/4/16 9:13:36
稳定性三十六计-幂等设计
引子群里发了一个总共1千元的拼手气红包，共10个。静儿点进去，额，抢到了0.05元。这个不甘心啊。退出来重新打开了这个红包，你猜怎样？显示我抢到了0.05元！这就是幂等(idempotence)，不管多少次请求某一个资源，对资源都具有相同的影响。幂等性是系统的接口对外一种承诺，承…...
2024/4/16 9:14:32
为什么APP业务更加容易受到攻击
如今，随着互联网的飞速发展，网络攻击日益频繁，很多企业都遭受到了不同程度的网络攻击，尤其是难以防御的DDoS攻击。APP也是最容易遭受DDoS攻击的行业之一。那么为什么APP会被DDOS攻击呢？1、APP基本都是含有盈利性，某些具有商城交易等性质的APP，如果经营状况很好的情况下，…...
2024/4/16 9:14:32
嵌入式设备交叉编译perf性能分析工具
嵌入式设备交叉编译perf性能分析工具 1.1 背景最近工作一直在做嵌入式相关的开发，主要是应用方面，随着程序的业务功能越写越复杂，加上嵌入式系统上的调试工具，少之又少，主要还是靠printf的传统调试大法，下面来整个perf分析程序的热点函数。 1.2 make menuconfig 进入内核…...
2024/4/16 9:14:12
在大型项目上，Python 是个烂语言吗？
作者：布丁链接：https://www.zhihu.com/question/21017354/answer/652602653 来源：知乎著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。前YouTuber来讲讲YouTube的故事吧。话说我们YouTuber跟Instagram的朋友吃饭，都要事先约定不谈Python，不然You…...
2024/4/13 1:03:10
不看一下TOP20的云排名，你都不好意思说自己懂云
不看一下TOP20的云排名，你都不好意思说自己懂云《2019年中国公有云厂商发展状况白皮书》第二部分2019年中国公有云厂商整体发展状况概述既然TOP5排名、TOP10排名出现了新状况，那么2019年中国公有云厂商综合收入排名TOP20将会呈现出怎样的变化呢？综合收入排名TOP20云厂商具体…...
2024/4/17 21:51:05
#Redis# python 连接redis
Python 要使用 redis，需要先安装 redis 模块：1、安装redis库pip3 install redis2、测试是否安装成功：(base) chenjd:~ xqdd$ pythonPython 3.7.1 (default, Dec 14 2018, 13:28:58)[Clang 4.0.1 (tags/RELEASE_401/final)] :: Anaconda, Inc. on darwinType "help"…...
2024/4/19 21:10:29
OSPF 详解
OSPF关系如何建立？使用hello报文建立邻居关系，ospf协议直接封装在IP报文中，协议号是89，证明IPV4和ospf协议直接存在相辅相成的关系，ospf是一种运行在ipv4之上的路由协议ospf协议报文头部字段，每种OSPF报文都包含。1、version v2 描述ospf协议版本2、message type：描述OS…...
2024/4/13 10:43:59
JS获取浏览器localStorage存储上限
基本思路：不断的给字符串增加字符，调用 localStorage.setItem 进行存储，直至其无法存储为止（exceeded the quota：超出限额）。注意事项：给定的字符串不能过小，否则 js 代码执行非常慢，也不能过大，否则会造成较大的误差。案例中误差控制在 1kb 以内。先按照 128kb 进…...
2024/4/4 20:49:25
js 递归函数之 arguments.callee
传统递归阶乘函数： function factorial(num){if (num <= 1){return 1;} else {return num * factorial(num-1); } }虽然这个函数表面看来没什么问题，但下面的代码却可能导致它出错。 var anotherFactorial = factorial; factorial = null; alert(anotherFactorial(4)); /…...
2024/4/16 9:13:46
Python MySQL
Python MySQL - mysql-connector 驱动 MySQL 是最流行的关系型数据库管理系统，如果你不熟悉 MySQL，可以阅读我们的 MySQL 教程。本章节我们为大家介绍使用 mysql-connector 来连接使用 MySQL， mysql-connector 是 MySQL 官方提供的驱动器。我们可以使用 pip 命令来安装 my…...
2024/4/16 9:13:21
代码优化之字符串拼接“+“和StringBulider
经常在代码中会涉及到字符串得拼接，我们自然而然得使用“+”来进行拼接得操作，这只是个语法糖，其实也使用了stringBuilser得append得操作，代码如下 String a ="aa";for (int i = 0; i <5; i++) {/*每次都要先实例化一个StringBuilder，然后调用其append方法进…...
2024/4/18 1:33:01
ALTERA DDR2 IP核调试记录
一、创建IP核以管理员方式打开QUARTUS II 13.0,新建一个工程，打开MegaWizard plug-in manager找到interface>>external memory>>ddr2，选择第二个IP核，IP核地址选择放在工程同一个文件夹下，等待IP核设置GUI跳出来，win7系统下该图形界面显示不全，可以先将屏幕…...
2024/4/16 9:14:08
论文阅读《Multi-Graph Convolution Collaborative Filtering》
多图卷积系统过滤《Multi-Graph Convolution Collaborative Filtering》对这篇文章比较感兴趣，先读一读，再做思考！摘要个性化推荐无处不在，在许多在线服务中发挥着重要作用。实体研究致力于学习用户和项目的向量表示，目的是根据表示的相似性预测用户对项目的偏好。技术范围…...
2024/4/16 9:13:51