RabbitMQ生产方式和解决消息可靠性投递及其他问题
Posted 幸福往往是摸得透彻
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RabbitMQ生产方式和解决消息可靠性投递及其他问题相关的知识,希望对你有一定的参考价值。
RabbitMQ生产方式和解决消息可靠性投递及其他问题
3种Exchange(交换机)与对应生产方式
- direct
将消息发送到与交换机绑定且对应routingkey的队列
生产方式有:Work queues(工作模式)、Routing(路由模式) - fanout
将消息发送到与交换机绑定的队列,不需要routingkey
生产方式有:Publish/Subscribe(订阅模式) - topic
将消息发送到与交换机绑定的队列,routingkey模糊匹配
生产方式有:Topics(主题模式)
Queue队列重载参数
durable:是否持久化,true 交换机或者队列会存到本地文件数据库,当mq重启,依然在,false的话,重启或者断电,就没了;默认true
autoDelete:是否自动删除,true当没有Consumer消费者时候,自动删除掉;默认false
exclusive:是否独占:true的话只有一个消费者监听这个队列。默认false
/**
* 定义一个direct队列
*
* @return
*/
@Bean
public Queue directQueue()
return new Queue(DIRECT_QUEUE);
/**
* 定义一个带参direct队列
*
* boolean durable 是否持久化
* boolean autoDelete 是否自动删除
* boolean exclusive 是否独占
* @return
*/
@Bean
public Queue directQueue2()
return new Queue(DIRECT_QUEUE, true, false, false);
RabbitMQ生产方式
-
Work queues(工作模式)
一个或者多个消费者共同消费一个队列中的消息(队列中的每一个消息只可能被其中一个消费者消费)
-
Routing(路由模式)
跟订阅模式类似,只不过在订阅模式的基础上加上了类型,订阅模式是分发到所有绑定到交换机的队列,路由模式
-
Publish/Subscribe(订阅模式)
生产者生产的消息,所有订阅过的消费者都能够接收到消息(队列中的每一个消息可以被多个消费者消费)
-
Topics(主题模式)
主题模式和路由模式很像,路由模式是精确匹配,主题模式是模糊匹配,* 匹配一个单词,# 匹配零个或者多个单词
消息投递流程
Producer(生产者)=>Channel(网络信道)=>Exchange(交换机)=>Queue(队列,到此会进行存储,即完成投递)=>Channel=>Consumer(消费者)
考虑以下问题
本文着重讲解消息可靠性投递
-
如何保证消息没有重复消费
保证消费者的幂等性
-
消息可靠性投递
生产端
confirm确认模式和return退回模式(一般是一起使用)
confirm确认模式(保证消息从生产者投递到交换机)
生产者消息投递后,如果Broker收到消息,则会给生产者一个应答(ack),生产者进行接收应答,用来确定这条消息是否正常的发送到Broker
实现如下:
1.配置文件:spring.rabbitmq.publisher-confirm-type=correlatedserver: port: 80 spring: rabbitmq: host: 192.168.31.6 port: 5672 username: linhy password: 123456 virtual-host: / # confirm 确认模式 publisher-confirm-type: correlated
2.消息发送类实现RabbitTemplate.ConfirmCallback接口,实现confirm方法
3.初始化RabbitTemplate,rabbitTemplate.setConfirmCallback(this);
4.发送消息new一个CorrelationData对象将主键ID放入,将该对象一起发送(confirm方法参数是一个消息唯一标识,即CorrelationData对象,所以需要将主键ID放入)
5.判断confirm方法ack参数进行补发操作
消息发送类具体代码示例public class RabbitMqServiceImpl implements RabbitMqService, RabbitTemplate.ConfirmCallback @Resource private AmqpTemplate amqpTemplate; @Resource private RabbitTemplate rabbitTemplate; /** * 初始化confirm 确认模式 */ @PostConstruct public void init() rabbitTemplate.setConfirmCallback(this); /** * String exchange 交换机名称 * String routingKey 路由Key * Object object 具体发送的消息 * * @param message */ @Override public void sendMessage(String message) CorrelationData correlationData = new CorrelationData(); correlationData.setId("12050"); rabbitTemplate.convertAndSend(RabbitMQConfig.DIRECT_EXCHANGE, RabbitMQConfig.ACK_DIRECT_ROUTING_KEY, message, correlationData);// 消息推送失败return /** * confirm只能保证消息从生产者投递到交换机后,进入队列前 * * @param correlationData 消息唯一标识 * @param ack 交换机是否成功收到消息 true成功 false失败 * @param cause 失败原因 */ @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) System.out.println("消息已被推送=========>" + correlationData); if (ack) System.out.println("交换机接受消息成功" + cause); else System.out.println("交换机接受消息失败" + cause); // 消息补发操作 System.out.println("失败id:" + correlationData.getId());
return退回模式(保证消息从交换机投递到队列)
用于处理交换机不存在或指定的路由key路由不到的情况
实现如下:
1.配置文件:publisher-returns: true(默认为false,即消息路由不可达时消息丢弃;true消息回退)server: port: 80 spring: rabbitmq: host: 192.168.31.6 port: 5672 username: linhy password: 123456 virtual-host: / # confirm 确认模式 publisher-confirm-type: correlated # return 退回模式 publisher-returns: true
2.消息发送类实现RabbitTemplate.ReturnCallback接口,实现returnedMessage方法
3.初始化RabbitTemplate,rabbitTemplate.setReturnCallback(this);
4.获取主键ID进行日志纪录或其他操作(returnedMessage方法参数是消息主体,生产者发送时可将主键ID作为消息主体发送)
消息发送类具体代码示例public class RabbitMqServiceImpl implements RabbitMqService, RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback @Resource private AmqpTemplate amqpTemplate; @Resource private RabbitTemplate rabbitTemplate; /** * 初始化confirm 确认模式和return 退回模式 */ @PostConstruct public void init() rabbitTemplate.setConfirmCallback(this); rabbitTemplate.setReturnCallback(this); /** * String exchange 交换机名称 * String routingKey 路由Key * Object object 具体发送的消息 * * @param message */ @Override public void sendMessage(String message) CorrelationData correlationData = new CorrelationData(); correlationData.setId("12050"); rabbitTemplate.convertAndSend(RabbitMQConfig.DIRECT_EXCHANGE, RabbitMQConfig.ACK_DIRECT_ROUTING_KEY, message, correlationData);// 消息推送失败return /** * confirm只能保证消息从生产者投递到交换机后,进入队列前 * * @param correlationData 消息唯一标识 * @param ack 交换机是否成功收到消息 true成功 false失败 * @param cause 失败原因 */ @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) System.out.println("消息已被推送=========>" + correlationData); if (ack) System.out.println("交换机接受消息成功" + cause); else System.out.println("交换机接受消息失败" + cause); // 消息补发操作 System.out.println("失败id:" + correlationData.getId()); /** * returnedMessage能保证消息从交换机投递到队列 * * @param message 消息主体 * @param replyCode 返回code * @param replyText 返回信息 * @param exchange 交换机 * @param routingKey 路由key */ @Override public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) System.out.println("消息推送失败,return方法被执行..."); System.out.println("消息主体:" + new String(message.getBody())); System.out.println("replyCode:" + replyCode); System.out.println("replyText:" + replyText); System.out.println("exchange:" + exchange); System.out.println("routingKey:" + routingKey);
消费端
消费端手动ACK确认
实现如下:
1.确认方法改为手动确认,配置文件:acknowledge-mode: manualserver: port: 81 spring: rabbitmq: host: 192.168.31.6 port: 5672 username: admin password: admin virtual-host: / # 消费端手动ack确认队列消息 listener: simple: acknowledge-mode: manual
2.消费类增加Channel对象参数和deliveryTag参数
3.channel.basicAck(deliveryTag, true)确认队列消息,channel.拒签,是否回到队列
消息接收类具体代码示例
Servicepublic interface RabbitMqService /** * 消费端手动ack确认队列消息 * * @param message 消息 * @param channel 管道 * @param deliveryTag 标号 */ public void receiveSubMessageAck(String message, Channel channel, long deliveryTag);
Service实现
public class RabbitMqServiceImpl implements RabbitMqService @Resource private AmqpTemplate amqpTemplate; @Override @RabbitListener(queues = RabbitMQConfig.ACK_QUEUE) public void receiveSubMessageAck(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) try System.out.println("接收到的mq消息:" + message); // 处理业务 System.out.println("开始处理业务=========>"); // 消费者手动确认队列消息(批量确认) if (deliveryTag == 5) channel.basicAck(deliveryTag, true);// long deliveryTag 标号(当前消息在队列中的的索引), boolean multiple 是否批量处理 // 消费者手动确认队列消息(单个确认,低效) // channel.basicAck(deliveryTag, true); catch (Exception e) // 报错,消费者进行拒签 e.printStackTrace(); try channel.basicNack(deliveryTag, false, false);//long deliveryTag 标号, boolean multiple 是否批量处理, boolean requeue 是否回到队列(false直接丢弃) catch (IOException ioException) ioException.printStackTrace();
-
怎么保证消息传递的顺序
开启多个队列,消息投递到同一个队列,一个消费者消费
-
一致性问题
一致性问题即事务问题,可使用nacos注册中心+seata解决分布式事务问题,下次更新
以上是关于RabbitMQ生产方式和解决消息可靠性投递及其他问题的主要内容,如果未能解决你的问题,请参考以下文章
消息队列专题(高级特性篇):RabbitMQ 如何保证消息的可靠性投递传输和消费
消息队列专题(高级特性篇):RabbitMQ 如何保证消息的可靠性投递传输和消费
消息队列专题(高级特性篇):RabbitMQ 如何保证消息的可靠性投递传输和消费