专业的网站制作公司地址,网站加盟代理,360开户推广,北京各大公司名称RabbitMQ消息的可靠性
一 生产者的可靠性
生产者重试 有时候由于网络问题#xff0c;会出现连接MQ失败的情况#xff0c;可以配置重连机制 注意#xff1a;SpringAMQP的重试机制是阻塞式的#xff0c;重试等待的时候#xff0c;当前线程会等待。
spring:rabbitmq:conne…RabbitMQ消息的可靠性
一 生产者的可靠性
生产者重试 有时候由于网络问题会出现连接MQ失败的情况可以配置重连机制 注意SpringAMQP的重试机制是阻塞式的重试等待的时候当前线程会等待。
spring:rabbitmq:connection-timout: 1s #设置MQ的连接超时时间templete:retry:enabled: true #开启超时重试机制initial-interval: 100ms #失败后的初始等待时间multipier: 1 #失败后下次的等待时长倍数 下次等待时长initial-interval*multipiermax-attempts: 3 #最大重试次数生产者确认 1在生产者服务的yaml文件中配置一下内容
spring:rabbitmq:publisher-confirm-type: correlated #开启publisher confirm机制并设置为MQ异步回调方式返回回执信息publisher-returns: true #开启publisher return机制2配置return-callback
Slf4j
Configuration
public class CommonConfig implements ApplicationContexAware{Overridepublic void setApplicationContext(ApplicationContent applicationContext){// 获取MQRabbitTemplate rabbitTemplate applicationContext.getBean(RabbitTemplate.class);// 设置returnCallbackrabbitTemplate.setReturnCallback((message,replyCode,replyText,exchange,routingKey)-{log.info(消息发送失败应答码{}原因{}交换机{}路由键{}消息{},replyCode,replyText,exchange,routingKey,message.toString());
});}
}3发送消息指定消息ID消息的ConfirmCallback 相比于发布消息多了消息的confirm
Test
public void testPubliserConfir()throw InterupteDException{// 创建correlationDataCorrelationData cd new CorrelationData(UUID.randowUUID().toString());// 给Future添加ConfirmCallbackcd.getFuture().addCallback(new ListenableFutureCllbackCorelationData.Confirm(){Overridepublic void onFailure(Throwable ex){// Future发生异常时的处理逻辑一般不触发log.error(handle message ack failex);}Overridepublic void onSuccess(CorrelationData.Confirm result){// Future接收到回执的处理逻辑if(result.isAck()){log.debug(发送消息成功收到ACK);}else{log.error(发送消息失败收到NACK,reason:{},result.getReson());}}
});
// 发送消息
rabbitTemplate.coverAndSend(hmall.direct,red,hello,cd);
}二 MQ的可靠性
MQ的持久化可以使用Lazy Queue 1通过配置类
Bean
public Queue lazyQueue(){return QueueBuilder.durable(lazy.queue)// 队列名称.lazy()//开启lazybulid();
}1基于注解
RabbitListener(queuesToDeclare Queue(namelazy.queue,durableture,argumentsArgument(namex-queue-mode,valuelazy)
))
public void listenLazyQueue(String msg){log.info(接收到 lazy.queue的消息{},msg);
}三 消费者确认
消费者确认机制 可以通过配置来进行确认
spring:rabbitmq:listenner:simple:prefetch: 1acknowledgs-mode: auto #确认机制 none-关闭ackmanual-手动ackauto-自动消费失败处理 重试机制
spring:rabbitmq:listenner:simple:prefetch: 1retry:enabled: true #开启超时重试机制initial-interval: 100ms #失败后的初始等待时间multipier: 1 #失败后下次的等待时长倍数 下次等待时长initial-interval*multipiermax-attempts: 3 #最大重试次数stateless: true #true为无状态若业务包含事务则使用false
失败处理策略 代码实现
Slf4j
Configureation
ConditionalOnProperty(prefixspring.rabbitmq.listenner.simple.retry,nameenable,havingValuetrue)// 只有重试机制是true才生效
public class ErrorConfiguration{Bbeanpublic DirectExchange errorExchange(){return new DirectExchange(error.direct);}Beanpublic Queue errorQueue(){return new Queue(error.queue);}Beanpublic Binding errorBinding(DirectExchange errorExchange,Queue errorQueue){return BindingBuilder.bind(errorQueue).to(errorExchange).with(eooro);}/*** 重试失败处理策略* RepublishMessageRecoverer:重试失败后将消息发送到指定的队列中*/Beanpublic MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate){log.info(MessageRecoverer 重试失败处理策略配置);return new RepublishMessageRecoverer(rabbitTemplate,error.direct,error);}
}四 业务幂等性
消息唯一id 业务判断
