大家好，我是君哥。

消息隊列的主要功能是系統(tǒng)間解耦，實現(xiàn)流量的削峰填谷。主流的消息隊列一般有三個核心操作：消費者發(fā)送消息，Broker 保存消息，消費者消費消息。如下圖：

圖片

對于一個完整的事務消息，可以理解為生產(chǎn)者生產(chǎn)消息和消費者消費消息，這兩個操作要不全部成功要不全部失敗。但事實上，很難有消息隊列來實現(xiàn)生產(chǎn)消息和消費消息的事務特性。

1 事務消息實現(xiàn)

1.1 生產(chǎn)者事務

一些消息隊列在生產(chǎn)端實現(xiàn)事務消息，對生產(chǎn)的一批消息要不全部發(fā)送成功，要不全部發(fā)送失敗。

下圖是 RabbitMQ 的事務消息：

圖片

RabbitMQ 的事務消息實現(xiàn)的就是生產(chǎn)者的事務消息，發(fā)送消息的時候，把 Channel 設置為事務模式，這樣這一批消息就會緩存在 Channel，等提交事務時才會提交到 Broker?？梢詤⒖枷旅娲a：

ConnectionFactory factory=new ConnectionFactory();
cnotallow=factory.newConnection();
Channel channel=connection.createChannel();
//開啟事務
channel.txSelect();
channel.basicPublish("directTransactionExchange","transactionRoutingKey",null,message.getBytes("utf-8"));
//提交事務 或者 channel.txRollback()回滾事務
channel.txCommit();

Kafka 的事務消息類似，指定一個事務 id，集群中不同的生產(chǎn)者都可以使用這個事務 id 來實現(xiàn)事務消息?？聪旅娲a：

String transactionId = "xxxx123yyy"
Properties properties = new Properties();
properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionId);
KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
//開啟事務消息
producer.initTransactions();
producer.beginTransaction();

try {
     //發(fā)送消息
     String topic = "testTransactionId"
     producer.send(new ProducerRecord<String, String>(topic, "msg1"));
     producer.send(new ProducerRecord<String, String>(topic, "msg2"));
     producer.send(new ProducerRecord<String, String>(topic, "msg3"));
     // 提交事務
     producer.commitTransaction();
} catch (ProducerFencedException e) {
     producer.abortTransaction();
}
producer.close();

使用同一個事務 id 后，集群中的多個生產(chǎn)者可以實現(xiàn)對不同 Topic、不同分區(qū)的原子性寫入。在提交事務之前，這些消息是不可見的，事務提交之后，這些消息才能變?yōu)榭梢姟?/p>

1.2 生產(chǎn)者+本地事務

生產(chǎn)者加本地事務也是消息隊列事務消息的一種實現(xiàn)方式。這種事務是指本地事務和生產(chǎn)者生產(chǎn)消息是一個原子性操作，要不都成功，要不都失敗。如下圖所示：

圖片

RocketMQ 的事務消息是首先發(fā)送 half 消息到 Broker，然后執(zhí)行本地事務，最后執(zhí)行 commit/rollback 操作。執(zhí)行了 commit 操作后，消費者就可以拉取到這條消息了。如下圖：

圖片

RocketMQ 的 half 消息是將消息投遞到 topic 為 RMQ_SYS_TRANS_HALF_TOPIC 中 queueId 等于 0 的隊列中，當 RocketMQ 發(fā)送 commit 消息后再把消息投遞到原始隊列，這樣消費者就可以拉取到這條消息了。

2 TCC 模式

從上面的講解可以看到，消息隊列的事務消息使用的是兩階段提交的方式。那跟 TCC 模式的兩階段提交有什么區(qū)別呢？這節(jié)我們來看一下 TCC 模式。

2.1 TCC 概述

TCC 模式將分布式事務分為 try 和 commit/rollback 兩個階段來執(zhí)行，try 階段對每個分支事務進行預留資源，如果 try 階段所有節(jié)點都執(zhí)行成功，則進入 commit 階段提交全局事務，只要有一個節(jié)點執(zhí)行失敗則進入 rollback 階段回滾全局事務。

TCC 模式有三個角色，TM、RM 和 TC：

• TC: 協(xié)調(diào)節(jié)點，保存分支事務元數(shù)據(jù)和狀態(tài)，對分支事務發(fā)送提交或回滾命令；
• TM：管理全局事務，包括開啟全局事務，提交/回滾全局事務；
• RM：管理分支事務，包括提交或回滾本地事務，向 TC 注冊分支事務并上報分支事務狀態(tài)。

以購物場景中保存訂單、扣減庫存、扣減金額的例子來講，try 階段如下圖：

圖片

commit/rollback 階段如下圖：

圖片

可以看到，TCC 模式的兩階段提交，第一階段是用來嘗試預留資源，第二階段來扣減資源或釋放資源。而消息隊列中生產(chǎn)消息、保存消息、消費消息是不需要預留資源的，這是完全不同的業(yè)務場景。

3 總結

本文主要講述了消息隊列中的事務消息跟分布式事務中兩階段提交的 TCC 模式在實現(xiàn)方式上的區(qū)別。TCC 模式中兩階段提交的目的主要是用來確定資源是否可用，而消息隊列的事務消息是不需要考慮資源的，二者是不同的業(yè)務場景。