当前位置：首页 > news >正文

Kafka与RocketMQ在事务消息实现上的区别是什么？

news 2025/5/10 5:03:58

一、Kafka事务消息核心实现（基于2.8+版本）

// KafkaProducer.java
public synchronized Future<RecordMetadata> send(ProducerRecord<K, V> record) {// 事务消息校验（第256行）if (transactionManager != null && transactionManager.isTransactional()) {// 事务ID绑定检查（需先初始化事务）transactionManager.maybeFailWithError();}// 消息实际发送（第412行）return appendTransactionalRecord(record);
}// TransactionManager.java（关键事务方法）
public synchronized void beginTransaction() {// 生成新事务ID（第134行）this.transactionalId = generateTransactionalId();// 与协调者建立连接（第152行）coordinator.ensureTransactionalIdReady();
}public void commitTransaction() {// 两阶段提交第一阶段：写入提交标记（第489行）coordinator.beginCommit();// 第二阶段：提交所有消息（第503行）coordinator.sendOffsetsToTransaction();
}

二、RocketMQ事务消息核心实现（基于4.9+版本）

// TransactionMQProducer.java
public TransactionSendResult sendMessageInTransaction(Message msg, Object arg) {// 1.发送半消息（第87行）msg.putUserProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");SendResult sendResult = this.send(msg);// 2.执行本地事务（第94行）LocalTransactionState state = transactionListener.executeLocalTransaction(msg, arg);// 3.提交事务状态（第101行）this.endTransaction(sendResult, state, null);
}// DefaultMQProducerImpl.java（事务回查机制）
private void checkTransactionState() {// Broker定时回查（第356行）for (MessageExt msg : halfMsgs) {// 查询本地事务状态（第372行）LocalTransactionState state = transactionListener.checkLocalTransaction(msg);// 根据状态提交/回滚（第379行）endTransaction(msg, state);}
}

三、核心差异对比

设计架构：
- Kafka：Exactly-Once语义，通过事务协调器实现（其实本质上也是2PC，不过是通过一个特定的主题去做事务的处理）
- RocketMQ：采用二阶段提交+定时回查机制（rocketmq事务消息会提供一个回查接口，目的是为了兜底，当你长时间未提交当前事务消息，通过回查机制让业务觉得该条消息是否提交）
存储机制：

// Kafka日志追加（第512行）
public void appendToTransactionLog() {// 使用__transaction_state特殊主题存储事务状态（需ISR确认）
}// RocketMQ事务存储（CommitLog.java第227行）
public PutMessageResult putMessage(final MessageExtBrokerInner msg) {// 半消息存储到RMQ_SYS_TRANS_HALF_TOPIC主题if (msg.isTransactionPrepared()) {topic = TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC;}
}

异常处理：

// Kafka事务恢复（第672行）
void initializeTransactions() {// 通过事务ID恢复未完成事务coordinator.initializeTransactions();
}// RocketMQ事务补偿（第415行）
public void compensateDoTransaction() {// 超过checkTimeout未提交的消息自动回滚if (msg.getStoreTimestamp() + checkTimeout < now) {endTransaction(msg, LocalTransactionState.ROLLBACK_MESSAGE);}
}