rabbitmq消息零丢失方案,生产者confirm机制,rabbitmq持久化设置,消费者手动ack

如何保证rabbitmq消息零丢失?

我们从三个角色开始分析

1.生产者发送消息不丢失

　　生产者发消息到rabbitmq的网络传输过程中丢失了

　　以及消息发送到了rabbitmq但是mq内部出错,没有保存

　上面的问题有两种方案

　　第一种:rabbitmq支持事务消息,通过开启事务->发送消息->异常捕获并回滚->发送成功提交事务的方式保证消息发送mq成功, 但是有个弊端,这种方式是同步的,会导致消息的吞吐量下降,一般不使用这种方式

　　第二种:rabbitmq的channel开启confirm,其实就是回调机制,发送完消息后不用管,让rabbitmq通知你消息是发送成功还是失败,这种方式是异步的,对消息的吞吐量没什么影响,主要使用这种方法.

2.rabbitmq消息保存失败

　　rabbitmq接收到消息之后暂存在内存之中,如果在消费者还没有消费的时候,消息还在内存中,rabbitmq宕机了,那么内存中的消息就会丢失

　　一种方案:

　　rabbitmq对queue设置持久化,就是写一条消息就直接存储到磁盘上

3.消费者消费消息失败

　　消费者默认是autoAck,就是收到消息就自动提交ack,这就导致消息还没处理完,消费者宕机了,那么正在处理的消息就丢失了,恢复了之后,消费者会拉取新的消息

　　一种方案:

　　消费者关闭自动提交,改为手动ack,等消息全部处理完毕再提交ack,通知rabbitmq消息处理完毕,再发新的消息过来;

下面是整合spring的rabbitmq生产者代码实现:由于rabbitmq的队列持久化设置在管理平台就可以操作,消费者设置手动提交也比较简单,主要贴上生产者的代码实现

配置发送模板

package cn.picclife.cust.rrd.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import org.springframework.context.annotation.Scope;

/**
 * rabbitmq 配置
 * @ClassName: RabbitMqConfig
 * @Description: 初始化Rabbitmq
 * @author bin.zhao
 */
@Configuration
public class RabbitMqConfig {

    @Value("${spring.rabbitmq.host}")
    private String host;
    @Value("${spring.rabbitmq.virtual-host}")
    private String vHost;
    @Value("${spring.rabbitmq.username}")
    private String username;
    @Value("${spring.rabbitmq.password}")
    private String password;
    @Value("${spring.rabbitmq.queue}")
    private String queue;
    @Value("${spring.rabbitmq.exchange}")
    private String exchange;
    @Value("${spring.rabbitmq.routing.key}")
    private String rountingKey;
    /**
     * 针对消费者配置
     * 1. 设置交换机类型
     * 2. 将队列绑定到交换机
     */
    //创建队列,如果已创建好,就不用写
//    @Bean
//    public Queue topicQueue(){
//        return new Queue(this.queue,true);//创建队列并持久化
//    }
//    //创建交换机
//    @Bean
//    public TopicExchange topicExchange(){
//        return new TopicExchange(this.exchange,true,false);
//    }
//
//    //创建绑定
//    @Bean
//    public Binding topicBinding(){
//        return BindingBuilder.bind(topicQueue()).to(topicExchange()).with(this.rountingKey);
//    }
//

    @Bean(name = "MQConnectionFactory")
    public CachingConnectionFactory connectionFactory() throws Exception {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setUsername(this.username);
        connectionFactory.setPassword(this.password);
        connectionFactory.setAddresses(this.host);
        connectionFactory.setVirtualHost(this.vHost);
        //消息是否投递到exchange成功
        connectionFactory.setPublisherConfirms(true);
        return connectionFactory;
    }

    @Scope//默认单例模式
    @Bean(name = "groupRabbitTemplate")
    public RabbitTemplate rabbitTemplate(
            //直接使用注解,把连接工厂注入到模板中,防止找不到,导致发消息到mq失败
            @Qualifier("MQConnectionFactory") CachingConnectionFactory connectionFactory) throws Exception {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        return template;
    }

}

发送消息

 /** 
 * @description 发送消息
 * @date 2020
 */
@Component
public class AppRabbitMQ implements RabbitTemplate.ReturnCallback,RabbitTemplate.ConfirmCallback {private static final String EXCHANGE =
            ResourceUtils.getResource("config").getValue("spring.rabbitmq.exchange");

    private static final String ROUTING_KEY =
            ResourceUtils.getResource("config").getValue("spring.rabbitmq.routing.key");
    @Autowired
    private RabbitTemplate groupRabbitTemplate;/**
　　　*发送消息*/
    public void sendMessage(String message) throws Exception {
        //设置由于网络问题导致的连接Rabbitmq失败的重试策略
        RetryTemplate retryTemplate = new RetryTemplate();
        retryTemplate.setRetryPolicy(new SimpleRetryPolicy(3));
        //发送之前可以先把消息保存到数据库
        groupRabbitTemplate.setEncoding("UTF-8");
　　　　 groupRabblitTemplate.setMandatory(true);
        //true当消息无法被正常送达的时候被返回给生产者，false丢弃
        groupRabbitTemplate.setConfirmCallback(this);//ack回调
        groupRabbitTemplate.setReturnCallback(this);//回退回调
        try {//消息发送带上correlationData这个对象中保存有消息的唯一id,以便在数据库中查找消息或者从缓存中获取消息为了发送失败从新发送.
            groupRabbitTemplate.convertAndSend(EXCHANGE,ROUTING_KEY, JSONObject.toJSONString(message), correlationData);
            logger.info("消息发送,id:{}",correlationData.getId());
            Thread.sleep(100);//不让线程直接结束,等待回调函数confirm,如果不等,会直接异常,因为rabbitmq找不到回调方法
        }catch (Exception e){
            logger.error("发送消息失败:{}",ExceptionUtils.getStackTrace(e));
　　　　　　　//可以重试发送消息,我这里直接保存到数据库,后续定时任务扫描表格进行补发
　　　　　 　 //记录失败消息到失败数据表,并且更新消息表状态为发送失败
        }finally {
            message=null;//强引用设置为null,便于gc回收
        }
    }

 

 　　/** @Description: 用于定时任务的消息发送
　　 * @param:
　　 * @date: 2020
　　 * @return: void
　　 */
　　public void sendMessageOfTimeTask(String message){
　　    RetryTemplate retryTemplate = new RetryTemplate();
　　    retryTemplate.setRetryPolicy(new SimpleRetryPolicy(3));
　　    CorrelationData correlationData = new CorrelationData(message.getId());
   　　 groupRabbitTemplate.setEncoding("UTF-8");
    　　//true当消息无法被正常送达的时候被返回给生产者，false丢弃
　　    groupRabbitTemplate.setMandatory(true);//设置手工ack确认,
   　　 groupRabbitTemplate.setConfirmCallback(this);//ack回调
   　　 groupRabbitTemplate.setReturnCallback(this);//回退回调
    　　groupRabbitTemplate.convertAndSend(EXCHANGE,ROUTING_KEY,JSONObject.toJSONString(message),correlationData);
   　　 try {
      　　  Thread.sleep(100);//线程休眠,为了不让方法直接结束,回调函数无法正常回调confirm方法
    　　} catch (InterruptedException e) {
       　　 e.printStackTrace();
    　　}finally {
       　　 message=null;//强引用设置为null,便于gc回收
    　　}
　　}

　　  /**
     * 如果消息没有到exchange,则confirm回调,ack=false
     * 如果消息到达exchange,则confirm回调,ack=true
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        logger.info("消息回调confirm函数:{},ack:{},cause:{}", JSONObject.toJSONString(correlationData),ack,cause);
        if (ack) {
            //消费成功更新数据库记录为已发送状态
            
        } else {
            logger.info("推送消息失败,id:{},原因:{}",correlationData.getId(),cause);
           //记录失败消息到失败数据表,并且更新消息表状态为发送失败

        }
    }

    /**
     * exchange到queue成功,则不回调return
     * exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        logger.info("return--message:" + new String(message.getBody()) + ",replyCode:" + replyCode
                + ",replyText:" + replyText + ",exchange:" + exchange + ",routingKey:" + routingKey);
        //记录失败的消息id,更新数据库失败表
        String messgage = new String(message.getBody());
    }
}