大家好，我是君哥。

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

圖片

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

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

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

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

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

圖片

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

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

Kafka 的事務(wù)消息類(lèi)似，指定一個(gè)事務(wù) id，集群中不同的生產(chǎn)者都可以使用這個(gè)事務(wù) id 來(lái)實(shí)現(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);
//開(kāi)啟事務(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();

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

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

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

圖片

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

圖片

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

2 TCC 模式

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

2.1 TCC 概述

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

TCC 模式有三個(gè)角色，TM、RM 和 TC：

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

以購(gòu)物場(chǎng)景中保存訂單、扣減庫(kù)存、扣減金額的例子來(lái)講，try 階段如下圖：

圖片

commit/rollback 階段如下圖：

圖片

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

3 總結(jié)

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