rabbitMQ最终一致性处理分布式事务简单demo

以你之姓@ 2023-07-12 14:50 71阅读 0赞

demo地址:链接：https://pan.baidu.com/s/1kGaSCHlfhm6UvbPcRUYp4g 
    提取码：11hs
    分布式事务产生的背景?
       1.RPC通讯中每个服务都有自己独立的数据源,每个数据源都互不影响.
       2.在单个项目中存在多个不同jdbc连接(多数据源)
       
    如何基于我们的MQ解决我们的分布式事务的问题(最终一致性)
       1.确保我们的生产者往我们的MQ投递消息一定要成功.(生产者消息确认机制confirm),实现重试.
       2.确保我们的消费者能够消费成功(手动ack机制),如果消费失败情况下,MQ自动帮消费者重试.
       3.确保我们的生产者第一事务先执行成功,如果执行失败采用补单队列.

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nc2hlbmdxaWFuZzE2OA_size_16_color_FFFFFF_t_70][]

# 生产者 #

配置交换机 队列 绑定  和创建

package com.zhang.config;
    
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.core.RabbitAdmin;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Bean;
    import org.springframework.stereotype.Component;
    
    @Component
    public class OrderRabbitMQConfig {
    
    
        @Autowired
        RabbitAdmin rabbitAdmin;
    
        /**
         * 派单队列
         */
        public static final String ORDER_DIC_QUEUE = "order_dic_queue";
        /**
         * 补单对接
         */
        public static final String ORDER_CREATE_QUEUE = "order_create_queue";
        /**
         * 订单交换机
         */
        private static final String ORDER_EXCHANGE_NAME = "order_exchange_name";
    
        /**
         * 定义派单队列
         */
        @Bean
        public Queue directOrderDicQueue() {
            return new Queue(ORDER_DIC_QUEUE);
        }
    
        /**
         * 定义补派单队列
         */
        @Bean
        public Queue directCreateOrderQueue() {
            return new Queue(ORDER_CREATE_QUEUE);
        }
    
    
        /**
         * 定义订单交换机
         */
        @Bean
        DirectExchange directOrderExchange() {
            return new DirectExchange(ORDER_EXCHANGE_NAME);
        }
    
        /**
         * 派单队列与交换机绑定
         */
        @Bean
        Binding bindingExchangeOrderDicQueue() {
            return BindingBuilder.bind(directOrderDicQueue()).to(directOrderExchange()).with("orderRoutingKey");
        }
    
        /**
         * 补单队列与交换机绑定
         */
        @Bean
        Binding bindingExchangeCreateOrder() {
            return BindingBuilder.bind(directCreateOrderQueue()).to(directOrderExchange()).with("orderRoutingKey");
        }
    
        //创建初始化RabbitAdmin对象
        @Bean
        public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
            String string = connectionFactory.toString();
            RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
            // 只有设置为 true，spring 才会加载 RabbitAdmin 这个类
            rabbitAdmin.setAutoStartup(true);
            return rabbitAdmin;
        }
        //创建交换机和对列
        @Bean
        public void createExchangeQueue (){
            rabbitAdmin.declareExchange(directOrderExchange());
            rabbitAdmin.declareQueue(directOrderDicQueue());
            rabbitAdmin.declareQueue(directCreateOrderQueue());
        }
    }

发送订单消息

@Component
    @Slf4j
    public class OrderProducer implements RabbitTemplate.ConfirmCallback{
    
        @Autowired
        private OrderMapper orderMapper;
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Transactional
        public String sendOrder() {
            // 1.先创建我们订单信息
            String orderId = System.currentTimeMillis() + "";
            OrderEntity orderEntity = createOrder(orderId);
            // 2.添加到我们的数据库中
            int result = orderMapper.addOrder(orderEntity);
            if (result <= 0) {
                return null;
            }
            // 3.订单数据库插入成功的情况下, 使用MQ异步发送派单信息
            String msgJson = JSONObject.toJSONString(orderEntity);
            System.out.println("<<<<<<<1线程名称是>>>>>:"+Thread.currentThread().getName());
            sendMsg(msgJson);
            int i=1/0;
            return orderId;
        }
    
        //发送消息
        public void sendMsg(String msgJson) {
            // 设置生产者消息确认机制
            this.rabbitTemplate.setMandatory(true);
            this.rabbitTemplate.setConfirmCallback(this);
            CorrelationData correlationData = new CorrelationData();
            correlationData.setId(msgJson);
            String orderExchange = "order_exchange_name"; //订单交换机
            String orderRoutingKey = "orderRoutingKey";  //订单路由key
            rabbitTemplate.convertAndSend(orderExchange, orderRoutingKey, msgJson, correlationData);
        }
    
        //创建订单信息
        public OrderEntity createOrder(String orderId) {
            OrderEntity orderEntity = new OrderEntity();
            orderEntity.setName("每特教育第六期平均就业薪资破10万");
            orderEntity.setOrderCreatetime(new Date());
            // 价格是300元
            orderEntity.setOrderMoney(300d);
            // 状态为 未支付
            orderEntity.setOrderState(0);
            Long commodityId = 30L;
            // 商品id
            orderEntity.setCommodityId(commodityId);
            orderEntity.setOrderId(orderId);
            return orderEntity;
        }
    
        /**
         * 实现ConfirmCallback接口 重写confirm方法
         * @param correlationData 投递失败回调消息
         * @param ack true 投递到MQ成功  false投递消息失败
         * @param cause
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            String msg = correlationData.getId();
            if(!ack){
                log.info("<<<往MQ投递消息失败>>>>: {}" , msg);
                //采用递归重试
                sendMsg(msg);
                return;
            }
            log.info("<<<往MQ投递消息成功>>>>: {}" , msg);
            // 生产者投递多次还是is的情况下应该 人工记录
        }
    }

# 消费者 #

派单消费者

@Component
    @Slf4j
    public class DispatchConsumer {
    
        @Autowired
        private DispatchMapper dispatchMapper;
    
        @RabbitListener(queues = "order_dic_queue")
        public void dispatchConsumer(Message message , Channel channel) throws IOException {
            // 1.获取消息
            String msg = new String(message.getBody());
            // 2.转换json
            JSONObject jsonObject = JSONObject.parseObject(msg);
            String orderId = jsonObject.getString("orderId");
            //主动根据orderId查询派单是否已经派单过,如果派单,不走下面插入代码
            // 计算分配的快递员id
            DispatchEntity dispatchEntity = new DispatchEntity(orderId, 1234L);
            // 3.插入我们的数据库
            int result = dispatchMapper.insertDistribute(dispatchEntity);
            if (result > 0) {
                // 手动将该消息删除
                // 手动ack 删除该消息
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                log.info("<<<消费者派单消费成功>>> 时间: {} " ,new Date());
            }
        }
    }

补单消费者

@Component
    @Slf4j
    public class OrderConsumer {
    
        @Autowired
        private OrderMapper orderMapper;
    
        /***
         * 补单的消费不应该和 订单生产者放到一个服务器节点
         * 补单消费者如果不存在的情况下 队列缓存补单消息
         * 补偿分布式事务解决框架 思想最终一致性
         */
        @RabbitListener(queues = "order_create_queue")
        public void dispatchConsumer(Message message , Channel channel) throws IOException {
            // 1.获取消息
            String msg = new String(message.getBody());
            // 2.转换json
            OrderEntity orderEntity = JSONObject.parseObject(msg,OrderEntity.class);
            String orderId = orderEntity.getOrderId();
            OrderEntity result = orderMapper.findOrderId(orderId);
            if (null!=result ) {
                // 手动将该消息删除
                // 手动ack 删除该消息
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                log.info("<<<消费者消费成功>>> 时间: {}",new Date());
                return;
            }
            //补单
            int i = orderMapper.addOrder(orderEntity);
            if(i>0){
                // 手动将该消息删除
                // 手动ack 删除该消息
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                log.info("<<<消费者补单消费成功>>> 时间: {}",new Date());
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nc2hlbmdxaWFuZzE2OA_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200309142645379.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nc2hlbmdxaWFuZzE2OA==,size_16,color_FFFFFF,t_70