java-rabbitmq RabbitMQ快速入门学习笔记

文章目录

RabbitMQ入门一、概述1.1 什么是MQ1.2 作用1.2.1 流量消峰1.2.2 应用解耦1.2.3 异步处理

1.3 分类1.4 核心概念1.5 工作原理1.6 核心部分

二、快速入门1. 简单模式2. 工作队列3. 消息应答机制3.1 消息自动重新入队3.2 手动应答

4. 持久化4.1 概述4.2 不公平分发4.3 预期值

5. 发布确认原理5.1 单个确认发布5.2 批量确认发布5.3 异步确认发布

三、交换机1. 交换机类型2. 绑定3. Fanout4. Direct exchange5. Topics

四、死信队列1. 概述2. 代码

五、延迟队列1. 基于死信消息2. 基于插件

六、参考资料

RabbitMQ入门

一、概述

1.1 什么是MQ

MQ本质上是一个队列，FIFO先入先出原则。只不过队列中存放的内容是message而已。还是一种跨进程的通信机制，用于上下游传递消息。在互联网架构中，MQ是一种非常常见的上下游“逻辑解耦+物理解耦”的消息通信服务。使用MQ之后，消息发送上游只需要依赖MQ，不用依赖其他服务。

1.2 作用

1.2.1 流量消峰

假设订单系统最多能处理一万次订单，这个能力应付正常时段的流量戳戳有余。但是在高峰期，如果有两万次下单操作则系统是处理不了的，只能限制订单超过一万后不允许用户下单。使用消息队列做缓冲，我们就可以取消这个限制，将一秒内下的订单分散成一段时间来处理，这是有些用户可能在下单十几秒后才能收到下单成功的操作，但是比不能下单的体验更好。

1.2.2 应用解耦

电商系统为例，应用中有订单系统、库存系统、物流系统等。用户创建订单后，如果耦合调用库存系统、物流系统等，任何一个子系统出了故障，都会造成下单操作异常。当转变成基于消息队列的方式后，系统间调用的问题会减少很多。

1.2.3 异步处理

有些服务调用时异步的，假设A调用B，B需要花费很长时间执行，但是A需要知道B什么时候可以执行完，之前的方式是A过一段时间去调用B的查询API查询。或者A提供一个callback api，B执行完之后调用api通知A的服务。使用消息队列可以很方便解决这个问题。A调用B服务后，只需要监听B处理完成的消息，当B处理完成后，会发送一条消息给MQ，MQ会将此消息转发给A服务。

1.3 分类

ActiveMQ

优点：单机吞吐量万级，时效性ms级，可用性高，基于主从架构实现高可用性，消息可靠性较低的概率丢失数据缺点：官方社区维护越来越少，高吞吐量场景较少 Kafka

优点：性能卓越，单机写入TPS约在百万条/秒，吞吐量高。时效性ms级可用性非常高，kafka是分布式的，一个数据多个副本，少数机器宕机，不会丢失数据，不会导致不可用，消费者采用Pull方式获取消息，消息有序，通过控制能够保证所有消息被消费且仅被消费一次。有优秀的第三方Kafka Web管理界面Kafka-Manager。在日志领域比较成熟，被多家公司和多个开源项目使用。缺点：Kafka单机超过64个队列/分区，Load会发生明显的飙高现象，队列越多，load越高，发送消息响应时间边长，使用短轮询方式，实时性取决于轮询间隔时间，消费失败不支持重试。 RocketMQ

出自阿里巴巴的开源产品，用Java语言实现，在设计时参考Kafka，并作了相应改进优点：单机吞吐量十万级，可用性非常高，分布式架构，消息可以做到0丢失，MQ功能较为完善，扩展性好，支持10亿级别的消息堆积。缺点：支持的客户端语言不多，目前只支持Java及C++，其中C++不成熟 RabbitMQ

是一个在AMQP(高级消息队列协议)基础上完成的，可复用的企业消息系统，是当前最流行的消息中间件之一优点：由于erlang语言的高并发特性，性能较好；吞吐量万集，MQ功能比较完备，支持多语言、健壮、稳定、易用、跨平台缺点：商业版需要收费，学习成本较高

1.4 核心概念

生产者

产生数据发送消息的程序交换机

交换机是RabbitMQ非常重要的一个组件，一方面负责接收来自生产的消息，另一方面它将消息推送给队列。交换机必须确切知道如何处理它接收到的消息，是将这些消息推送到特定队列还是推送到多个队列，亦或者是把消息丢弃。队列

队列是RabbitMQ内部使用的一种数据结构，尽管消息流经RabbitMQ和应用程序，但它们只能存储在队列中。队列仅受主机的内存和磁盘限制的约束，本质上是一个大的消息缓冲区。许多生产者可以将消息发送到一个队列中，许多消费者可以尝试从一个队列接收数据。消费者

消费者大多时候是一个等待接收消息的程序。请注意生产者、消费者和消息中间件很多时候并不在同一个机器上。同一个应用程序既可以是生产者也可以是消费者。

1.5 工作原理

Broker：接收和分发消息的应用，RabbitMQ Server就是Message BrokerVirtual host：出于多租户和安全因素的设计，把AMQP的基本组件划分到一个虚拟的分组中，类似于网络中的namespace的概念。Connection：publisher/consumer和Broker之间的TCP连接Channel：如果每一次访问RabbitMQ都建立一个Connection，在消息量大的时候建立TCP Connection的开销是巨大的，效率较低。Channel是在connection内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的channel进行通讯，AMQP method包含了channel id帮助客户端和message Broker识别channel，所以channel之间完全隔离Exchange：message到达Broker的第一站，根据分发规则，匹配查询表中的routing key，分发消息到queue中。常用的类型有：direct，topic and fanout

1.6 核心部分

二、快速入门

1. 简单模式

引入pom依赖

com.rabbitmq

amqp-client

5.16.0

生产者 public class Producer {

private static final String QUEUE_NAME = "hello";

public static void main(String[] args) throws IOException, TimeoutException {

//创建连接工厂

ConnectionFactory factory = new ConnectionFactory();

factory.setHost("192.168.40.128");

factory.setUsername("admin");

factory.setPassword("admin");

Connection connection = factory.newConnection();

System.out.println("开始发送消息......");

//获取信道

Channel channel = connection.createChannel();

//创建队列

//1. param1：队列名称

//2. param2：是否持久化

//3. param3：是否只供一个消费者消费，及消息共享

//4. param4：是否自动删除最后一个消费者断开连接后，该队列是否自动删除

channel.queueDeclare(QUEUE_NAME, true, false, false, null);

// 1. param1：发送到哪个交换机

// 2. param2：路由的Key值

// 3. param3：其他参数信息

// 4. param4：发送消息的消息体

channel.basicPublish("", QUEUE_NAME, null, "hello world".getBytes());

System.out.println("发送消息结束......");

}

消费者 public class Consumer {

private static final String QUEUE_NAME="hello";

public static void main(String[] args) throws IOException, TimeoutException {

ConnectionFactory factory = new ConnectionFactory();

factory.setHost("192.168.40.128");

factory.setUsername("admin");

factory.setPassword("admin");

Connection connection = factory.newConnection();

Channel channel = connection.createChannel();

//声明接收消息

DeliverCallback deliverCallback=(consumerTag,message)->{

System.out.println(new String(message.getBody()));

};

//取消消息时的回调

CancelCallback cancelCallback=consumerTag->{

System.out.println("消费消息被中断");

};

//消费者消费消息

//1. param1：消费哪个队列

//2. param2：消费成功之后是否自动应答

//3. param3：消费者接收消息的回调

//4. param4：消费者取消消息时的回调

channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);

}

2. 工作队列

工作队列主要思想是避免立即执行资源密集型任务，而不得不等待它完成。相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当有多个工作线程时，这些工作线程将一起处理这些任务。

3. 消息应答机制

消费者完成一个任务可能需要一段时间，如果其中一个消费者处理一个长的任务仅只完成了部分就宕机了，RabbitMQ一旦向消费者传递了一条消息，便立即将该消息标记为删除。在这种情况下，突然有个消费者挂掉了，我们将丢失正在处理的消息。以及后续发送给该消费者的消息，因为它无法接收到。

为了保证消息不丢失，RabbitMQ引入了消息应答机制，消息应答：消费者在接收到消息并且处理该消息之后，告诉RabbitMQ它已经处理了，RabbitMQ可以把该消息删除了。

自动应答：消息发送后立即被认为已经发送成功，这种模式需要在高吞吐量和数据传输安全性方面做权衡，因为这种模式如果消息在接收到之前，消费者那边出现连接或者channel关闭，那么消息就丢失了，当然另一方面这种模式消费者那边可以传递过载消息，没有传递的消息数量进行限制，当然这样有可能使得消费者这边由于接收太多还来不及处理的消息，导致这些消息的积压，最终使得内存耗尽，最终这些消费者线程被操作系统杀死，所以这种模式仅适用在消费者可以高效并以某种速率能够处理这些消息的情况下使用。手动应答的方法

Channel.basicAck(用于肯定确认)：已知道该消息并且成功的处理消息，可以将其丢弃了Channel.basicNack(用于否定确认)Channel.basicReject(用于否定确认) 批量应答：Multiple：手动应答的好处是可以批量应答并且减少网络拥堵

3.1 消息自动重新入队

如果消费者由于某些原因失去连接（其通道已关闭，连接已关闭或TCP连接丢失），导致消息未发送ACK确认，RabbitMQ将了解到消息未完全处理，并将对其重新排队。如果此时其他消费者可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。

3.2 手动应答

public class ConsumerWorker01 {

private static final String QUEUE_NAME = "queue_ack";

public static void main(String[] args) throws IOException {

Channel channel = RabbitMqUtils.getChannel();

System.out.println("C1等待接收消息，处理时间为2s");

DeliverCallback deliverCallback = (tag, mess) -> {

try {

//模拟该消息需要处理2s

Thread.sleep(2000);

} catch (InterruptedException e) {

throw new RuntimeException(e);

}

System.out.println("接收到的消息：" + new String(mess.getBody(), StandardCharsets.UTF_8));

//手动应答

//1. param1：消息标记tag

//2. param2：是否批量应答

channel.basicAck(mess.getEnvelope().getDeliveryTag(), false);

};

CancelCallback cancelCallback = tag -> {

System.out.println(tag + "接收消息被取消！");

};

// 采用手动应答

channel.basicConsume(QUEUE_NAME, false, deliverCallback, cancelCallback);

}

public class ConsumerWorker02 {

private static final String QUEUE_NAME = "queue_ack";

public static void main(String[] args) throws IOException {

Channel channel = RabbitMqUtils.getChannel();

System.out.println("C2等待接收消息，处理时间为10s");

DeliverCallback deliverCallback = (tag, mess) -> {

try {

//模拟该消息需要处理10s

Thread.sleep(10000);

} catch (InterruptedException e) {

throw new RuntimeException(e);

}

System.out.println("接收到的消息：" + new String(mess.getBody(), StandardCharsets.UTF_8));

//手动应答

//1. param1：消息标记tag

//2. param2：是否批量应答

channel.basicAck(mess.getEnvelope().getDeliveryTag(), false);

};

CancelCallback cancelCallback = tag -> {

System.out.println(tag + "接收消息被取消！");

};

// 采用手动应答

channel.basicConsume(QUEUE_NAME, false, deliverCallback, cancelCallback);

}

4. 持久化

4.1 概述

默认情况下RabbitMQ退出或由于某种原因崩溃时，它忽略队列和消息，除非告知它不要这样做。确保消息不会丢失需要做两件事：我们需要将队列和消息都标记为持久化

//队列持久化

Channel channel = RabbitMqUtils.getChannel();

//创建队列

//参数列表：durable：是否持久化

//queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments)

channel.queueDeclare(QUEUE_NAME, true, false, false, null);

//消息持久化

//参数列表：props：队列的属性

basicPublish(String exchange, String routingKey, AMQP.BasicProperties props, byte[] body)

//发送消息

channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));

4.2 不公平分发

采用轮询分发时，如果多个消费者的响应速率有差异，例如某些消费者的响应速率比较快，此时采用该方式则不是很友好。为了避免这种情况，可以设置参数

channel.basicQos(1)

4.3 预期值

channel.basicQos(prefetch)

5. 发布确认原理

设置队列持久化设置消息持久化发布确认

开启发布确认方法 //调用该方法

channel.confirmSelect();

5.1 单个确认发布

单个确认发布是一种简单的确认方式，它是一种同步确认发布的方式，也就是发布一个消息之后只有它被确认发布，后续的消息才能继续发布。缺点是发布速度特别慢

public class ProducerWorker {

private static final String QUEUE_NAME = "queue_ack";

public static void main(String[] args) throws IOException, InterruptedException {

Channel channel = RabbitMqUtils.getChannel();

//开启发布确认

channel.confirmSelect();

//创建队列

channel.queueDeclare(QUEUE_NAME, true, false, false, null);

Scanner scanner = new Scanner(System.in);

for (int i = 0; i < 1000; i++) {

String mess = i + "";

channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, mess.getBytes(StandardCharsets.UTF_8));

//发布确认

boolean b = channel.waitForConfirms();

if (b) {

System.out.println("生产者发送消息成功：" + mess);

}

5.2 批量确认发布

与单个确认相比，批量确认的问题是：当发生故障时我们不知道是哪条消息出现了问题，我们必须将整个批处理保存在内存中，以记录重要的消息而后重新发布消息。

public class ProducerWorker {

private static final String QUEUE_NAME = "queue_ack";

public static void main(String[] args) throws IOException, InterruptedException {

Channel channel = RabbitMqUtils.getChannel();

//开启发布确认

channel.confirmSelect();

//创建队列

channel.queueDeclare(QUEUE_NAME, true, false, false, null);

Scanner scanner = new Scanner(System.in);

for (int i = 0; i < 1000; i++) {

String mess = i + "";

channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, mess.getBytes(StandardCharsets.UTF_8));

//100条确认一次

if (i % 100 == 0) {

boolean b = channel.waitForConfirms();

}

5.3 异步确认发布

它是利用回调函数来达到消息可靠性的传递，这个中间件也是通过函数回调来保证是否投递成功

public class ProducerWorker02 {

private static final String QUEUE_NAME = "queue_ack";

public static void main(String[] args{}

/**

* 异步发布确认

* @author yunshuaiwei

* @date 2023/3/2 10:31

**/

public static void publishMessageAsync() throws Exception {

Channel channel = RabbitMqUtils.getChannel();

//创建队列

channel.queueDeclare(QUEUE_NAME, true, false, false, null);

//开启发布确认

channel.confirmSelect();

//线程安全有序的哈希表

//可以将序号和消息进行关联，可以批量删除消息条目，且支持高并发

ConcurrentSkipListMap map = new ConcurrentSkipListMap<>();

//消息确认成功后的回调

ConfirmCallback confirmCallback = (tag, multiple) -> {

//删除已经确认的消息

if (multiple){//批量

ConcurrentNavigableMap confirmed = map.headMap(tag);

confirmed.clear();

}else{

map.remove(tag);

}

System.out.println("确认成功的消息：" + tag);

};

//消息确认失败后的回调

ConfirmCallback nackCallback = (tag, multiple) -> {

Object o = map.get(tag);

System.out.println("未确认的消息：" + o.toString());

};

//消息监听器

channel.addConfirmListener(confirmCallback, nackCallback);

for (int i = 0; i < 1000; i++) {

String mess = i + "";

channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, mess.getBytes(StandardCharsets.UTF_8));

//1.记录所有要发送的消息

map.put(channel.getNextPublishSeqNo(),mess);

}

三、交换机

上一部分中，每个任务恰好交付给一个消费者(工作进程)。在这一部分中，我们将消息发布给多个消费者。这种模式称为"发布/订阅"模式

RabbitMQ消息传递模型的核心思想是：生产者生产的消息从不会直接发送到队列。实际上，通常生产者甚至都不知道这些消息传递到哪些队列中。相反，生产者只能将消息发送到交换机，交换机工作的内容非常简单，一方面它接收来自生产者的消息，另一方面将它们推入队列。交换机必须确切知道如何处理收到的消息。是用该把这些消息放到特定队列还是说把他们放到许多队列中，这些都由交换机来决定。

1. 交换机类型

直接(direct)主题(topic)标题(headers)扇出(fanout)

2. 绑定

3. Fanout

它是将接收到的所有消息广播到它知道的所有队列中

//消费者

public class ConsumerWorker {

private static final String EXCHANGE_NAME = "exchange01";

public static void main(String[] args) throws IOException {

Channel channel = RabbitMqUtils.getChannel();

//param1：交换机名称 param2：交换机类型--扇出

channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

//获取一个临时队列

String queue = channel.queueDeclare().getQueue();

//param1：队列名称 param2：交换机名称 param3：routingKey

channel.queueBind(queue, EXCHANGE_NAME, "");

System.out.println("等待接收消息......");

channel.basicConsume(queue, true, (tag, mess) -> {

System.out.println("接收到消息：" + new String(mess.getBody(), StandardCharsets.UTF_8));

}, tag -> {

System.out.println("消息接收失败！");

});

}

//生产者

public class ProducerWorker {

private static final String EXCHANGE_NAME = "exchange01";

public static void main(String[] args) throws IOException, InterruptedException {

Channel channel = RabbitMqUtils.getChannel();

channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

Scanner scanner = new Scanner(System.in);

while (scanner.hasNext()) {

String mess = scanner.next();

channel.basicPublish(EXCHANGE_NAME, "", null, mess.getBytes(StandardCharsets.UTF_8));

System.out.println("发送消息：" + mess);

}

4. Direct exchange

Fanout不能给我们带来很大的灵活性，它只能进行无意识的广播，在这里我们将使用Direct这种类型进行替换，这种类型的消息只去它绑定的routingKey队列中。

5. Topics

发送到类型是Topics交换机的消息routing_key不能随意写，必须满足一定的要求，它必须是一个单词列表，以逗号隔开。

Q1绑定中间带有orange带三个单词的字符串Q2绑定最后一个单词是rabbit的三个单词；第一个单词是lazy的多个单词

//消费者01

public class ConsumerWorker01 {

private static final String EXCHANGE_NAME = "exchange01";

public static void main(String[] args) throws IOException {

Channel channel = RabbitMqUtils.getChannel();

//param1：交换机名称 param2：交换机类型--扇出

channel.exchangeDeclare(EXCHANGE_NAME, "topic");

//获取一个临时队列

String queue = channel.queueDeclare().getQueue();

//param1：队列名称 param2：交换机名称 param3：routingKey

channel.queueBind(queue, EXCHANGE_NAME, "*.orange.*");

System.out.println("等待接收消息......");

channel.basicConsume(queue, true, (tag, mess) -> {

System.out.println("接收到消息：" + new String(mess.getBody(), StandardCharsets.UTF_8));

System.out.println("接收队列：Q1，绑定键：" + mess.getEnvelope().getRoutingKey());

}, tag -> {

System.out.println("消息接收失败！");

});

}

//消费者02

public class ConsumerWorker02 {

private static final String EXCHANGE_NAME = "exchange01";

public static void main(String[] args) throws IOException {

Channel channel = RabbitMqUtils.getChannel();

//param1：交换机名称 param2：交换机类型--扇出

channel.exchangeDeclare(EXCHANGE_NAME, "topic");

//获取一个临时队列

String queue = channel.queueDeclare().getQueue();

//param1：队列名称 param2：交换机名称 param3：routingKey

//绑定交换机

channel.queueBind(queue, EXCHANGE_NAME, "*.*.rabbit");

channel.queueBind(queue, EXCHANGE_NAME, "lazy.#");

System.out.println("等待接收消息......");

channel.basicConsume(queue, true, (tag, mess) -> {

System.out.println("接收到消息：" + new String(mess.getBody(), StandardCharsets.UTF_8));

System.out.println("接收队列：Q2，绑定键：" + mess.getEnvelope().getRoutingKey());

}, tag -> {

System.out.println("消息接收失败！");

});

}

//生产者

public class ProducerWorker {

private static final String EXCHANGE_NAME = "exchange01";

public static void main(String[] args) throws IOException, InterruptedException {

Channel channel = RabbitMqUtils.getChannel();

channel.exchangeDeclare(EXCHANGE_NAME, "topic");

//测试代码

HashMap map = new HashMap<>();

map.put("quick.orange.rabbit", "队列Q1 Q2接收");

map.put("lazy.orange.elephant", "队列Q1 Q2接收");

map.put("quick.orange.fox", "队列Q1接收");

map.put("lazy.brown.fox", "队列Q2接收");

map.put("lazy.pink.rabbit", "满足两个绑定，但是只被Q2接收一次");

map.put("quick.brown.fox", "不匹配任何绑定不会被接收");

for (String key : map.keySet()) {

channel.basicPublish(EXCHANGE_NAME, key, MessageProperties.PERSISTENT_TEXT_PLAIN, map.get(key).getBytes(StandardCharsets.UTF_8));

System.out.println("生产者发出消息：" + map.get(key));

}

四、死信队列

1. 概述

死信及无法被消费的消息。一般来说，producer将消息投递到broker或者直接到queue中，consumer从queue取出消息进行消费，但某些时候由于特定原因导致queue中的某些消息无法被消费，这些消息如果没有进行后续处理则会变成死信，有死信自然就有了死信队列

消息TTL过期队列达到最大长度(队列满了，无法再添加数据到mq中)消息被拒绝（basic.reject或basic.nack）并且requeue=false

2. 代码

//生产者

public class ProducerWorker {

private static final String NORMAL_EXCHANGE = "normal_exchange";

public static void main(String[] args) {

Channel channel = RabbitMqUtils.getChannel();

//模拟由TTL超时导致的死信

AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();

for (int i = 0; i < 10; i++) {

String mess = "info" + i;

try {

channel.basicPublish(NORMAL_EXCHANGE, "ZhangSan", properties, mess.getBytes(StandardCharsets.UTF_8));

} catch (IOException e) {

throw new RuntimeException(e);

}

//消费者

public class ConsumerWorker {

private static final String NORMAL_EXCHANGE = "normal_exchange";

private static final String DEAD_EXCHANGE = "dead_exchange";

private static final String NORMAL_QUEUE = "normal_queue";

private static final String DEAD_QUEUE = "dead_queue";

public static void main(String[] args) {

Channel channel = RabbitMqUtils.getChannel();

try {

//声明交换机

channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);

channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

//声明队列

Map map = new HashMap<>();

//正常队列设置死信交换机

map.put("x-dead-letter-exchange", DEAD_EXCHANGE);

//设置路由RoutingKey

map.put("x-dead-letter-routing-key", "lisi");

channel.queueDeclare(NORMAL_QUEUE, false, false, false, map);

channel.queueDeclare(DEAD_QUEUE, false, false, false, null);

//队列和交换机进行绑定

channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"ZhangSan");

channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");

//发送消息

System.out.println("等待接收消息......");

channel.basicConsume(NORMAL_QUEUE, true, (consumerTag, mess) -> {

System.out.println("Consumer01接收到消息："+new String(mess.getBody(), StandardCharsets.UTF_8));

}, consumerTag -> {

System.out.println("消息接收失败：" + consumerTag);

});

} catch (IOException e) {

throw new RuntimeException(e);

}

五、延迟队列

延迟队列内部是有序的，最重要的特性就体现在延时属性上，延时队列中的元素是希望在指定时间到了以后或者之前取出和处理，简单来说，延时队列就是用来存放需要在指定时间被处理的元素的队列。

1. 基于死信消息

//配置类

@Configuration

public class TtlQueueConfig {

private static final String X_EXCHANGE = "X";

private static final String Y_DEAD_LETTER_EXCHANGE = "Y";

public static final String QUEUE_A = "QA";

public static final String QUEUE_B = "QB";

public static final String DEAD_LETTER_QUEUE = "QD";

@Bean("xExchange")

public DirectExchange xExchange() {

return new DirectExchange(X_EXCHANGE);

}

@Bean("yExchange")

public DirectExchange yExchange() {

return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);

}

@Bean("queueA")

public Queue aQueue() {

Map arguments = new HashMap<>(2);

//设置死信交换机

arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);

arguments.put("x-dead-letter-routing-key", "YD");

return QueueBuilder

.durable(QUEUE_A)

.ttl(10000)

.withArguments(arguments).build();

}

@Bean("queueB")

public Queue bQueue() {

Map arguments = new HashMap<>(2);

//设置死信交换机

arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);

arguments.put("x-dead-letter-routing-key", "YD");

return QueueBuilder

.durable(QUEUE_B)

.ttl(40000)

.withArguments(arguments).build();

}

@Bean("queueD")

public Queue dQueue() {

return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();

}

@Bean

public Binding queueABindingX(@Qualifier("queueA") Queue queueA,

@Qualifier("xExchange") DirectExchange xExchange) {

return BindingBuilder.bind(queueA).to(xExchange).with("XA");

}

@Bean

public Binding queueBBindingX(@Qualifier("queueB") Queue queueB,

@Qualifier("xExchange") DirectExchange xExchange) {

return BindingBuilder.bind(queueB).to(xExchange).with("XB");

}

@Bean

public Binding queueDBindingY(@Qualifier("queueD") Queue queueD,

@Qualifier("yExchange") DirectExchange yExchange) {

return BindingBuilder.bind(queueD).to(yExchange).with("YD");

}

//生产者

@RestController

@RequestMapping("/ttl")

@Slf4j

public class SendMsg {

@Autowired

private RabbitTemplate rabbitTemplate;

@GetMapping("/sendMsg/{message}")

public void sendMsg(@PathVariable String message) {

log.info("当前时间：{}，发送一条消息给两个TTL队列:{}", DateUtil.now(), message);

//发送消息

rabbitTemplate.convertAndSend("X", "XA", "消息来自ttl为10s的队列：" + message);

rabbitTemplate.convertAndSend("X", "XB", "消息来自ttl为40s的队列：" + message);

}

//消费者

@Slf4j

@Component

public class DeadLetterQueueConsumer {

@RabbitListener(queues = "QD")

public void receiveQueueD(Message message, Channel channel) {

String msg = new String(message.getBody(), UTF_8);

log.info("当前时间：{},收到死信队列的消息：{}", DateUtil.now(), msg);

}

2. 基于插件

首先需要安装RabbitMQ延时插件

//配置类

@Configuration

public class DelayedQueueConfig {

public static final String DELAYED_QUEUE_NAME = "delayed.queue";

public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";

@Bean

public Queue delayedQueue() {

return QueueBuilder.durable(DELAYED_QUEUE_NAME).build();

}

@Bean

public CustomExchange delayedExchange() {

Map arguments = new HashMap<>();

//延迟类型：直接类型

arguments.put("x-delayed-type", "direct");

return new CustomExchange(DELAYED_EXCHANGE_NAME, "x-delayed-message", true, false, arguments);

}

@Bean

public Binding delayedQueueBinding(@Qualifier("delayedQueue") Queue delayedQueue,

@Qualifier("delayedExchange") CustomExchange delayedExchange) {

return BindingBuilder.bind(delayedQueue).to(delayedExchange).with("delayed.routingKey").noargs();

}

//生产者

@RestController

@RequestMapping("/ttl")

@Slf4j

public class SendMsg {

@Autowired

private RabbitTemplate rabbitTemplate;

@GetMapping("/sendMsg/{message}/{delayTime}")

public void sendDelayedMsg(@PathVariable("message") String message, @PathVariable("delayTime") Integer delayTime) {

log.info("当前时间：{}，发送一条时长为{}毫秒的信息给延时队列delayed.queue：{}", DateUtil.now(), delayTime, message);

rabbitTemplate.convertAndSend(DelayedQueueConfig.DELAYED_EXCHANGE_NAME, "delayed.routingKey", message, msg -> {

//设置延时时间

msg.getMessageProperties().setDelay(delayTime);

return msg;

});

}

//消费者

@Slf4j

@Component

public class DeadLetterQueueConsumer {

@RabbitListener(queues = "delayed.queue")

public void receiveQueueC(Message message,Channel channel){

String msg=new String(message.getBody(), UTF_8);

log.info("当前时间：{},收到延时消息：{}", DateUtil.now(), msg);

}

六、参考资料

尚硅谷

好文链接

评论可见，请评论后查看内容，谢谢！！！

您阅读本篇文章共花了：

金钥匙

java-rabbitmq RabbitMQ快速入门学习笔记

分布式 RabbitMQ--消息模型

分布式 RabbitMQ基础概念-02

发表评论取消回复

金钥匙

java-rabbitmq RabbitMQ快速入门 学习笔记

分布式 RabbitMQ--消息模型

分布式 RabbitMQ基础概念-02

相关文章

发表评论取消回复

java-rabbitmq RabbitMQ快速入门学习笔记