本文共 8082 字,大约阅读时间需要 26 分钟。
文章目录
1. 消息确定机制及其配置
RabbitMq消费者的消息确定机制:
NONE
:无应答,rabbitmq默认consumer正确处理所有请求。AUTO
:consumer自动应答,处理成功(注意:此处的成功确认是没有发生异常)发出ack,处理失败发出nack。rabbitmq发出消息后会等待consumer端应答,只有收到ack确定信息后才会将消息在rabbitmq清除掉。收到nack异常信息的处理方法由setDefaultRequeueReject()方法设置,这种模式下,发送错误的消息可以恢复。MANUAL
:基本等同于AUTO模式,区别是需要人为调用方法确认。
在配置文件中:
spring: profiles: active: test rabbitmq: host: localhost port: 5672 username: guest password: guest template: receive-timeout: 2000 reply-timeout: 2000 listener: simple: concurrency: 1 max-concurrency: 3 # 消费者预取1条数据到内存,默认为250条 prefetch: 1 # 确定机制 acknowledge-mode: manual
可以进行配置,也可以在代码中进行配置:
@Configurationpublic class RabbitConfig { @Autowired private CachingConnectionFactory connectionFactory; @Bean(name = "singleListenerContainer") public SimpleRabbitListenerContainerFactory listenerContainerFactory() { SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); /* setConnectionFactory:设置spring-amqp的ConnectionFactory。 */ factory.setConnectionFactory(connectionFactory); factory.setConcurrentConsumers(1); factory.setMaxConcurrentConsumers(10); factory.setPrefetchCount(1); factory.setDefaultRequeueRejected(true); //手动确认。 factory.setAcknowledgeMode(AcknowledgeMode.MANUAL); return factory; }}
2. 自动确认和手动确认细节
2.1 测试代码配置
队列配置:
@Configurationpublic class RabbitMqConfig { // 简单的声明一个队列 @Bean public Queue kinsonQueue1() { return new Queue("kinson1"); }}
生产者配置:
@RestControllerpublic class RabbitMQController { @Autowired private RabbitTemplate rabbitTemplate; //直接向队列中发送数据 @GetMapping("send") public String send() { for (int i = 0; i < 10; i++) { String content = "Date:" + System.currentTimeMillis(); content = content + ":::" + i; rabbitTemplate.convertAndSend("kinson1",i+""); } return "success"; }}
消费者配置
@Component@Slf4jpublic class CustomerRev { //消费者处理消息缓慢 @RabbitListener(queues = { "kinson1"}) public void receiver3(Message msg, Channel channel) { try { //打印数据 String message = new String(msg.getBody(), StandardCharsets.UTF_8); log.info("【开始】:{}",message); Thread.sleep(30000); log.info("【结束】:{}", message); } catch (Exception e) { log.error("错误信息:{}", e.getMessage()); } }}
2.2 自动确认
生产者产生10笔消息,自动确认模式下,消息处理成功,消费者才会去获取下一笔消息;消息处理抛出异常,那么将会消息重回队列。
注:此时的prefetch=1
,该消费者unack的消息只有一条。其余9条均在MQ中。
若是设置
prefetch=250
(默认),那么消费者会预期10条消息,unack的消息是10条,但是日志中依旧只是消费了一条消息。
【开始】:Date:1604557983496:::0
**总结:**对于自动确认来说,当方法没有异常执行完毕后,会对MQ发出ACK;若方法出现异常,会对MQ发出nack,消息会重回队列。要分清哪些是可以恢复的异常,哪些是不可以恢复的异常。不可恢复的异常,在消费者代码中捕获异常,并记录日志表或放入死信队列。可恢复的异常,那么放入业务队列中重试。
2.3 手动确认
2.3.1 常用API
-
ack表示确认消息。multiple:false只确认该delivery_tag的消息,true确认该delivery_tag的所有消息channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false);
-
Reject表示拒绝消息。requeue:false表示被拒绝的消息是丢弃;true表示重回队列channel.basicReject(msg.getMessageProperties().getDeliveryTag(),false);
-
nack表示拒绝消息。multiple表示拒绝指定了delivery_tag的所有未确认的消息,requeue表示不是重回队列channel.basicNack(msg.getMessageProperties().getDeliveryTag(),false,false);
2.3.2 代码案例
案例:消费者只有发送ACK
后才会开始消费下一条消息。
@Component@Slf4jpublic class CustomerRev { //消费者处理消息缓慢 @RabbitListener(queues = { "kinson1"}) public void receiver3(Message msg, Channel channel) { try { //打印数据 String message = new String(msg.getBody(), StandardCharsets.UTF_8); log.info("【开始】:{}",message); if("0".equals(message)){ Thread.sleep(1000); }else { Thread.sleep(300000000L); } log.info("【结束】:{}", message); //ack表示确认消息。multiple:false只确认该delivery_tag的消息,true确认该delivery_tag的所有消息 channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false); } catch (Exception e) { log.error("错误信息:{}", e.getMessage()); } }}
3. 消息重回队列
自动确认模式下抛出异常,和手动确认模式下nack消息都可以重回队列。
3.1 默认—消息nack后重回队头
@Component@Slf4jpublic class CustomerRev { //消费者处理消息缓慢 @RabbitListener(queues = { "kinson1"}) public void receiver3(Message msg, Channel channel) { try { //打印数据 String message = new String(msg.getBody(), StandardCharsets.UTF_8); log.info("【开始】:{}",message); if("0".equals(message)){ throw new RuntimeException("0的消息消费异常"); } log.info("【结束】:{}", message); //ack表示确认消息。multiple:false只确认该delivery_tag的消息,true确认该delivery_tag的所有消息 channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false); } catch (Exception e) { throw new RuntimeException(e); } }}
(自动ACK下)当消息为0时,抛出异常,生产者产生10笔消息。
代码轮询打印0的消息的消费异常,其他消息均不消费。
当消息回滚到消息队列时,这条消息不会回到队列尾部,而是仍是在队列头部,这时消费者会立马又接收到这条消息进行处理,接着抛出异常,进行 回滚,如此反复进行。这种情况会导致消息队列处理出现阻塞,消息堆积,导致正常消息也无法运行。
3.2 解决方案—消息重发送到队尾
//消费者处理消息缓慢 @RabbitListener(queues = { "kinson1"}) public void receiver3(Message msg, Channel channel) throws IOException { try { //打印数据 String message = new String(msg.getBody(), StandardCharsets.UTF_8); log.info("【开始】:{}",message); if("0".equals(message)){ throw new RuntimeException("0的消息消费异常"); } log.info("【结束】:{}", message); //ack表示确认消息。multiple:false只确认该delivery_tag的消息,true确认该delivery_tag的所有消息 channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false); } catch (Exception e) { //捕获异常后,重新发送到指定队列,自动ack不抛出异常即为ack channel.basicPublish(msg.getMessageProperties().getReceivedExchange(), msg.getMessageProperties().getReceivedRoutingKey(), MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBody()); } }
手动确认的模式:
//手动进行应答channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);//重新发送消息到队尾channel.basicPublish(message.getMessageProperties().getReceivedExchange(),message.getMessageProperties().getReceivedRoutingKey(), MessageProperties.PERSISTENT_TEXT_PLAIN,JSON.toJSONBytes(new Object()));
这个不会导致其他消息被阻塞消费,但是会大量的抛出:
3.3 如何处理异常消息
如果一个消息体本身有误,会导致该消息体,一直无法进行处理,而服务器中刷出大量无用日志。解决这个问题可以采取两种方案:
1.一种是对于日常细致处理,分清哪些是可以恢复的异常,哪些是不可以恢复的异常。对于可以恢复的异常我们采取第三条中的解决方案,对于不可以处理的异常,我们采用记录日志,直接丢弃该消息方案。
2.另一种是我们对每条消息进行标记,记录每条消息的处理次数,当一条消息,多次处理仍不能成功时,处理次数到达我们设置的值时,我们就丢弃该消息,但需要记录详细的日志。
将业务队列绑定死信队列,当消息被丢弃后,进入到死信队列(代码修复后监听死信队列补偿消息)。可以避免我们手动的恢复消息。
重试机制(自动确认模式):
@Component@Slf4jpublic class CustomerRev { @RabbitListener(queues = { "kinson1"}) public void receiver3(Message msg, Channel channel) throws IOException { try { //打印数据 String message = new String(msg.getBody(), StandardCharsets.UTF_8); log.info("【开始】:{}",message); if("0".equals(message)){ throw new RuntimeException("0的消息消费异常"); } log.info("【结束】:{}", message); } catch (Exception e) { //捕获异常后,重新发送到指定队列,自动确认不抛出异常即为ack Integer retryCount; Mapheaders = msg.getMessageProperties().getHeaders(); if(!headers.containsKey("retry-count")){ retryCount=0; }else { retryCount = (Integer)headers.get("retry-count"); } //判断是否满足最大重试次数(重试3次) if(retryCount++<3) { headers.put("retry-count",retryCount); //重新发送到MQ中 AMQP.BasicProperties basicProperties = new AMQP.BasicProperties().builder().contentType("text/plain").headers(headers).build(); channel.basicPublish(msg.getMessageProperties().getReceivedExchange(), msg.getMessageProperties().getReceivedRoutingKey(), basicProperties, msg.getBody()); } } }}
每次收到消息后,修改消息的header信息。来记录重试的次数。
推荐阅读
转载地址:https://blog.csdn.net/qq_29595463/article/details/109527115 如侵犯您的版权,请留言回复原文章的地址,我们会给您删除此文章,给您带来不便请您谅解!