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