@Test
    public void sendMessage() throws Exception {
        //事務(wù)生產(chǎn)者
        TransactionMQProducer producer = new TransactionMQProducer("defaultGroup");
        producer.setNamesrvAddr(SpringUtil.getBean(RocketMqConfig.class).getNamesrvAddr());
        //設(shè)置檢查本地事務(wù)狀態(tài)的線程池
        //producer.setExecutorService(null);
        //本地事務(wù)執(zhí)行監(jiān)聽器
        TransactionListener transactionListener = new TransactionListenerImpl();
        producer.setTransactionListener(transactionListener);
        producer.start();
        Message message = new Message(RocketMqUtil.TOPIC, "transaction", "transaction-message".getBytes(Charset.forName("UTF-8")));
        //發(fā)送事務(wù)消息
        producer.sendMessageInTransaction(message, null);
    }
    class TransactionListenerImpl implements TransactionListener {
        @Override
        public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            //執(zhí)行本地事務(wù)（數(shù)據(jù)庫(kù)）操作......
            int num = new Random().nextInt(10);
            if (num < 3) {
                //本地事務(wù)執(zhí)行成功，提交消息
                return LocalTransactionState.COMMIT_MESSAGE;
            } else if (num < 6) {
                //本地事務(wù)執(zhí)行失敗，刪除消息
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
            //等待本地事務(wù)check，即執(zhí)行checkLocalTransaction()方法
            return LocalTransactionState.UNKNOW;
        }
        /**
         * 回查邏輯
         * @param msg
         * @return
         */
        @Override
        public LocalTransactionState checkLocalTransaction(MessageExt msg) {
            int num = new Random().nextInt(10);
            if (num < 3) {
                //提交消息
                return LocalTransactionState.COMMIT_MESSAGE;
            } else if (num < 6) {
                //刪除消息
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
            return LocalTransactionState.UNKNOW;
        }
    }

@Test
    public void consumeMessage() throws Exception {
        DefaultMQPushConsumer defaultMQPushConsumer = RocketMqUtil.getDefaultMQPushConsumer();
        defaultMQPushConsumer.subscribe(RocketMqUtil.TOPIC, "*");
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                                                            ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                log.info("消費(fèi)到消息條數(shù)：{}", list.size());
                list.stream().map(messageExt -> new String(messageExt.getBody(), Charset.forName("UTF-8")))
                        .map(String::new).forEach(System.out::println);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        defaultMQPushConsumer.start();
        Thread.sleep(5000L);
    }