rocketMq事务消息

    在RocketMq4.3版本之后就已经对事务消息的支持了，rocketmq事务消息采用的是两阶段提交的原理进行实现的，保证了producer消息的发送和producer本地任务的一致性。

rocketmq事务消息流程图.png

rocketMq事务消息流程：

• producer采用同步发送消息到broker，这个时候的消息是prepare消息，发送到了RMQ_SYS_TRANS_HALF_TOPIC这个主题下，原主题作为properties存起来了
• broker会把事务消息存到RMQ_SYS_TRANS_HALF_TOPIC主题的messageQueue=0的队列下，和普通同步消息没什么差别
• producer接收到同步消息发送成功之后，会执行本地事务，并给出事务执行状态告知broker
• broker接收到commit或rollback状态后会提交或删除消息
• 对于unkown状态的消息，broker会进行消息状态的回调，查询producer本地事务执行状态来决定消息的提交或删除

事务消息使用实例

     实现TransactionListener接口，实现executeLocalTransaction本地事务执行接口和checkLocalTransaction本地事务执行状态回查接口

public class TransactionListenerImpl implements TransactionListener {
    private AtomicInteger transactionIndex = new AtomicInteger(0);
    //本地事务执行状态的简单模拟，可自己存到实际的存储介质中
    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        int value = transactionIndex.getAndIncrement();
        int status = value % 3;
        localTrans.put(msg.getTransactionId(), status);
        return LocalTransactionState.UNKNOW;
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        Integer status = localTrans.get(msg.getTransactionId());
        if (null != status) {
            switch (status) {
                case 0:
                    return LocalTransactionState.UNKNOW;
                case 1:
                    return LocalTransactionState.COMMIT_MESSAGE;
                case 2:
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                default:
                    return LocalTransactionState.COMMIT_MESSAGE;
            }
        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

    TransactionProducer接口发送事务消息，这里Producer设置TransactionListener和executorService，这里的线程池主要是用来执行回调任务线程的

public class TransactionProducer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        TransactionListener transactionListener = new TransactionListenerImpl();
        TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-check-thread");
                return thread;
            }
        });

        producer.setExecutorService(executorService);
        producer.setTransactionListener(transactionListener);
        producer.start();

        String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
        for (int i = 0; i < 10; i++) {
            try {
                Message msg =
                    new Message("TopicTest1234", tags[i % tags.length], "KEY" + i,
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.sendMessageInTransaction(msg, null);
                System.out.printf("%s%n", sendResult);

                Thread.sleep(10);
            } catch (MQClientException | UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }
    }
}