提示:文章写完后,目录可以自动生成,如何生成可参考右边的帮助文档
前言
在使用rabbitmq时,会因为各种原因(网络波动,系统宕机,程序异常等)导致消息发送失败。rabbitmq也提供了相应的处理机制。
提示:以下是本篇文章正文内容,下面案例可供参考
一、rabbitmq消息发送失败处理机制
生产法发送失败
配置回调器。
yml配置开启确认和返回机制
confirm:发送给exchange时的回调,不管是否成功发送给队列。
return:消息没有发送给exchange时的回调。
#成功发送到exchange时的回调
spring.rabbitmq.publisher-confirm-type=correlated
#exchange未发送到队列时回调
spring.rabbitmq.publisher-returns=true
回调函数配置方式
@Component
@Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {
/**
* 交换机不管是否收到消息的一个回调方法
*
* @param correlationData 消息相关数据
* @param ack 交换机是否收到消息
* @param cause 未收到消息的原因
*/
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
String id = correlationData != null ? correlationData.getId() : "";
if (ack) {
log.info("交换机已经收到 id 为:{}的消息", id);
} else {
//重发处理,可以入库,死信队列处理....
log.info("交换机还未收到 id 为:{}消息,原因:{}", id, cause);
}
}
//当消息无法路由的时候触发回调方法
@Override
public void returnedMessage(ReturnedMessage returned) {
//重发处理,可以入库,死信队列处理....
log.error("消息:{},被交换机 {} 退回,原因:{},路由key:{},code:{}",
new String(returned.getMessage().getBody()), returned.getExchange(),
returned.getReplyText(), returned.getRoutingKey(),
returned.getReplyCode());
}
}
rabbitmqTemplate在启动时注入:
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private RabbitTemplate.ConfirmCallback confirmCallback;
@Autowired
private RabbitTemplate.ReturnsCallback returnsCallback;
//依赖注入 rabbitTemplate 之后再设置它的回调对象
@PostConstruct
public void init() {
rabbitTemplate.setConfirmCallback(confirmCallback);
rabbitTemplate.setReturnsCallback(returnsCallback);
}
消费者消费失败后处理
通过自动ack+retry配置+私信队列方式实现
spring:
rabbitmq:
host: 127.0.0.1
port: 5672
username: guest
password: guest
listener:
simple:
# acknowledge-mode: manual # 配置该消费者的ack方式为手动
acknowledge-mode: auto # 配置该消费者的ack方式为自动
default-requeue-rejected: false
#设置消费失败后重发
retry:
#重发次数
max-attempts: 3
#开启重发
enabled: true
# 重试间隔(ms)
initial-interval: 5000
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = QUEUE_NAME1, durable = “true”, autoDelete = “false”,
arguments = {@Argument(name = “x-dead-letter-exchange”, value = “dead-exchange”),
@Argument(name = “x-dead-letter-routing-key”, value = “dead-routing-key”),
@Argument(name = “x-message-ttl”, value = “1000”,type = “java.lang.Long”)
}),
exchange = @Exchange(value = “first_exchange”, type = ExchangeTypes.DIRECT),
key = “queue_one_key1”))
public void handleMessage1(Message message, Channel channel) throws IOException {
log.info(“OrderConsumer handleMessage {} , error:”, message);
//模拟消费异常,自动进入私信队列
throw new RuntimeException(“抛出异常,模拟消费失败,触发spring-retry”);
}
/**
* 死信队列消费者
*
* @param data
* @param channel
* @throws Exception
*/
@RabbitListener(queues = "dead-queue")
public void consumeDL(String data, Channel channel) throws Exception {
//处理消费失败的消息
log.info(">>>> 死信队列消费 tag = {},消息内容 : {}", data);
// channel.basicNack(tag, false, false);
}
通过手动ack+私信队列实现(不要配置retry!!!)文章来源:https://www.toymoban.com/news/detail-406881.html
spring.rabbitmq.publisher-returns=true
spring.rabbitmq.listener.simple.acknowledge-mode=manual
@RabbitListener(queues = CONFIRM_QUEUE_NAME)
public void receiveMsg(String data, Message message, Channel channel) throws IOException {
String msg = new String(message.getBody());
log.info("接受到队列 confirm.queue 消息:{}", msg);
// throw new RuntimeException("抛出异常,模拟消费失败,触发spring-retry");
//模拟消费失败,重发n次后仍然失败。调用basicNack 抛给私信队列处理
channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
}
//接收消息
@RabbitListener(queues = "QD")
public void receiveD(Message message, Channel channel) throws IOException {
String msg = new String(message.getBody());
log.info("当前时间:{},收到死信队列信息{}", new Date().toString(), msg);
}
总结
rabbitmq消息失败处理需要谨慎对待,因为容易产生资源消耗殆尽的问题!!!文章来源地址https://www.toymoban.com/news/detail-406881.html
到了这里,关于rabbitmq消息异常处理的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!