CREATE TABLE `tx_record` (
    `tx_no` varchar(64) NOT NULL COMMENT '事務(wù)Id',
    `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '創(chuàng)建時(shí)間',
    PRIMARY KEY (`tx_no`) 
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='事務(wù)記錄表'

public interface TransactionListener {
    //當(dāng)發(fā)送事務(wù)消息成功之后，該方法會(huì)被觸發(fā)，本地事務(wù)將被執(zhí)行
    LocalTransactionState executeLocalTransaction(final Message msg, final Object arg);

    //當(dāng)沒有收到事務(wù)消息的響應(yīng)時(shí)，服務(wù)器會(huì)發(fā)送確認(rèn)消息來檢查事務(wù)狀態(tài)，該方法會(huì)被觸發(fā)并獲取本地事務(wù)狀態(tài)
    LocalTransactionState checkLocalTransaction(final MessageExt msg);
}

executeLocalTransaction {
 執(zhí)行本地事務(wù)
 如果失敗就選擇回滾事務(wù)，反之提交事務(wù)
}

checkLocalTransaction {
 實(shí)現(xiàn)事務(wù)回查
 根據(jù)事務(wù)執(zhí)行記錄判斷，已執(zhí)行則提交事務(wù)
}

@Service
public class CustomerTicketServiceImpl implements ICustomerTicketService {
    @Autowired
    TxRecordMapper txRecordMapper;

    @Autowired
    RocketMQTemplate rocketMQTemplate;

    @Override
    public void generateTicket(AddCustomerTicketReqVO addCustomerTicketReqVO) {
        //從VO中創(chuàng)建TicketGeneratedEvent
        TicketGeneratedEvent ticketGeneratedEvent = createTicketGeneratedEvent(addCustomerTicketReqVO);

        //將Event轉(zhuǎn)化為JSON對象
        JSONObject jsonObject =new JSONObject();
        jsonObject.put("ticketGeneratedEvent",ticketGeneratedEvent);
        String jsonString = jsonObject.toJSONString();

        //生成消息對象
        Message<String> message = MessageBuilder.withPayload(jsonString).build();

        //發(fā)送事務(wù)消息
        rocketMQTemplate.sendMessageInTransaction("producer_group_ticket","topic_ticket",message,null);
    }

    @Override
    @Transactional
    public void doGenerateTicket(TicketGeneratedEvent ticketGeneratedEvent) {
        //冪等判斷
        if(Objects.nonNull(txRecordMapper.findTxRecordByTxNo(ticketGeneratedEvent.getTxNo()))){
            return ;
        }

        //插入工單
        CustomerTicket customerTicket = CustomerTicketConverter.INSTANCE.convertEvent(ticketGeneratedEvent);
        customerTicket.setStatus(1);
        save(customerTicket);

        //添加事務(wù)日志
        txRecordMapper.addTxRecord(ticketGeneratedEvent.getTxNo());
    }
    ...
}

@Component
@RocketMQTransactionListener(txProducerGroup = "producer_group_ticket")
public class ProducerListener implements RocketMQLocalTransactionListener {
    @Autowired
    ICustomerTicketService customerTicketService;

    @Autowired
    TxRecordMapper txRecordMapper;

    //事務(wù)消息發(fā)送后的回調(diào)方法，當(dāng)消息發(fā)送給MQ成功，此方法被回調(diào)
    @Override
    @Transactional
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        try {
            //解析消息，轉(zhuǎn)成Event對象
            TicketGeneratedEvent ticketGeneratedEvent = convertEvent(message);

            //執(zhí)行本地事務(wù)
            customerTicketService.doGenerateTicket(ticketGeneratedEvent);

            //當(dāng)返回RocketMQLocalTransactionState.COMMIT，自動(dòng)向MQ發(fā)送commit消息，MQ將消息的狀態(tài)改為可消費(fèi)
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            e.printStackTrace();
            //如果本地事務(wù)執(zhí)行失敗，就將消息設(shè)置為回滾狀態(tài)
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    //事務(wù)狀態(tài)回查
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        //解析消息，轉(zhuǎn)成Event對象
        TicketGeneratedEvent ticketGeneratedEvent = convertEvent(message);

        //根據(jù)事務(wù)Id判斷是否存在已執(zhí)行的事務(wù)
        Boolean isTxNoExisted = Objects.nonNull(txRecordMapper.findTxRecordByTxNo(ticketGeneratedEvent.getTxNo()));

        //如果事務(wù)已執(zhí)行則返回COMMIT，反之返回UNKNOWN狀態(tài)
        if(isTxNoExisted){
            return RocketMQLocalTransactionState.COMMIT;
        }else{
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }
    ...
}

@Component
@RocketMQMessageListener(consumerGroup = "consumer_group_ticket",topic = "topic_ticket")
public class Consumer implements RocketMQListener<String> {
    @Autowired
    IChatRecordService chatRecordService;

    //接收消息
    @Override
    public void onMessage(String message) {
        log.info("開始消費(fèi)消息:{}",message);

        //解析消息
        JSONObject jsonObject = JSONObject.parseObject(message);
        String ticketGeneratedEventString = jsonObject.getString("ticketGeneratedEvent");

        //轉(zhuǎn)成TicketGeneratedEvent
        TicketGeneratedEvent ticketGeneratedEvent = JSONObject.parseObject(ticketGeneratedEventString, TicketGeneratedEvent.class);

        //添加本地聊天記錄
        chatRecordService.generateChatRecord(ticketGeneratedEvent);
    }
}

@Service
public class ChatRecordServiceImpl implements IChatRecordService {
    @Autowired
    TxRecordMapper txRecordMapper;

    @Override
    @Transactional
    public void generateChatRecord(TicketGeneratedEvent ticketGeneratedEvent) {
        //冪等判斷
        if(Objects.nonNull(txRecordMapper.findTxRecordByTxNo(ticketGeneratedEvent.getTxNo()))){
            return ;
        }

        //插入聊天記錄
        ChatRecord chatRecord = ChatRecordConverter.INSTANCE.convertEvent(ticketGeneratedEvent);
        save(chatRecord);

        //添加事務(wù)日志
        txRecordMapper.addTxRecord(ticketGeneratedEvent.getTxNo());
    }
}