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();

}

}

for (int i = 0; i < 100000; i++) {

Thread.sleep(1000);

}

producer.shutdown();

}

}

public interface TransactionListener {

/**

* When send transactional prepare(half) message succeed, this method will be invoked to execute local transaction.

*

* @param msg Half(prepare) message

* @param arg Custom business parameter

* @return Transaction state

*/

LocalTransactionState executeLocalTransaction(final Message msg, final Object arg);

/**

* When no response to prepare(half) message. broker will send check message to check the transaction status, and this

* method will be invoked to get local transaction status.

*

* @param msg Check message

* @return Transaction state

*/

LocalTransactionState checkLocalTransaction(final MessageExt msg);

}

• HALF消息：RMQ_SYS_TRANS_HALF_TOPIC(临时存放消息信息)
  • 事务消息替换主体，保存原主题和对列信息
  • 半消息对Consumer不可见，不会被投递

• OP消息：RMQ_SYS_TARNS_OP_HALF_TOPIC(记录二阶段的操作)
  • Rollback：只做记录
  • Commit:根据备份信息重新构造消息并投递

CREATE TABLE mq_message(

id bigint NOT NULL AUTO_INCREMENT,

content varchar(255) NOT NULL,

topic char(64) NOT NULL,

tag char(64),

status tinyint,

createtime timestamp,

PRIMARY KEY(id)

)

class CommitLogDispatcherBuildConsumeQueue implements CommitLogDispatcher {

@Override

public void dispatch(DispatchRequest request) {

final int tranType = MessageSysFlag.getTransactionValue(request.getSysFlag());

switch (tranType) {

case MessageSysFlag.TRANSACTION_NOT_TYPE:

case MessageSysFlag.TRANSACTION_COMMIT_TYPE:

DefaultMessageStore.this.putMessagePositionInfo(request);

break;

case MessageSysFlag.TRANSACTION_PREPARED_TYPE:

case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:

break;

}

}

}