springboot RabbitMQ分布式事务-可靠消息的最终一致性方案代码示例

朴灿烈づ我的快乐病毒、 2023-07-07 08:23 5阅读 0赞

**分布式事务主要问题**

如果先发送消息，再执行本地事务，可能会出现消息已经发送成功，但本地事务发生错误； 如果先执行本地事务，再发送消息，可能本地事务执行成功，但消息发送发生错误。

# Rabbit mq分布式事务实现思路 #

*  生产者向**Rabbit mq**发送消息，消费者此时不消费消息
 *  生产者接收**Rabbit mq**返回的消息确认接收成功通知（ACK）
 *  生产者判断ACK，执行本地事务。
 *  本地事务执行失败，往Redis缓存里面存入消息ID，消费者消费前查询是否存在该ID，如果存在则不消费，否则消费。
 *  本地事务执行成功，消费者消费消息。

引入了Redis缓存,生产者预发送消息，消费者并未直接消费；生产者本地事务出现问题，消费者也不会消费消息；从而实现了数据一致性。

**pom引入以下依赖**

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
     </dependency>
      <dependency>
                <groupId>redis.clients</groupId>
                <artifactId>jedis</artifactId>
                <version>2.9.0</version>
     </dependency>
      <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.2.5</version>
       </dependency>

在application.properties中配置RabbitMQ连接信息

spring.rabbitmq.addresses=192.168.3.12
    spring.rabbitmq.port=15672
    spring.rabbitmq.username=wangsheng
    spring.rabbitmq.password=123
    #确认是否正确到达exchange，只要正确的到达exchange中，即可确认该消息返回给客户端ack
    spring.rabbitmq.publisher-confirms=true
    #确认消息是否正确到达queue,如果没有则触发，如果有则不触发
    spring.rabbitmq.publisher-returns=true
    #采用手动应答
    spring.rabbitmq.listener.simple.acknowledge-mode=manual
    #是否支持重试
    spring.rabbitmq.listener.simple.retry.enabled=true
    #最大重试次数
    spring.rabbitmq.template.retry.max-attempts=5
    #最大重试时间
    spring.rabbitmq.template.retry.max-interval=1200
    #exchange到queue失败,则回调return
    spring.rabbitmq.template.mandatory=true

配置Jedis连接

import redis.clients.jedis.Jedis;
    import redis.clients.jedis.JedisPool;
    import redis.clients.jedis.JedisPoolConfig;
    /**
     * 创 建 人 ： wangsheng 创建日期：2020年2月
     */
    public class RedisPool {
        private static JedisPool pool;//jedis连接池
    
        private static int maxTotal = 20;//最大连接数
    
        private static int maxIdle = 10;//最大空闲连接数
    
        private static int minIdle = 5;//最小空闲连接数
    
        private static boolean testOnBorrow = true;//在取连接时测试连接的可用性
    
        private static boolean testOnReturn = false;//再还连接时不测试连接的可用性
    
        static {
            initPool();//初始化连接池
        }
    
        public static Jedis getJedis(){
            return pool.getResource();
        }
    
        public static void close(Jedis jedis){
            jedis.close();
        }
    
        private static void initPool(){
            JedisPoolConfig config = new JedisPoolConfig();
            config.setMaxTotal(maxTotal);
            config.setMaxIdle(maxIdle);
            config.setMinIdle(minIdle);
            config.setTestOnBorrow(testOnBorrow);
            config.setTestOnReturn(testOnReturn);
            config.setBlockWhenExhausted(true);
            pool = new JedisPool(config, "127.0.0.1", 6379, 5000);
        }
    }

配置一个direct队列

import org.springframework.amqp.core.Queue;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
     /**
     * 创 建 人 ： wangsheng 创建日期：2020年2月
     */
    @Configuration
    public class RabbitDirectConfig {
        public final static String DIRECTNAME="wangsheng-direct";
        @Bean
        public Queue queue() {
            /** 参数含义
             * 第一个是队列名
             *  第二个 durable="true" 持久化 rabbitmq重启的时候不需要创建新的队列
             *  第三个exclusive  表示该消息队列是否只在当前connection生效,默认是false
             *  第四个auto-delete 表示消息队列没有在使用时将被自动删除 默认是false
             */
            return new Queue("hello",true,false,false);// 使用DirectExchange时，routingkey与队列名相同，只配置一个Queue即可，DirectExchange和Binging两个的Bean配置可以省略。
     
        }
    }

配置一个生产者,需要注意的是只要消息到达了exchange中，返回confirm的ACK就是true，与消费者签收与否无关。

import com.alibaba.fastjson.JSONObject;
    import com.zx.zhuangxiu.cache.RedisPool;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageBuilder;
    import org.springframework.amqp.core.MessageDeliveryMode;
    import org.springframework.amqp.core.MessageProperties;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    import java.util.UUID;
    /**
     * 创 建 人 ： wangsheng 创建日期：2020年2月
     */
    @RestController
    @RequestMapping("/send/")
    public class test {
        @Autowired
        RabbitTemplate rabbitTemplate;
    
        //回调函数: confirm确认
        RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
            /**
             * CorrelationData 消息的附加信息，即自定义id
             * ack 代表消息是否被broker（MQ）接收 true 代表接收 false代表拒收。
             * cause 如果拒收cause则说明拒收的原因，帮助我们进行后续处理
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("correlationData-----------" + correlationData);
                System.out.println("ack-----------" + ack);
                if(ack){
                    try{  //更新数据库，可靠性投递机制
                          // db操作略。。。
                        System.err.println("消息已确认....");
                    }catch (Exception e)
                    {//如果更新数据库，可靠性投递机制发生错误，就往redis里面存消息ID CorrelationData，后面消费者要判断redis里面是否有该ID,过期时间设置一周
                       RedisPool.getJedis().set(correlationData.getId(), correlationData.getId(),"NX", "EX",7*24*3600);
                        System.out.println("操作数据库发生错误");
                    }
                }else {
                    //ack错误打印
                    System.err.println(cause);
                }
            }
        };
        //消息失败回调函数
       RabbitTemplate.ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.err.println("return exchange: " + exchange + ", routingKey: "
                        + routingKey + ", replyCode: " + replyCode + ", replyText: " + replyText);
    
            }
        };
    
        //发送消息方法调用: 构建Message消息
        @RequestMapping("send")
        public void sends() throws Exception {
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("email", "wangsheng@163.com");
            jsonObject.put("timestamp", System.currentTimeMillis());
            String jsonString = jsonObject.toJSONString();
            String orderId = UUID.randomUUID().toString();
            Message messages = MessageBuilder.withBody(jsonString.getBytes())
                    .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                    .setContentType(MessageProperties.CONTENT_TYPE_JSON).setContentEncoding("utf-8")
                    .setMessageId(orderId)
                    .build();
            CorrelationData correlationData = new CorrelationData(orderId);
            rabbitTemplate.setConfirmCallback(confirmCallback);
            rabbitTemplate.setReturnCallback(returnCallback);
            rabbitTemplate.convertAndSend("hello",messages,correlationData);//routingkey与队列名相同都是hello
    
        }
    }

配置一个消费者

import com.alibaba.fastjson.JSONObject;
    import com.rabbitmq.client.Channel;
    import com.zx.zhuangxiu.cache.RedisPool;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    import java.io.IOException;
    /**
     * 创 建 人 ： wangsheng 创建日期：2020年2月
     */
    @Component
    public class DirectReceiver {
        @RabbitListener(queues = "hello")//监听队列hello
        public void handler1(Message message, Channel channel) throws IOException {
            // 每次只接收一个信息
            channel.basicQos(1);
            String msg = new String(message.getBody(), "UTF-8");
            String messageId = message.getMessageProperties().getMessageId();
            JSONObject jsonObject = JSONObject.parseObject(msg);
            try {
                //channel.waitForConfirms()回调生产者confirm方法，只要回调成功了confirm，返回就是true，即使confrim里执行本地事务操作发生错误也是true
                if (channel.waitForConfirms()) {
                    //redis消息ID不为空说明该消息在事务操作发生了错误
                    if (RedisPool.getJedis().get(messageId) != null) {
                        // 丢弃该消息
                        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
                        //redis删除消息ID
                        RedisPool.getJedis().del(messageId);
                    } else {
                        //执行更新数据库，可靠性投递机制，执行前要先查询一下数据库是否存在，避免签收前mq挂掉，重启后重发导致重复消费。
                        //db略
                        //消息确认
                        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
                        System.out.println("消费消息：" + jsonObject);
                    }
                }
            } catch (Exception e) {
                // 丢弃该消息
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
                //失败后消息被确认
                // channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
                //失败后消息重新放回队列，一般不用这个，可能会导致无限循环
                // channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
                System.out.println("签收失败:" + jsonObject);
            }
        }
    }

测试结果，成功消费。

![20200222001302571.png][]

[20200222001302571.png]: https://img-blog.csdnimg.cn/20200222001302571.png