大家好，我是君哥。

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

圖片

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

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

1.1 生產(chǎn)者事務(wù)

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

下圖是 RabbitMQ 的事務(wù)消息：

圖片

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

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

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

String transactionId = "xxxx123yyy"
Properties properties = new Properties();
properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionId);
KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
//開啟事務(wù)消息
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"));
     // 提交事務(wù)
     producer.commitTransaction();
} catch (ProducerFencedException e) {
     producer.abortTransaction();
}
producer.close();

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

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

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

圖片

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

圖片

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

2 TCC 模式

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

2.1 TCC 概述

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

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

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

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

圖片

commit/rollback 階段如下圖：

圖片

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

3 總結(jié)

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