3.1 二阶段提交(2PC)

①阶段一:表决阶段，协调者将事务信息发送给各参与者，然后各参与者接收到事务请求后向协调者反馈自身是否有能力执行事务提交并记录undo和redo日志

②阶段二:执行阶段，协调者接收到各参与者的反馈后，再通知各参与者进行真正的事务提交或回滚，如果各个参与者都收到回滚，则会根据undo日志执行回滚操作

优点:

提高了达到数据一致性的可能性，原理简单，实现成本低。

缺点:

①单点问题:如果协调者宕机，整个流程将不可用

②性能问题，在第一阶段要等待所有节点反馈，才能进入第二阶段

③数据不一致，在执行阶段如果协调者发送崩溃，导致只有部分参与者收到提交消息，那么就会存在数据不一致的问题

3.4 创建RocketMQ事务监听

@Component
@RocketMQTransactionListener
public class OrderProducerListener implements RocketMQLocalTransactionListener {
    private static final Logger logger = LoggerFactory.getLogger(OrderProducerListener.class);
    @Autowired
    OrderService orderService;
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        logger.info("开始执行本地事务{}", JSONUtil.toJsonStr(message));
        //根据不同的header信息分发到不同的业务场景中
        //MessageHeaders messageHeaders= message.getHeaders();
        //messageHeaders.get("");
        String msgJson = new String((byte[])message.getPayload());
        OrderDto orderDto=JSONUtil.toBean(msgJson,OrderDto.class);
        if (orderDto == null) {
            logger.info("订单数据为空事务状态设置为ROLLBACK。。。");
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        try {
            //执行生成订单业务
            Integer result = orderService.createOrder(orderDto);
            if (result > 0) {
                //数据执行成功则确认消息
                logger.info("插入订单数据成功{}", result);
                //模拟发生异常
                //int i=1/0;
                return RocketMQLocalTransactionState.COMMIT;
            } else {
                //数据执行失败则回滚消息
                logger.info("插入订单数据失败{}", result);
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        } catch (Exception e) {
            //发生异常进行回滚
            logger.error("执行本地事务发生异常:{}", e);
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        logger.info("开始回查本地事务{}", JSONUtil.toJsonStr(message));
        try {
            String msgJson = new String((byte[])message.getPayload());
            //TODO
            if (ObjectUtil.isNotNull(msgJson)) {
                OrderDto orderDto=JSONUtil.toBean(msgJson,OrderDto.class);
                //回查时检查是否生成订单数据
                Integer result = orderService.queryOrder(orderDto.getOrderId());
                if (result > 0) {
                    logger.info("数据已插入{}", result);
                    return RocketMQLocalTransactionState.COMMIT;
                } else {
                    //如果没有数据可以重新提交或者进行其他处理
                    return RocketMQLocalTransactionState.ROLLBACK;
                }
            }
        }catch (Exception e){
            logger.error("检查本地事务发生异常:{}",e);
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        logger.info("结束本地事务状态查询：{}", JSONUtil.toJsonStr(message));
        return RocketMQLocalTransactionState.UNKNOWN;
    }
}

@Service
public class DistributionServiceImpl implements DistributionService {
    private  final List<Integer> distributorList = new ArrayList<Integer>(){
        {
            //模拟配送员id
            add(1);
            add(2);
            add(3);
        }
    };
    @Autowired
    JdbcTemplate jdbcTemplate;
    @Override
    @Transactional
    public Integer distribution(String orderId) {
        String sql="insert into order_distribution (order_id,distributor) values (?,?)";
        //随机获取配送员id
        Integer distributor=distributorList.get(RandomUtil.randomInt(0,distributorList.size()));
        //插入数据
        int updateResult=jdbcTemplate.update(sql,new Object[]{orderId,distributor});
        return updateResult;
    }
}

@RocketMQMessageListener(topic = "springboot-producer-group", consumerGroup = "spring_boot_distribution", selectorExpression = "*")
@Service
public class DistributionLisTener implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {
    private static final Logger logger = LoggerFactory.getLogger(DistributionLisTener.class);
    @Autowired
    DistributionService distributionService;
    @Override
    public void onMessage(String o) {
        logger.info("开始消费消息:{}", o);
        Map<String, Object> msgMap = JSONUtil.toBean(o, Map.class);
        //模拟异常
        int i = 1 / 0;
        //执行分配业务
        distributionService.distribution((String) msgMap.get("orderId"));
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        // 设置最大重试次数
        defaultMQPushConsumer.setMaxReconsumeTimes(2);
    }
}