SpringBoot集成RabbitMq，RabbitMq消费与生产，消费失败重发机制，发送签收确认机制

RabbitMq消费与生产，消费失败重发机制，发送确认机制，消息发送结果回执

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#支持消息发送失败返回队列,默认为falsepublisher-returns:true# 消费者listener:type: simple simple:#自动签收auto 手动 manualacknowledge-mode: auto #个字段一定要设置成 false 不然无法消费的数据不会进入死信队列的default-requeue-rejected:falseprefetch:1#限制每次发送一条数据max-concurrency:1#启动消费者最大数量concurrency:1#同一个队列启动几个消费者retry:enabled:true#是否支持重试max-attempts:3# 最大重试次数，默认为3initial-interval: 30s # 重试间隔时间，默认1000(单位毫秒)max-interval: 120s # 重试最大间隔# 时间间隔的乘子，下一次间隔的时间=间隔时间 × 乘子，但最大不超过重试最大间隔multiplier:1 ``````

• RabbitMq生产者，队列，交换通道配置，消费者示例Exchange 交换机配置@ComponentpublicclassDnfxExchangeConfig{@AutowiredRabbitMqConfig rabbitMqConfig;/** * topic交换机起名 * 如果rabbitmq设置的类型是topic 就用topic类型的Exchange * * @return */@BeanTopicExchangednfxOrderExchange(){returnnewTopicExchange(rabbitMqConfig.getFxexchange());}}队列queue配置@ComponentpublicclassDnfxQueueConfig{@AutowiredRabbitMqConfig rabbitMqConfig;/** * 队列起名 * * @return */@BeanpublicQueuednfxOrderQueue(){Map<String,Object> argsMap =newHashMap<String,Object>();//队列优先级 argsMap.put("x-max-priority", 5);//true 是否持久 returnnewQueue(rabbitMqConfig.getFxqueue(),true,false,false, argsMap);}}将队列和交换机绑定, 并设置用于匹配键@ComponentpublicclassDnfxRoutingConfig{@AutowiredRabbitMqConfig rabbitMqConfig;@AutowiredDnfxQueueConfig queueConfig;@AutowiredDnfxExchangeConfig exchangeConfig;/** * 绑定：将队列和交换机绑定, 并设置用于匹配键 myDirectRouting * * @return */@BeanBindingbindingOrderRouting(){returnBindingBuilder.bind(queueConfig.dnfxOrderQueue()).to(exchangeConfig.dnfxOrderExchange()).with(rabbitMqConfig.getFxrouting());}}配置加载@Configuration@ConfigurationProperties(prefix ="xx.mq")@DatapublicclassRabbitMqConfig{privateString fxqueue;privateString fxexchange;privateString fxrouting;}RabbitTemplate@ConfigurationpublicclassDnfxRabbitMqConfig{@AutowiredRabbitMqConfig rabbitMqConfig;@BeanpublicRabbitTemplaterabbitTemplate(ConnectionFactory connectionFactory){returnnewRabbitTemplate(connectionFactory);}}生产者@ComponentpublicclassDemoTestProduce{@AutowiredprivateRabbitTemplate rabbitTemplate;@AutowiredRabbitMqConfig rabbitMqConfig;publicvoidsendDemoMsg(){String message ="测试消息发送"; rabbitTemplate.convertAndSend(rabbitMqConfig.getFxexchange(), rabbitMqConfig.getFxrouting(), message);}}消费者@ComponentpublicclassDnfxAliBbMessageListener{privatefinalstaticLogger logger =LoggerFactory.getLogger(DnfxAliBbMessageListener.class);@RabbitListener(containerFactory ="rabbitListenerContainerFactory", queues ="${xx.mq.fxqueue}")publicvoidlistenSimpleQueueMessage(String msg)throwsIOException{ logger.info("接收到的1688回执消息：{}", msg);}}

2. RabbitMq消息确认机制

• 消息确认机制分自动确认，和手动确认消息确认签收配置# 消费者listener:type: simple simple:#自动签收 auto 手动 manualacknowledge-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}")publicvoidlistenSimpleQueueMessage(Message message,Channel channel)throwsIOException{String msgBody =newString(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# 最大重试次数，默认为3initial-interval: 30s # 重试间隔时间，默认1000(单位毫秒)max-interval: 120s # 重试最大间隔# 时间间隔的乘子，下一次间隔的时间=间隔时间 × 乘子，但最大不超过重试最大间隔multiplier:1

• 消息重发如何触发 消息重发机制，与消息确认签收机制是两种不同的机制，这个概念不要弄混了，消息确认签收机制亦可以将消息重新放入队列进行二次消费 消息重发机制，在消费者进行消费时，如果rabbitmq开启了消息重发机制，当消费者处理消息时候抛出了异常，即触发消息重发机制，注意，处理消息逻辑不要用try-catch捕捉异常，异常被捕捉后，会抛出异常信息，但不会影响代码正常执行，amqp aop会视为正常消费，不会触发重发机制。@RabbitListener(containerFactory ="rabbitListenerContainerFactory", queues ="${zcwl.mq.fxqueue}")publicvoidlistenSimpleQueueMessage(Message message,Channel channel)throwsIOException{String msgBody =newString(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 参数，队列注册，以及队列交换机绑定与普通队列一样即可

@BeanCustomExchangednfxOrderDelayExchange(){Map<String,Object> args =newHashMap<String,Object>();
        args.put("x-delayed-type","topic");returnnewCustomExchange(rabbitMqConfig.getFxOrderDelayExchange(),"x-delayed-message",true,false, args);}

测试发送延时消息方法，队列监听与普通消息一样即可

publicvoidsendDelayMsg(){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即可，与 创建的交换机类型保持一致