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

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());
    }
}