RabbitMq消费与生产，消费失败重发机制，发送确认机制，消息发送结果回执
- 1. RabbitMq集成spring boot
- RabbitMq集成依赖
- RabbitMq配置
- RabbitMq生产者，队列，交换通道配置，消费者示例
- 2. RabbitMq消息确认机制
- 消息确认机制分自动确认，和手动确认
- 3. 消息重发机制
- 消息重发配置
- 消息重发如何触发
- 4. 延时消息队列
- 5. 接收返回结果队列
- 尚未研究后续用到补充
- 6. 遇到的报错
- 启动报错 Channel shutdown: channel error; protocol method:
1. RabbitMq集成spring boot
-
RabbitMq集成依赖
这里spring-boot依赖版本为2.3.7版本，RabbitMq集成amqp包，版本在spring-boot中有涵盖，不单独指明版本了。
<parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.3.7.RELEASE</version> </parent> <dependencyManagement> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-dependencies</artifactId> <version>2.3.7.RELEASE</version> <type>pom</type> </dependency> </dependencies> </dependencyManagement> <dependencies> <!-- rabbitMQ --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> </dependencies> -
RabbitMq配置
spring: rabbitmq: # 基础项 host: ip地址 port: 端口 username: 用户名 password: 密码 # virtualhost需要提前在MQ的Web管理界面里手动创建，或者配置默认host"/" virtual-host: / # 生产者 #确认消息已发送到交换机(Exchange) publisher-confirm-type: correlated #开启消息发送确认机制，默认为false #如果没有本条配置信息，当消费者收到生产者发送的消息后，生产者无法收到确认成功的回调信息 publisher-confirms: true #支持消息发送失败返回队列,默认为false publisher-returns: true # 消费者 listener: type: simple simple: #自动签收auto 手动 manual acknowledge-mode: auto #个字段一定要设置成 false 不然无法消费的数据不会进入死信队列的 default-requeue-rejected: false prefetch: 1 #限制每次发送一条数据 max-concurrency: 1 #启动消费者最大数量 concurrency: 1 #同一个队列启动几个消费者 retry: enabled: true #是否支持重试 max-attempts: 3 # 最大重试次数，默认为3 initial-interval: 30s # 重试间隔时间，默认1000(单位毫秒) max-interval: 120s # 重试最大间隔 # 时间间隔的乘子，下一次间隔的时间=间隔时间 × 乘子，但最大不超过重试最大间隔 multiplier: 1 ``` -
RabbitMq生产者，队列，交换通道配置，消费者示例
Exchange 交换机配置
@Component public class DnfxExchangeConfig { @Autowired RabbitMqConfig rabbitMqConfig; /** * topic交换机起名 * 如果rabbitmq设置的类型是topic 就用topic类型的Exchange * * @return */ @Bean TopicExchange dnfxOrderExchange() { return new TopicExchange(rabbitMqConfig.getFxexchange()); } }队列queue配置
@Component public class DnfxQueueConfig { @Autowired RabbitMqConfig rabbitMqConfig; /** * 队列起名 * * @return */ @Bean public Queue dnfxOrderQueue() { Map<String, Object> argsMap = new HashMap<String, Object>(); //队列优先级 argsMap.put("x-max-priority", 5); //true 是否持久 return new Queue(rabbitMqConfig.getFxqueue(), true, false, false, argsMap); } }将队列和交换机绑定, 并设置用于匹配键
@Component public class DnfxRoutingConfig { @Autowired RabbitMqConfig rabbitMqConfig; @Autowired DnfxQueueConfig queueConfig; @Autowired DnfxExchangeConfig exchangeConfig; /** * 绑定：将队列和交换机绑定, 并设置用于匹配键 myDirectRouting * * @return */ @Bean Binding bindingOrderRouting() { return BindingBuilder.bind(queueConfig.dnfxOrderQueue()).to(exchangeConfig.dnfxOrderExchange()).with(rabbitMqConfig.getFxrouting()); } }配置加载
@Configuration @ConfigurationProperties(prefix = "xx.mq") @Data public class RabbitMqConfig { private String fxqueue; private String fxexchange; private String fxrouting; }RabbitTemplate
@Configuration public class DnfxRabbitMqConfig { @Autowired RabbitMqConfig rabbitMqConfig; @Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { return new RabbitTemplate(connectionFactory); } }生产者
@Component public class DemoTestProduce { @Autowired private RabbitTemplate rabbitTemplate; @Autowired RabbitMqConfig rabbitMqConfig; public void sendDemoMsg() { String message = "测试消息发送"; rabbitTemplate.convertAndSend(rabbitMqConfig.getFxexchange(), rabbitMqConfig.getFxrouting(), message); } }消费者
@Component public class DnfxAliBbMessageListener { private final static Logger logger = LoggerFactory.getLogger(DnfxAliBbMessageListener.class); @RabbitListener(containerFactory = "rabbitListenerContainerFactory", queues = "${xx.mq.fxqueue}") public void listenSimpleQueueMessage(String msg) throws IOException { logger.info("接收到的1688回执消息：{}", msg); } }
2. RabbitMq消息确认机制
-
消息确认机制分自动确认，和手动确认
消息确认签收配置
# 消费者 listener: type: simple simple: #自动签收 auto 手动 manual acknowledge-mode: auto消息确认签收机制不过多赘述，网上有大把说明，这里简单描述一下，以及记录一下个人使用心得。
acknowledge-mode: auto 配置为自动签收时候，消息送达至消费者手上后，Mq自动签收，并移除消息出消息队列。
acknowledge-mode: false 配置为手动签收时候，消息送达至消费者后，消费者需要手动触发签收动作，如果消费者没有发送ACK消息，RabbitMQ服务器就会认为该消息还没有被消费，会将该消息重新发送给其他消费者。例如下图，手动签收模式，没有主动向MQ发送签收讯息，那么当前消费的这条消息会被标记为Unacked
关于签收 确认机制可以参考 https://blog.csdn.net/qq_42331185/article/details/131696949 ，这里贴部分这个博主的结论@RabbitListener(containerFactory = "rabbitListenerContainerFactory", queues = "${xx.mq.fxqueue}") public void listenSimpleQueueMessage(Message message, Channel channel) throws IOException { String msgBody = new String(message.getBody()); logger.info("接收到的1688回执消息：{}", msgBody); long deliveryTag = message.getMessageProperties().getDeliveryTag(); channel.basicReject(deliveryTag, false);deliveryTag：消息传递标签，格式为序列号，必须使用这个标签，不然信道会关闭，详情下面会说到
multiple：为true则表示序号deliverTag之前的消息均被确认或拒绝（basicNack），false表示当前消息。为true的时候就可以做到批量确认
requeue：为true表示，失败的消息将会重新排队，不会丢弃或者死信，为false则表示丢弃1、消息成功签收 basicAck(deliveryTag，multiple) channel.basicAck(message.getEnvelope().getDeliveryTag(), false); 2、失败确认 basicNack(deliveryTag，multiple，requeue) channel.basicNack(message.getEnvelope().getDeliveryTag(),false, true); 3、失败确认：basicReject(deliveryTag，requeue) channel.basicReject(message.getEnvelope().getDeliveryTag(), true);注：关于以上手动确认multiple属性为true时，批量确认这个元素个人未进行验证，失败确认requeue为true时，当前消息会重新丟至MQ队列中，等待下次消费（已验证）
关于消息确认机制，自动确认可能导致消息丢失，如果单条消息发送至消费者后，消费者处理报错，最多触发消息重发机制，重发达到重发上限后，便会抛弃此消息，造成消息丢失。
手动确认签收，千万不要在cath中或者final中进行失败重发签收，即basicNack basicReject 失败签收时requeue 为true，否则当前消息若真为异常消息，此消息会一直消费，失败签收，重新排队，进行循环，导致消息积压或者资源浪费
3. 消息重发机制
-
消息重发配置
注意： 如果遗漏 max-interval multiplier两个属性，消息重发机制仍会生效，但是重发间隔时间为默认10秒重发， initial-interval 重发间隔时间将不会生效。此处已验证，尚未确认是bug或者本身就是联动配置
# 消费者 listener: type: simple simple: retry: enabled: true #是否支持重试 max-attempts: 3 # 最大重试次数，默认为3 initial-interval: 30s # 重试间隔时间，默认1000(单位毫秒) max-interval: 120s # 重试最大间隔 # 时间间隔的乘子，下一次间隔的时间=间隔时间 × 乘子，但最大不超过重试最大间隔 multiplier: 1 -
消息重发如何触发
消息重发机制，与消息确认签收机制是两种不同的机制，这个概念不要弄混了，消息确认签收机制亦可以将消息重新放入队列进行二次消费
消息重发机制，在消费者进行消费时，如果rabbitmq开启了消息重发机制，当消费者处理消息时候抛出了异常，即触发消息重发机制，注意，处理消息逻辑不要用try-catch捕捉异常，异常被捕捉后，会抛出异常信息，但不会影响代码正常执行，amqp aop会视为正常消费，不会触发重发机制。@RabbitListener(containerFactory = "rabbitListenerContainerFactory", queues = "${zcwl.mq.fxqueue}") public void listenSimpleQueueMessage(Message message, Channel channel) throws IOException { String msgBody = new String(message.getBody()); logger.info("接收到的1688回执消息：{}", msgBody); long deliveryTag = message.getMessageProperties().getDeliveryTag(); //此处会抛出异常 int a = 1 / 0; //确认签收机制为手动签收，一定要进行签收，否则触发重发机制后，此条消息仍会被标记为unacked channel.basicReject(deliveryTag, false);
4. 延时消息队列
延时消息队列需要配合RabbitMq延时消息队列插件使用，安装延时消息队列插件此处不赘述，网上搜一大把
延时消息队列创建队列以及绑定key时没什么特殊的，在创建exchange交换机时，需要注意选项，如下图所示即可。
x-delayed-type = redirect 如果不能创建，报错时，那么=topic也是可以的
注册exchange交换机时候，注意给入 x-delayed-type 参数，队列注册，以及队列交换机绑定与普通队列一样即可
@Bean
CustomExchange dnfxOrderDelayExchange() {
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-delayed-type", "topic");
return new CustomExchange(rabbitMqConfig.getFxOrderDelayExchange(), "x-delayed-message", true, false, args);
}
测试发送延时消息方法，队列监听与普通消息一样即可
public void sendDelayMsg() {
System.out.println(LocalDateTime.now() + "：发送延时消息");
String message = "这里是测试延时发送消息";
this.rabbitTemplate.convertAndSend(rabbitMqConfig.getFxOrderDelayExchange(), rabbitMqConfig.getFxOrderDelayRouting(), message, message1 -> {
//delay的单位是毫秒
message1.getMessageProperties().setDelay(1000 * 60);
return message1;
});
}
5. 接收返回结果队列
尚未研究后续用到补充
6. 遇到的报错
-
启动报错 Channel shutdown: channel error; protocol method:
报错详情：
Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'type' for exchange 'fx-bb-msg-exchange' in vhost '/': received 'direct' but current is 'topic', class-id=40, method-id=10)此错误为注册交换机时候抛出的错误，错误信息为注册交换机的属性，与RabbitMq已经创建好的交换机属性不一致，程序试图修改属性报错。
错误示范：
当前exchange交换机创建时候，创建的类型Type为topic类型，在注册exchange交换机时，返回的却是DirectExchange，那么系统便会尝试修改属性，从而引发报错
修复方式：
创建时，返回TopicExchange即可，与 创建的交换机类型保持一致
