RabbitMQ超详学习

一、MQ概述
- 1.MQ的优势
- 2.MQ的劣势
- 3.AMQP
- 4.JMS
- 5.AMQP 与 JMS 区别
二、RabbitMQ介绍
三、RabbitMQ安装
四、RabbitMQ入门
- 1.简单模式
- - 创建工程
  - 生产者
  - 消费者
- 2.RabbitMQ工作模式
- - 2.1Work queues工作队列模式
  - 生产者
  - 消费者1
  - 消费者2
- 3.Publish/Subscribe发布与订阅模式
- - 生产者
  - 消费者1
  - 消费者2
- 4.Routing路由模式
- - 生产者
  - 消费者1
  - 消费者2
- 5.Topics通配符模式
- - 生产者
  - 消费者1
  - 消费者2
五、Spring 整合RabbitMQ
- 1.生产者
- 2.消费者
- 3.消息监听器
- 4.广播监听器
- 5.（*）星号通配符监听器
- 6.（#）井号通配符监听器
六、Spring Boot整合RabbitMQ
- 1.生产者
- 2.消费者
- 3.消息监听
七、RabbitMQ的高级特性
- 1.消息的可靠投递
- - 确认模式
  - 退回模式
- 2.Consumer Ack
- 3.消费端限流
- 4.TTL
- 5.死信队列
- 6.延迟队列

一、MQ概述

MQ全称 Message Queue（消息队列），是在消息的传输过程中保存消息的容器。多用于分布式系统之间进行通信。
普通的调用：
加上RabbitMQ的调用：

1.MQ的优势

应用解耦：
系统的耦合性越高，容错性就越低，可维护性就越低。MQ相当于一个中介，生产方通过MQ与消费方交互，它将应用程序进行解耦合。提升容错性和可维护性。
任务异步处理：
将不需要同步处理的并且耗时长的操作由消息队列通知消息接收方进行异步处理，提高了应用程序的响应时间。
削峰填谷

如系统A，在执行时候会往数据库写数据。但是数据库只能支撑每秒1000左右的并发写入，并发量再高就容易宕机。低峰期的时候并发也就100多个，但是在高峰期时候，并发量会突然激增到5000以上，这个时候数据库肯定卡死了。但是使用了MQ之后，限制消费消息的速度为1000，但是这样一来，高峰期产生的数据势必会被积压在MQ中，高峰就被“削”掉了。但是因为消息积压，在高峰期过后的一段时间内，消费消息的速度还是会维持在1000QPS，直到消费完积压的消息,这就叫做“填谷”

2.MQ的劣势

系统可用性降低：
系统引入的外部依赖越多，系统稳定性越差。一旦 MQ 宕机，就会对业务造成影响。
系统复杂度提高
MQ 的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过 MQ 进行异步调用。

3.AMQP

AMQP，即 Advanced Message Queuing Protocol（高级消息队列协议），是一个网络协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，遵循此协议，不收客户端和中间件产品和开发语言限制。2006年，AMQP 规范发布。类比HTTP。

4.JMS

JMS 即 Java 消息服务（JavaMessage Service）应用程序接口，是一个 Java 平台中关于面向消息中间件的API

5.AMQP 与 JMS 区别

JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式
JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
JMS规定了两种消息模式；而AMQP的消息模式更加丰富

二、RabbitMQ介绍

目前业界有很多的 MQ 产品，例如 RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、MetaMq等，也有直接使用 Redis 充当消息队列的案例，而这些消息队列产品，各有侧重，在实际选型时，需要结合自身需求及 MQ 产品特征，综合考虑。在这里我们选择RabbitMQ
RabbitMQ 是一个消息代理：它接受和转发消息。您可以将其视为邮局：当您将要邮寄的邮件放入邮箱时，您可以确定邮递员最终会将邮件递送给您的收件人。在这个类比中，RabbitMQ 是一个邮箱、一个邮局和一个邮递员。 RabbitMQ 和邮局之间的主要区别在于它不处理纸张，而是接受、存储和转发二进制数据块 -消息。
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，在消息量大的时候建立 TCPConnection的开销将是巨大的，效率也较低。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) andfanout (multicast)
Queue	消息最终被送到这里等待 consumer 取走
Binding	exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key。Binding 信息被保存到 exchange 中的查询表中，用于 message 的分发依据

三、RabbitMQ安装

安装及控制台部分操作请点击链接:
《手把手教你在Linux上和docker上安装RabbitMQ以及部分控制台操作》

四、RabbitMQ入门

RabbitMQ提供了6种模式：简单模式，work模式，Publish/Subscribe发布与订阅模式，Routing路由模式，Topics主题模式，RPC远程调用模式

先登录要使用的用户：

1.简单模式

如图：

在下图中，“P”是我们的生产者：

生产无非就是发送，发送消息的程序是生产者。
“C”是我们的消费者：

消费与接收具有相似的含义。消费者是一个主要等待接收消息的程序
中间的框是一个队列：

队列是位于 RabbitMQ 中的邮箱的名称。尽管消息流经 RabbitMQ 和您的应用程序，但它们只能存储在队列中。队列仅受主机的内存和磁盘限制，它本质上是一个大的消息缓冲区。许多生产者可以发送去一个队列的消息，许多消费者可以尝试从一个队列接收数据。这就是我们表示队列的方式：，一个 RabbitMQ 代表消费者保存的消息缓冲区

创建工程

pox.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>org.lrk</groupId><artifactId>rabbitmq-demo</artifactId><version>1.0-SNAPSHOT</version><dependencies><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.6.0</version></dependency></dependencies></project>

由于每次都要连接rabbitmq，所以我们抽取连接代码为工具类，每次连接我们只要调工具类就行了：

package com.lrk.rabbitmq.utils;import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class ConnectionUtils {public static Connection getConnection() throws IOException, TimeoutException {ConnectionFactory connectionFactory = new ConnectionFactory();//主机地址;默认为 localhostconnectionFactory.setHost("192.168.220.129");//连接端口;默认为 5672connectionFactory.setPort(5672);//虚拟主机名称;默认为 /connectionFactory.setVirtualHost("/lin");//连接用户名；默认为guestconnectionFactory.setUsername("Link");//连接密码；默认为guestconnectionFactory.setPassword("123");//创建链接Connection connection = connectionFactory.newConnection();return connection;}
}

生产者

package com.lrk.rabbitmq.simple;import com.lrk.rabbitmq.utils.ConnectionUtils;
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 {public static final String QUEUE_NAME = "simple_queue";public static void main(String[] args) throws IOException, TimeoutException {//创建链接Connection connection = ConnectionUtils.getConnection();//创建频道Channel channel = connection.createChannel();//声明（创建）队列/*** 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(QUEUE_NAME,true,false,false,null);//发送消息String messgae="hello";/*** 参数1：交换机名称，如果没有指定则使用默认Default Exchage* 参数2：路由key,简单模式可以传递队列名称* 参数3：消息其它属性* 参数4：消息内容*/channel.basicPublish("",QUEUE_NAME,null,messgae.getBytes());System.out.println("已发送消息");//释放资源channel.close();connection.close();}
}

执行：

如图：
产生了一个待消费的消息：

点击name进去可以看到发送的消息：

消费者

package com.lrk.rabbitmq.simple;import com.lrk.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer {public static void main(String[] args) throws IOException, TimeoutException {//创建链接Connection connection = ConnectionUtils.getConnection();//创建频道Channel channel = connection.createChannel();//声明（创建）队列/*** 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(Producer.QUEUE_NAME,true,false,false,null);//接收消息DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {/*** 接收到消息执行的回调* consumerTag 消息者标签，在channel.basicConsume时候可以指定* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)* properties 属性信息* 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("接收到的消息为：" + new String(body, "utf-8"));}};/*** 参数1：队列名称* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认* 参数3：消息接收到后回调*/channel.basicConsume(Producer.QUEUE_NAME,true,defaultConsumer);System.out.println("已发送消息");//释放资源
//        channel.close();
//        connection.close();}
}

执行：

如图：
产生了消费者一个，消息已被消费

2.RabbitMQ工作模式

2.1Work queues工作队列模式

工作队列（又名：任务队列）背后的主要思想是避免立即执行资源密集型任务而不得不等待它完成。相反，我们将任务安排在以后完成。我们将任务封装为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当您运行许多工作人员时，任务将在他们之间共享。
Work Queues 与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。
应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

生产者

package com.lrk.rabbitmq.work;import com.lrk.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer {public static final String QUEUE_NAME = "work_queue";public static void main(String[] args) throws IOException, TimeoutException {//创建链接Connection connection = ConnectionUtils.getConnection();//创建频道Channel channel = connection.createChannel();//声明（创建）队列/*** 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(QUEUE_NAME,true,false,false,null);//发送消息/*** 参数1：交换机名称，如果没有指定则使用默认Default Exchage* 参数2：路由key,简单模式可以传递队列名称* 参数3：消息其它属性* 参数4：消息内容*/for (int i = 0; i <30 ; i++) {String messgae="hello"+i;channel.basicPublish("",QUEUE_NAME,null,messgae.getBytes());System.out.println("已发送消息");}//释放资源channel.close();connection.close();}
}

执行：
发送30条消息

如图：

消费者1

package com.lrk.rabbitmq.work;import com.lrk.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//创建链接Connection connection = ConnectionUtils.getConnection();//创建频道Channel channel = connection.createChannel();//声明（创建）队列/*** 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(Producer.QUEUE_NAME,true,false,false,null);//接收消息DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {/*** 接收到消息执行的回调* consumerTag 消息者标签，在channel.basicConsume时候可以指定* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)* properties 属性信息* 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(Producer.QUEUE_NAME,true,defaultConsumer);//释放资源
//        channel.close();
//        connection.close();}
}

消费者2

package com.lrk.rabbitmq.work;import com.lrk.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {//创建链接Connection connection = ConnectionUtils.getConnection();//创建频道Channel channel = connection.createChannel();//声明（创建）队列/*** 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(Producer.QUEUE_NAME,true,false,false,null);//接收消息DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {/*** 接收到消息执行的回调* consumerTag 消息者标签，在channel.basicConsume时候可以指定* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)* properties 属性信息* 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(Producer.QUEUE_NAME,true,defaultConsumer);//释放资源
//        channel.close();
//        connection.close();}
}

执行2个消费者：
消费者1，接收了15条，分别是：hello0，hello2，hello4，hello6，…,hello30

消费者2，接收了15条，分别是：hello1，hello3，hello5，hello7，… ,hello29

如图：

3.Publish/Subscribe发布与订阅模式

发布订阅模式： 1、每个消费者监听自己的队列。 2、生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息

有几种可用的交换类型：direct、topic、headers 和fanout

生产者

package com.lrk.rabbitmq.ps;import com.lrk.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer {public static final String FANOUT_EXCHANGE = "fanout_exchange";public static final String FANOUT_QUEUE1 = "fanout_queue1";public static final String FANOUT_QUEUE2 = "fanout_queue2";public static void main(String[] args) throws IOException, TimeoutException {//创建链接Connection connection = ConnectionUtils.getConnection();//创建频道Channel channel = connection.createChannel();//创建交换机/*** 声明交换机* 参数1：交换机名称* 参数2：交换机类型，fanout、topic、direct、headers*/channel.exchangeDeclare(FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT);//声明（创建）队列/*** 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(FANOUT_QUEUE1,true,false,false,null);channel.queueDeclare(FANOUT_QUEUE2,true,false,false,null);//        队列绑定交换机channel.queueBind(FANOUT_QUEUE1,FANOUT_EXCHANGE,"");channel.queueBind(FANOUT_QUEUE2,FANOUT_EXCHANGE,"");//发送消息/*** 参数1：交换机名称，如果没有指定则使用默认Default Exchage* 参数2：路由key,简单模式可以传递队列名称* 参数3：消息其它属性* 参数4：消息内容*/for (int i = 0; i <30 ; i++) {String messgae="hello"+i;channel.basicPublish(FANOUT_EXCHANGE,"",null,messgae.getBytes());System.out.println("已发送消息");}//释放资源channel.close();connection.close();}
}

执行：

消费者1

package com.lrk.rabbitmq.ps;import com.lrk.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//创建链接Connection connection = ConnectionUtils.getConnection();//创建频道Channel channel = connection.createChannel();//声明（创建）队列/*** 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(Producer.FANOUT_QUEUE1,true,false,false,null);//        队列绑定交换机channel.queueBind(Producer.FANOUT_QUEUE1,Producer.FANOUT_EXCHANGE,"");//接收消息DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {/*** 接收到消息执行的回调* consumerTag 消息者标签，在channel.basicConsume时候可以指定* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)* properties 属性信息* 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(Producer.FANOUT_QUEUE1,true,defaultConsumer);//释放资源
//        channel.close();
//        connection.close();}
}

执行：

消费者2

package com.lrk.rabbitmq.ps;import com.lrk.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {//创建链接Connection connection = ConnectionUtils.getConnection();//创建频道Channel channel = connection.createChannel();//声明（创建）队列/*** 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(Producer.FANOUT_QUEUE2,true,false,false,null);//        队列绑定交换机channel.queueBind(Producer.FANOUT_QUEUE2,Producer.FANOUT_EXCHANGE,"");//接收消息DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {/*** 接收到消息执行的回调* consumerTag 消息者标签，在channel.basicConsume时候可以指定* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)* properties 属性信息* 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(Producer.FANOUT_QUEUE2,true,defaultConsumer);//释放资源
//        channel.close();
//        connection.close();}
}

执行：

绑定：
将交换机和队列绑定在一起

查看创建的交换机：

查看绑定信息：

消费消息：

4.Routing路由模式

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

生产者

package com.lrk.rabbitmq.routing;import com.lrk.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer {public static final String DIRECT_EXCHANGE = "direct_exchange";public static final String DIRECT_QUEUE1 = "direct_queue1";public static final String DIRECT_QUEUE2 = "direct_queue2";public static void main(String[] args) throws IOException, TimeoutException {//创建链接Connection connection = ConnectionUtils.getConnection();//创建频道Channel channel = connection.createChannel();//创建交换机/*** 声明交换机* 参数1：交换机名称* 参数2：交换机类型，fanout、topic、direct、headers*/channel.exchangeDeclare(DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT);//声明（创建）队列/*** 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(DIRECT_QUEUE1,true,false,false,null);channel.queueDeclare(DIRECT_QUEUE2,true,false,false,null);//        队列绑定交换机channel.queueBind(DIRECT_QUEUE1,DIRECT_EXCHANGE,"insert");channel.queueBind(DIRECT_QUEUE2,DIRECT_EXCHANGE,"update");//发送消息/*** 参数1：交换机名称，如果没有指定则使用默认Default Exchage* 参数2：路由key,简单模式可以传递队列名称* 参数3：消息其它属性* 参数4：消息内容*/String messgae="insert ";channel.basicPublish(DIRECT_EXCHANGE,"insert",null,messgae.getBytes());System.out.println("已发送消息");String messgae2="update ";channel.basicPublish(DIRECT_EXCHANGE,"update",null,messgae2.getBytes());System.out.println("已发送消息");//释放资源channel.close();connection.close();}
}

执行：

消费者1

package com.lrk.rabbitmq.routing;import com.lrk.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//创建链接Connection connection = ConnectionUtils.getConnection();//创建频道Channel channel = connection.createChannel();//声明（创建）队列/*** 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(Producer.DIRECT_EXCHANGE,true,false,false,null);//        队列绑定交换机channel.queueBind(Producer.DIRECT_QUEUE1, Producer.DIRECT_EXCHANGE,"insert");//接收消息DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {/*** 接收到消息执行的回调* consumerTag 消息者标签，在channel.basicConsume时候可以指定* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)* properties 属性信息* 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(Producer.DIRECT_QUEUE1,true,defaultConsumer);//释放资源
//        channel.close();
//        connection.close();}
}

执行：

消费者2

package com.lrk.rabbitmq.routing;import com.lrk.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {//创建链接Connection connection = ConnectionUtils.getConnection();//创建频道Channel channel = connection.createChannel();//声明（创建）队列/*** 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(Producer.DIRECT_QUEUE2,true,false,false,null);//        队列绑定交换机channel.queueBind(Producer.DIRECT_QUEUE2, Producer.DIRECT_EXCHANGE,"update");//接收消息DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {/*** 接收到消息执行的回调* consumerTag 消息者标签，在channel.basicConsume时候可以指定* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)* properties 属性信息* 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(Producer.DIRECT_QUEUE2,true,defaultConsumer);//释放资源
//        channel.close();
//        connection.close();}
}

执行：

如图：

创建了一个交换机：

查看绑定：

5.Topics通配符模式

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

通配符规则：

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

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

生产者

package com.lrk.rabbitmq.topic;import com.lrk.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer {public static final String TOPIC_EXCHANGE = "topic_exchange";public static final String TOPIC_QUEUE_ALL = "topic_queue_all";public static final String TOPIC_QUEUE2a_INSERT_UPDATE = "topic_queue_insert_update";public static void main(String[] args) throws IOException, TimeoutException {//创建链接Connection connection = ConnectionUtils.getConnection();//创建频道Channel channel = connection.createChannel();//创建交换机/*** 声明交换机* 参数1：交换机名称* 参数2：交换机类型，fanout、topic、direct、headers*/channel.exchangeDeclare(TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC);//发送消息/*** 参数1：交换机名称，如果没有指定则使用默认Default Exchage* 参数2：路由key,简单模式可以传递队列名称* 参数3：消息其它属性* 参数4：消息内容*/String messgae="insert ";channel.basicPublish(TOPIC_EXCHANGE,"item.insert",null,messgae.getBytes());System.out.println("已发送消息");messgae="update ";channel.basicPublish(TOPIC_EXCHANGE,"item.update",null,messgae.getBytes());System.out.println("已发送消息");messgae="delete ";channel.basicPublish(TOPIC_EXCHANGE,"item.delete",null,messgae.getBytes());System.out.println("已发送消息");//释放资源channel.close();connection.close();}
}

执行：

消费者1

package com.lrk.rabbitmq.topic;import com.lrk.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//创建链接Connection connection = ConnectionUtils.getConnection();//创建频道Channel channel = connection.createChannel();//声明交换机channel.exchangeDeclare(Producer.TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC);//声明（创建）队列/*** 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(Producer.TOPIC_QUEUE_ALL,true,false,false,null);//        队列绑定交换机channel.queueBind(Producer.TOPIC_QUEUE_ALL, Producer.TOPIC_EXCHANGE,"item.*");//接收消息DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {/*** 接收到消息执行的回调* consumerTag 消息者标签，在channel.basicConsume时候可以指定* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)* properties 属性信息* 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(Producer.TOPIC_QUEUE_ALL,true,defaultConsumer);//释放资源
//        channel.close();
//        connection.close();}
}

消费者2

package com.lrk.rabbitmq.topic;import com.lrk.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {//创建链接Connection connection = ConnectionUtils.getConnection();//创建频道Channel channel = connection.createChannel();//声明交换机channel.exchangeDeclare(Producer.TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC);//声明（创建）队列/*** 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(Producer.TOPIC_QUEUE2a_INSERT_UPDATE,true,false,false,null);//        队列绑定交换机channel.queueBind(Producer.TOPIC_QUEUE2a_INSERT_UPDATE, Producer.TOPIC_EXCHANGE,"item.update");channel.queueBind(Producer.TOPIC_QUEUE2a_INSERT_UPDATE, Producer.TOPIC_EXCHANGE,"item.insert");//接收消息DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {/*** 接收到消息执行的回调* consumerTag 消息者标签，在channel.basicConsume时候可以指定* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)* properties 属性信息* 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(Producer.TOPIC_QUEUE2a_INSERT_UPDATE,true,defaultConsumer);//释放资源
//        channel.close();
//        connection.close();}
}

如图：

查看交换机的绑定：

五、Spring 整合RabbitMQ

1.生产者

pox.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.link</groupId><artifactId>producer</artifactId><version>1.0-SNAPSHOT</version><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></project>

编写配置文件rabbitmq.properties：

配置rabbitmq的IP地址、端口号、账号、密码、虚拟机

rabbitmq.host=192.168.220.129
rabbitmq.port=5672
rabbitmq.username=Link
rabbitmq.password=123
rabbitmq.virtual-host=/lin

整合配置文件

<?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:properties/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="Link.*" queue="spring_topic_queue_star"/><rabbit:binding pattern="Link.#" queue="spring_topic_queue_well"/><rabbit:binding pattern="lin.#" queue="spring_topic_queue_well2"/></rabbit:bindings></rabbit:topic-exchange><!--定义rabbitTemplate对象操作可以在代码中方便发送消息--><rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>
</beans>

创建测试文件

package com.link.rabbitmq;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.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:/spring/rabbitmq.xml")
public class ProducerTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void testQueue() {rabbitTemplate.convertAndSend("spring_queue", "只发送spring_queue的消息");}/*** 发送广播* 交换机类型为 fanout* 绑定到该交换机的所有队列都能够收到消息*/@Testpublic void fanoutTest(){/*** 参数1：交换机名称* 参数2：路由键名（广播设置为空）* 参数3：发送的消息内容*/rabbitTemplate.convertAndSend("spring_fanout_exchange", "", "发送到spring_fanout_exchange交换机的广播消息");}/*** 通配符* 交换机类型为 topic* 匹配路由键的通配符，*表示一个单词，#表示多个单词* 绑定到该交换机的匹配队列能够收到对应消息*/@Testpublic void topicTest(){/*** 参数1：交换机名称* 参数2：路由键名* 参数3：发送的消息内容*/rabbitTemplate.convertAndSend("spring_topic_exchange", "hello1", "发送到spring_topic_exchange交换机hello1的消息");rabbitTemplate.convertAndSend("spring_topic_exchange", "hello2", "发送到spring_topic_exchange交换机hello2的消息");rabbitTemplate.convertAndSend("spring_topic_exchange", "hello3", "发送到spring_topic_exchange交换机hello3的消息");rabbitTemplate.convertAndSend("spring_topic_exchange", "hello14", "发送到spring_topic_exchange交换机hello14的消息");}}

2.消费者

pox.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.link</groupId><artifactId>consumer</artifactId><version>1.0-SNAPSHOT</version><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></project>

编写配置文件rabbitmq.properties：

配置rabbitmq的IP地址、端口号、账号、密码、虚拟机

rabbitmq.host=192.168.220.129
rabbitmq.port=5672
rabbitmq.username=Link
rabbitmq.password=123
rabbitmq.virtual-host=/lin

整合配置文件

<?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:properties/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.link.rabbitmq.listener.SpringQueueListener"/><bean id="fanoutListener1" class="com.link.rabbitmq.listener.FanoutListener1"/><bean id="fanoutListener2" class="com.link.rabbitmq.listener.FanoutListener2"/><bean id="topicListenerStar" class="com.link.rabbitmq.listener.TopicListenerStar"/><bean id="topicListenerWell" class="com.link.rabbitmq.listener.TopicListenerWell"/><bean id="topicListenerWell2" class="com.link.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>

3.消息监听器

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();}}
}

4.广播监听器

public class FanoutListener2 implements MessageListener {public 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();}}}

public class FanoutListener2 implements MessageListener {public 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();}}}

5.（*）星号通配符监听器

public class TopicListenerStar implements MessageListener {public 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();}}}

6.（#）井号通配符监听器

public class TopicListenerWell implements MessageListener {public 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();}}}

public class TopicListenerWell2 implements MessageListener {public 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 Boot整合RabbitMQ

1.生产者

pox.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><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.1.4.RELEASE</version></parent><groupId>com.link</groupId><artifactId>producer</artifactId><version>1.0-SNAPSHOT</version><dependencies><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>

启动类

@SpringBootApplication
public class ProducerApplication {public static void main(String[] args) {SpringApplication.run(ProducerApplication.class, args);}
}

配置yml文件

spring:rabbitmq:host: 192.168.220.129port: 5672virtual-host: /linusername: Linkpassword: 123

绑定交换机和队列

@Configuration
public class RabbitMQConfig {//交换机名称public static final String ITEM_TOPIC_EXCHANGE = "springboot_item_topic_exchange";//队列名称public static final String ITEM_QUEUE = "springboot_item_queue";/*** 声明交换机* @return*/@Beanpublic Exchange itemTopicExchange() {return ExchangeBuilder.topicExchange(ITEM_TOPIC_EXCHANGE).durable(true).build();}@Beanpublic Queue itemQueue() {return QueueBuilder.durable(ITEM_QUEUE).build();}@Beanpublic Binding itemQueueExchange(@Qualifier("itemQueue") Queue queue, @Qualifier("itemTopicExchange") Exchange exchange) {return BindingBuilder.bind(queue).to(exchange).with("item.#").noargs();}}

测试

@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");}
}

2.消费者

pox.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><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.1.4.RELEASE</version></parent><groupId>com.link</groupId><artifactId>consumer</artifactId><version>1.0-SNAPSHOT</version><dependencies><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>

启动类

@SpringBootApplication
public class ConsumerApplication {public static void main(String[] args) {SpringApplication.run(ConsumerApplication.class, args);}
}

配置yml文件

spring:rabbitmq:host: 192.168.220.129port: 5672virtual-host: /linusername: Linkpassword: 123

3.消息监听

@Component
public class MyListener {@RabbitListener(queues = "springboot_item_queue")public void myListener1(String message) {System.out.println("消费者接收到的消息为：" + message);}}

七、RabbitMQ的高级特性

在使用 RabbitMQ 的时候，作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ 为我
们提供了两种方式用来控制消息的投递可靠性模式。
confirm 确认模式
return 退回模式
rabbitmq 整个消息投递的路径为：
producer—>rabbitmq broker—>exchange—>queue—>consumer
消息从 producer 到 exchange 则会返回一个 confirmCallback 。
消息从 exchange–>queue 投递失败则会返回一个 returnCallback 。
我们将利用这两个 callback 控制消息的可靠性投递

1.消息的可靠投递

确认模式

 @Testpublic void testConfirm() {rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {/**** @param correlationData 相关配置信息* @param ack   exchange交换机 是否成功收到了消息。true 成功，false代表失败* @param cause 失败原因*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {System.out.println("confirm方法被执行了。。。");if (ack) {System.out.println("成功接收消息：" + cause);} else {System.out.println("接收消息失败：" + cause);//业务处理}}});rabbitTemplate.convertAndSend("test_exchange_confirm22", "confirm", "message confirm....");}

退回模式

 @Testpublic void testReturn() {//设置交换机处理失败消息的模式rabbitTemplate.setMandatory(true);//2.设置ReturnCallBackrabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {/**** @param message   消息对象* @param replyCode 错误码* @param replyText 错误信息* @param exchange  交换机* @param routingKey 路由键*/@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {System.out.println("return 执行了....");System.out.println(message);System.out.println(replyCode);System.out.println(replyText);System.out.println(exchange);System.out.println(routingKey);//处理}});//3. 发送消息rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm", "message confirm....");}

2.Consumer Ack

ack指Acknowledge，确认。表示消费端收到消息后的确认方式。
有三种确认方式：
自动确认：acknowledge=“none”
手动确认：acknowledge=“manual”
根据异常情况确认：acknowledge=“auto”，（这种方式使用麻烦，不作讲解）
其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从
RabbitMQ 的消息缓存中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么
该消息就会丢失。如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手
动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

@Component
public class AckListener implements ChannelAwareMessageListener {@Overridepublic void onMessage(Message message, Channel channel) throws Exception {long deliverTag = message.getMessageProperties().getDeliveryTag();try {System.out.println(new String(message.getBody()));System.out.println("处理业务逻辑");int i = 3/0;//模拟业务处理异常channel.basicAck(deliverTag, true);} catch (Exception e) {e.printStackTrace();//拒绝签收//requeue：true：表示重回队列channel.basicNack(deliverTag, true, true);
//            channel.basicReject(deliverTag, true);}}
}

3.消费端限流

@Component
public class QosListener implements ChannelAwareMessageListener {@Overridepublic void onMessage(Message message, Channel channel) throws Exception {Thread.sleep(1000);//1.获取消息System.out.println(new String(message.getBody()));//2. 处理业务逻辑//3. 签收channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);}
}

4.TTL

 <!--ttl--><rabbit:queue name="test_queue_ttl" id="test_queue_ttl"><rabbit:queue-arguments><entry key="x-message-ttl" value="100000" value-type="java.lang.Integer"></entry></rabbit:queue-arguments></rabbit:queue><rabbit:topic-exchange name="test_exchange_ttl"><rabbit:bindings><rabbit:binding pattern="ttl.#" queue="test_queue_ttl"></rabbit:binding></rabbit:bindings></rabbit:topic-exchange>

@Testpublic void testTtl() {//        for (int i = 0; i < 10; i++) {//            // 发送消息
//            rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.hehe", "message ttl....");
//        }//        // 消息后处理对象，设置一些消息的参数信息MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {//1.设置message的信息message.getMessageProperties().setExpiration("5000");//消息的过期时间//2.返回该消息return message;}};
//        //消息单独过期
//        rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.hehe", "message ttl....", messagePostProcessor);for (int i = 0; i < 10; i++) {if(i == 5){//消息单独过期rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.hehe", "message ttl....",messagePostProcessor);}else{//不过期的消息rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.hehe", "message ttl....");}}}

5.死信队列

    <rabbit:queue name="test_queue_dlx" id="test_queue_dlx"><!--3. 正常队列绑定死信交换机--><rabbit:queue-arguments><!--3.1 x-dead-letter-exchange：死信交换机名称--><entry key="x-dead-letter-exchange" value="exchange_dlx" /><!--3.2 x-dead-letter-routing-key：发送给死信交换机的routingkey--><entry key="x-dead-letter-routing-key" value="dlx.hehe" /><!--4.1 设置队列的过期时间 ttl--><entry key="x-message-ttl" value="10000" value-type="java.lang.Integer" /><!--4.2 设置队列的长度限制 max-length --><entry key="x-max-length" value="10" value-type="java.lang.Integer" /></rabbit:queue-arguments></rabbit:queue><rabbit:topic-exchange name="test_exchange_dlx"><rabbit:bindings><rabbit:binding pattern="test.dlx.#" queue="test_queue_dlx"></rabbit:binding></rabbit:bindings></rabbit:topic-exchange><!--2. 声明死信队列(queue_dlx)和死信交换机(exchange_dlx)--><rabbit:queue name="queue_dlx" id="queue_dlx"></rabbit:queue><rabbit:topic-exchange name="exchange_dlx"><rabbit:bindings><rabbit:binding pattern="dlx.#" queue="queue_dlx"></rabbit:binding></rabbit:bindings></rabbit:topic-exchange>

生产端测试：

@Testpublic void testDlx(){//1. 测试过期时间，死信消息
//        rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息，我会死吗？");//2. 测试长度限制后，消息死信
//        for (int i = 0; i < 20; i++) {//            rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息，我会死吗？");
//        }//3. 测试消息拒收rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息，我会死吗？");}

消费端

@Component
public class DlxListener implements ChannelAwareMessageListener {@Overridepublic void onMessage(Message message, Channel channel) throws Exception {long deliveryTag = message.getMessageProperties().getDeliveryTag();try {//1.接收转换消息System.out.println(new String(message.getBody()));//2. 处理业务逻辑System.out.println("处理业务逻辑...");int i = 3/0;//出现错误//3. 手动签收channel.basicAck(deliveryTag,true);} catch (Exception e) {//e.printStackTrace();System.out.println("出现异常，拒绝接受");//4.拒绝签收，不重回队列 requeue=falsechannel.basicNack(deliveryTag,true,false);}}
}

6.延迟队列

<rabbit:queue id="order_queue" name="order_queue"><!-- 3. 绑定，设置正常队列过期时间为30分钟--><rabbit:queue-arguments><entry key="x-dead-letter-exchange" value="order_exchange_dlx" /><entry key="x-dead-letter-routing-key" value="dlx.order.cancel" /><entry key="x-message-ttl" value="10000" value-type="java.lang.Integer" /></rabbit:queue-arguments></rabbit:queue><rabbit:topic-exchange name="order_exchange"><rabbit:bindings><rabbit:binding pattern="order.#" queue="order_queue"></rabbit:binding></rabbit:bindings></rabbit:topic-exchange><!--  2. 定义死信交换机（order_exchange_dlx）和队列(order_queue_dlx)--><rabbit:queue id="order_queue_dlx" name="order_queue_dlx"></rabbit:queue><rabbit:topic-exchange name="order_exchange_dlx"><rabbit:bindings><rabbit:binding pattern="dlx.order.#" queue="order_queue_dlx"></rabbit:binding></rabbit:bindings></rabbit:topic-exchange>

生产端测试

@Testpublic  void testDelay() throws InterruptedException {//1.发送订单消息。 将来是在订单系统中，下单成功后，发送消息SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");rabbitTemplate.convertAndSend("order_exchange","order.msg","订单信息：id=1,time=" + sdf.format(new Date()));/*//2.打印倒计时10秒for (int i = 10; i > 0 ; i--) {System.out.println(i+"...");Thread.sleep(1000);}*/}

消费端

@Component
public class OrderListener implements ChannelAwareMessageListener {@Overridepublic void onMessage(Message message, Channel channel) throws Exception {long deliveryTag = message.getMessageProperties().getDeliveryTag();try {//1.接收转换消息System.out.println(new String(message.getBody()));//2. 处理业务逻辑System.out.println("处理业务逻辑...");System.out.println("根据订单id查询其状态...");System.out.println("判断状态是否为支付成功");System.out.println("取消订单，回滚库存....");//3. 手动签收channel.basicAck(deliveryTag,true);} catch (Exception e) {//e.printStackTrace();System.out.println("出现异常，拒绝接受");//4.拒绝签收，不重回队列 requeue=falsechannel.basicNack(deliveryTag,true,false);}}
}

RabbitMQ超详学习相关推荐

消息队列超详解（以RabbitMQ和Kafka为例，为何使用消息队列、优缺点、高可用性、问题解决）
消息队列超详解(以RabbitMQ和Kafka为例) 为什么要用消息队列这个东西? 先说一下消息队列的常见使用场景吧,其实场景有很多,但是比较核心的有3个:解耦.异步.削峰. 解耦:现场画个图来说明一 ...
计算机网络原理超详解说
计算机网络原理超详解说前言大家好,我是泰斗贤若如,一个专注于用大白话讲解技术的号主,这次给大家分享计算机网络原理的相关知识,我自认为文章内容已经很通俗易懂了,祝您阅读愉快! 一.计算机网络概述时 ...
Android vector标签 PathData 画图超详解
此文章来源于https://www.cnblogs.com/yuhanghzsd/p/5466846.html点击打开链接 Android vector标签 PathData 画图超详解 SVG是一种 ...
超全、超详的Spring Boot配置讲解笔记
超全.超详的Spring Boot配置讲解笔记 springboot默认加载配置 SpringBoot使用两种全局的配置文件,全局配置文件可以对一些默认配置进行修改. application.prop ...
Unity超基础学习笔记（二）
Unity超基础学习笔记(二) 1. 基本数据类型的扩展之前在K12中学习了一些基本的数据类型,实际上C#支持更多的数据类型.如下: 注意无符号整型数和有符号整型数的表示范围,例如: int 能表示 ...
Mybatis案例超详解
Mybatis案例超详解前言: 本来是想像之前一样继续跟新Mybatis,但由于种种原因,迟迟没有更新,快开学了,学了一个暑假,博客也更新了不少,我觉得我得缓缓,先整合一些案例练练,等我再成熟点理解 ...
python控制手机模拟器_Appium+python自动化之连接模拟器并启动淘宝APP（超详解）...
简介上一篇讲解完模拟器的安装.配置好以后,就好比我们手机已经买好,并且系统已经做好了,就差我们用数据线和电脑连接开始实战了,这篇宏哥就带着小伙伴们和童鞋们趁热打铁,讲解和分享一下如何连接模拟器(电脑 ...
线性规划之单纯形法【超详解+图解】-转载
线性规划之单纯形法[超详解+图解] 目录 1.作用 2.线性规划的一般形式 5.1几何意义 5.2如何判断最优 5.3如何选择新的基变量 5.4如何选择被替换的基变量 5.5终止条件标准型: 转化为 ...
Java数据库部分（MySQL+JDBC）（一、MySQL超详细学习笔记）
所有示例使用的数据表均为Oracle提供的SQL基础数据表(t_employees.sql dept.sql emp.sql salgrade.sql) 熟练掌握多多练习即可达到完成后端开发所需具备的 ...

RabbitMQ超详学习

RabbitMQ超详学习

一、MQ概述

1.MQ的优势

2.MQ的劣势

3.AMQP

4.JMS

5.AMQP 与 JMS 区别

二、RabbitMQ介绍

三、RabbitMQ安装

四、RabbitMQ入门

1.简单模式

创建工程

生产者

消费者

2.RabbitMQ工作模式

2.1Work queues工作队列模式

生产者

消费者1

消费者2

3.Publish/Subscribe发布与订阅模式

生产者

消费者1

消费者2

4.Routing路由模式

生产者

消费者1

消费者2

5.Topics通配符模式

生产者

消费者1

消费者2

五、Spring 整合RabbitMQ

1.生产者

2.消费者

3.消息监听器

4.广播监听器

5.（*）星号通配符监听器

6.（#）井号通配符监听器

六、Spring Boot整合RabbitMQ

1.生产者

2.消费者

3.消息监听

七、RabbitMQ的高级特性

1.消息的可靠投递

确认模式

退回模式

2.Consumer Ack

3.消费端限流

4.TTL

5.死信队列

6.延迟队列

RabbitMQ超详学习相关推荐

最新文章

热门文章