事务消息

    // 发送事务消息的实例
    public boolean sendMessageInTransaction( String msg)  {
        try {
            Message message = new Message();
            message.setKeys(msg);
            System.out.println("========sending message=========");
            TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction("tx-group", "topic-tx", MessageBuilder.withPayload(message).build(), null);
            System.out.println("========finish send =========");
            
            //查看半消息状态
            LocalTransactionState localTransactionState = transactionSendResult.getLocalTransactionState();
            System.out.println(localTransactionState.name());
            return true;
        } catch (MessagingException e) {
            e.printStackTrace();
            return false;
        }
    }

@RocketMQTransactionListener(txProducerGroup = "tx-group")
public class TransactionMsgListener implements RocketMQLocalTransactionListener {

    @Reference
    GoodService goodService;

    @Autowired
    UserServiceImpl userServiceImpl;

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(org.springframework.messaging.Message message, Object o) {
        System.out.println("执行本地事务=====");
        String s = new String((byte[]) message.getPayload());
        System.out.println(s);
        JSONObject jsonObject = JSONObject.parseObject(s);
        String myinput = (String)jsonObject.get("keys");
        String[] split = myinput.split(":");


        String myUserId=split[0];
        int goodId=Integer.parseInt(split[1]);
        int count=Integer.parseInt(split[2]);
        GoodRequest goodRequest = new GoodRequest();
        goodRequest.setCount(count);
        goodRequest.setGoodId(goodId);

        try {
            GoodResponse goodResponse = goodService.reduceStock(goodRequest);
            if(goodResponse.isResult()){
                //扣库存成功
                System.out.println("成功");
                return RocketMQLocalTransactionState.COMMIT;
            } else{
                System.out.println("回滚");
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        } catch (Exception e){
            e.printStackTrace();
            System.out.println("未知");
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(org.springframework.messaging.Message message) {
        System.out.println("执行事务补偿======");
        //事务补偿提交
        return RocketMQLocalTransactionState.COMMIT;
        //事务补偿回滚
        //return RocketMQLocalTransactionState.ROLLBACK;
        //如果事务补偿过程还是UNKNOWN 就会一直进行事务补偿，60s一次
        //return RocketMQLocalTransactionState.UNKNOWN;
    }
}