RabbitMQ消息队列笔记
Posted 知道什么是码怪吗？
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了RabbitMQ消息队列笔记相关的知识，希望对你有一定的参考价值。
目录
发布确认高级
在生产环境中由于一些不明原因，导致 RabbitMQ 重启，在 RabbitMQ 重启期间生产者消息投递失败，导致消息丢失，需要手动处理和恢复。如何才能进行 RabbitMQ 的消息可靠投递呢？特别是在比较极端的情况，RabbitMQ 集群不可用时，无法投递的消息该如何处理呢？
模拟消息丢失
目录结构
配置类新增代码
spring.rabbitmq.publisher-confirm-type = correlatedconfirmConfig类代码
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 ConfirmConfig
//交换机
public static final String CONFIRM_EXCHANGE_NAME = "confirm_exchange";
//队列
public static final String CONFIRM_QUEUE_NAME = "confirm_queue";
//RoutingKey
public static final String CONFIRM_ROUTING_KEY = "key1";
//声明交换机
@Bean
public DirectExchange confirmExchange()
return new DirectExchange(CONFIRM_EXCHANGE_NAME);
@Bean
public Queue confirmQueue()
return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
//绑定
@Bean
public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue,
@Qualifier("confirmExchange") DirectExchange confirmExchange)
return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);
MyCallBack类代码
@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback
@Autowired
private RabbitTemplate rabbitTemplate;
@PostConstruct
public void init()
rabbitTemplate.setConfirmCallback(this);
/*
* 交换机确认回调方法,发消息后，交换机接收到了就回调
* 1.1 correlationData：保存回调消息的ID及相关信息
* 1.2 b:交换机收到消息，为true
* 1.3 s:失败原因，成功为null
*
* 发消息，交换机接受失败，也回调
* 2.1 correlationData：保存回调消息的ID及相关信息
* 2.2 b:交换机没收到消息，为false
* 2.3 s:失败的原因
* */
@Override
public void confirm(CorrelationData correlationData, boolean b, String s)
String id = correlationData != null ? correlationData.getId() : "";
if (b)
log.info("交换机已经收到id为：的信息", id);
else
log.info("交换机未收到id为：的信息，失败的原因：", id, s);
Consumer类代码
@Slf4j
@Component
public class Consumer
@RabbitListener(queues = ConfirmConfig.CONFIRM_QUEUE_NAME)
public void receiveConfirmMessage(Message message)
String s = new String(message.getBody());
log.info("接收到的队列confirm.queue消息：", s);
ProducerController类代码
// 开始发消息 测试确认
@RestController
@Slf4j
@RequestMapping("/confirm")
public class ProducerController
@Autowired
private RabbitTemplate rabbitTemplate;
//发消息
@GetMapping("/sendMessage/message")
public void sendMessage(@PathVariable String message)
CorrelationData correlationData1 = new CorrelationData("1");
rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME
, ConfirmConfig.CONFIRM_ROUTING_KEY
, message + "key1", correlationData1);
log.info("发送消息内容：", message + " key1");
CorrelationData correlationData2 = new CorrelationData("2");
rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME
, ConfirmConfig.CONFIRM_ROUTING_KEY + "2"
, message + "key2", correlationData2);
log.info("发送消息内容：", message + " key2");
测试结果
localhost:8080/confirm/sendMessage/123
发送了两条消息给交换机，但是只有一个消息被接收，另一个消息丢失。
消息回退
在仅开启了生产者确认机制的情况下，交换机接收到消息后，会直接给消息生产者发送确认消息，如果发现该消息不可路由，那么消息会被直接丢弃，此时生产者是不知道消息被丢弃这个事件的。通过设置 mandatory 参数可以在当消息传递过程中不可达目的地时将消息返回给生产者，让生产者自己处理消息。
回退测试
配置类新增代码
spring.rabbitmq.publisher-returns = trueMycallBack类内容更新
@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback
@Autowired
private RabbitTemplate rabbitTemplate;
@PostConstruct
public void init()
rabbitTemplate.setConfirmCallback(this);
rabbitTemplate.setReturnsCallback(this);
@Override
public void confirm(CorrelationData correlationData, boolean b, String s)
String id = correlationData != null ? correlationData.getId() : "";
if (b)
log.info("交换机已经收到id为：的信息", id);
else
log.info("交换机未收到id为：的信息，失败的原因：", id, s);
//可以在当消息传递过程中不可达目的的时将消息返回给生产者
//只有不可达目的地的时候才可回退
@Override
public void returnedMessage(ReturnedMessage returnedMessage)
log.error("消息，被交换机退回，退回原因：，路由key:", new String(returnedMessage.getMessage().getBody())
, returnedMessage.getExchange()
, returnedMessage.getReplyText()
, returnedMessage.getRoutingKey());
测试结果
发送两条消息，不可路由的消息将会被退回。
备份交换机
备份交换机可以理解为 RabbitMQ 中交换机的“备胎”，当我们为某一个交换机声明一个对应的备份交换机时， 就是为它创建一个备胎，当交换机接收到一条不可路由消息时，将会把这条消息转发到备份交换机中，由 备份交换机来进行转发和处理，通常备份交换机的类型为 Fanout ，这样就能把所有消息都投递到与其绑定的队列中，然后我们在备份交换机下绑定一个队列，这样所有那些原交换机无法被路由的消息，就会都进入这个队列了。当然，我们还可以建立一个报警队列，用独立的消费者来进行监测和报警。
备份交换机测试
更新ConfirmConfig类代码
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 ConfirmConfig
//交换机
public static final String CONFIRM_EXCHANGE_NAME = "confirm_exchange";
//队列
public static final String CONFIRM_QUEUE_NAME = "confirm_queue";
//RoutingKey
public static final String CONFIRM_ROUTING_KEY = "key1";
//备份交换机
public static final String BACKUP_EXCHANGE_NAME = "backup_exchange";
//备份队列
public static final String BACKUP_QUEUE_NAME = "backup_queue";
//报警队列
public static final String WARNING_QUEUE_NAME = "warning_queue";
//声明交换机
@Bean
public DirectExchange confirmExchange()
return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE_NAME).durable(true)
.withArgument("alternate-exchange", BACKUP_EXCHANGE_NAME).build();
//声明队列
@Bean
public Queue confirmQueue()
return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
//绑定
@Bean
public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue,
@Qualifier("confirmExchange") DirectExchange confirmExchange)
return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);
//备份交换机
@Bean
public FanoutExchange backupExchange()
return new FanoutExchange(BACKUP_EXCHANGE_NAME);
//备份队列
@Bean
public Queue backupQueue()
return QueueBuilder.durable(BACKUP_QUEUE_NAME).build();
//报警队列
@Bean
public Queue warningQueue()
return QueueBuilder.durable(WARNING_QUEUE_NAME).build();
//备份交换机与备份队列绑定
@Bean
public Binding backupQueueBindingBackupExchange(@Qualifier("backupQueue") Queue backupQueue,
@Qualifier("backupExchange") FanoutExchange backupExchange)
return BindingBuilder.bind(backupQueue).to(backupExchange);
//备份交换机与报警队列绑定
@Bean
public Binding warningQueueBindingBackupExchange(@Qualifier("warningQueue") Queue backupQueue,
@Qualifier("backupExchange") FanoutExchange backupExchange)
return BindingBuilder.bind(backupQueue).to(backupExchange);
新增WarningConsumer类
// 报警消费者
@Component
@Slf4j
public class WarningConsumer
//接受报警消息
@RabbitListener(queues = ConfirmConfig.WARNING_QUEUE_NAME)
public void receiveWarningMsg(Message message)
String msg = new String(message.getBody());
log.error("报警发现不可路由消息：", msg);
测试结果
不可路由的消息被报警队列接收到了。输出的是报警发现不可路由消息，并不是消息回退。
mandatory参数与备份交换机可以一起使用时，如果两者同时开启，备份交换机优先级高。
优先队列
带有优先级的队列，处理消息时根据优先级排序，优先级高的先处理。类似于工厂或者商店的大订单先处理一个道理。
实现优先队列
队列实现优先级需要做的事情有如下事情：（1）队列需要设置为优先级队列，（2）消息需要设置消息的优先级，消费者需要等待消息已经发送到队列中才去消费，这样才有机会对消息进行排序。
队列设置优先级
Map<String, Object> params = new HashMap();
params.put("x-max-priority", 10);
channel.queueDeclare("hello", true, false, false, params);消息设置优先级
AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().priority(5).build();生产者
public class Producer
// 队列名称
public static final String QUEUE_NAME = "hello";
// 发消息
public static void main(String[] args) throws IOException, TimeoutException
// 创建一个连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 工厂IP连接RabbitMQ的队列
factory.setHost("192.168.193.147");
// 用户名
factory.setUsername("admin");
// 密码
factory.setPassword("123");
factory.setPort(5672);
// 创建连接
Connection connection = factory.newConnection();
// 获取信道
Channel channel = connection.createChannel();
Map<String, Object> arguments = new HashMap<>();
//官方允许是0-255之间，此处设置10，允许优化级范围为0-10，不要设置过大，浪费CPU与内存
arguments.put("x-max-priority", 10);
channel.queueDeclare(QUEUE_NAME, true, false, false, arguments);
// 发消息
for (int i = 0; i < 10; i++)
String message = "info" + i;
if (i == 5) //当i等于5时设置这一条消息优先级为5
AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().priority(5).build();
channel.basicPublish("", QUEUE_NAME, properties, message.getBytes(StandardCharsets.UTF_8));
else
channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));
System.out.println("消息发送完毕！");
消费者
public class Consumer
// 队列名称
public static final String QUEUE_NAME = "hello";
// 接受消息
public static void main(String[] args) throws IOException, TimeoutException, TimeoutException
// 创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.193.147");
factory.setUsername("admin");
factory.setPassword("123");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 声明 接受消息
DeliverCallback deliverCallback = (consumerTag, message) ->
System.out.println(new String(message.getBody()));
;
// 声明 取消消息
CancelCallback cancelCallback = consumer ->
System.out.println("消息消费被中断");
;
System.out.println("等待接收消息……");
channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
测试结果
如果我们先启动消费者，那么出现的结果将会按照顺序显示，因为没有出现消息的堆积，来一个就处理一个。
然而当我们先启动生产者，产生消息在信道当中堆积，那么处理时就会先按照优先级排序再进行处理。
以上是关于RabbitMQ消息队列笔记的主要内容，如果未能解决你的问题，请参考以下文章