RocketMQ的事务消息模式

客官°小女子只卖身不卖艺 2022-11-15 03:50 240阅读 0赞

## 1. 事务消息简介 ##

即，在分布式系统中保证最终一致性的两阶段提交的消息实现。

#### 侧重点：保证*本地事务执行* 与*消息发送*两个操作的原子性 ####

## 2. 事务消息发送流程 ##

1.  消息生产者向Broker投递一个half（事务）消息（如需保证幂等性，可以携带一个唯一的key作为参数）
2.  确认Broker接收到消息（消息会在Broker本地进行存储，但是该消息的状态对消费者不可见）
3.  生产者确认消息发送成功后，回调executeLocalTransaction方法来执行本地事务
4.  生产者提交本地事务的执行状态
5.  Broker进行最终处理

过程中可能的情况：

*  第4步提交事务成功状态，Broker将消息投递到消费者
 *  第4步提交事务失败（回滚）状态，Broker丢弃该消息
 *  第4步提交超时时，Broker会定时回查  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h1ZWd1Y2hlbg_size_16_color_FFFFFF_t_70]

## 3. 代码实现 ##

package com.example.mq.transaction;
    
    import org.apache.commons.lang3.StringUtils;
    import org.apache.rocketmq.client.producer.LocalTransactionState;
    import org.apache.rocketmq.client.producer.TransactionListener;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.common.message.MessageExt;
    
    
    /** * 描述: * 事务消息监听器 * * @author XueGuCheng * @create 2021-03-30 21:59 */
    public class TransactionListenerImpl implements TransactionListener { 
    
        /** * 提交完事务消息后执行此方法 * 返回COMMIT_MESSAGE（正常提交）状态的消息会立即被消费者消费到 * 返回ROLLBACK_MESSAGE（回滚）状态的消息会被丢弃 * 返回UNKNOWN（消息状态未知，因为网络等原因）状态的消息会由Broker过一段时间再来回查事务的状态 * * @param message 消息内容 * @param o * @return */
        @Override
        public LocalTransactionState executeLocalTransaction(Message message, Object o) { 
            // 编写业务逻辑
            /* 业务逻辑：根据消息携带的不同的Tag属性进行不同的业务操作 Tag_A的消息会提交，即立即被消费者消费到 Tag_B的消息会回滚，即会被丢弃 其他Tag属性消息会被标记为未知状态，等待Broker进行事务状态回查 */
            String tags = message.getTags();
            if (StringUtils.contains(tags, "Tag_A")) { 
                return LocalTransactionState.COMMIT_MESSAGE;
            } else if (StringUtils.contains(tags, "Tag_B")) { 
                return LocalTransactionState.ROLLBACK_MESSAGE;
            } else { 
                return LocalTransactionState.UNKNOW;
            }
    
        }
    
        /** * 对UNKNOWN状态的消息进行事务状态回查 * * @param messageExt UNKNOWN状态的消息 * @return 事务状态 */
        @Override
        public LocalTransactionState checkLocalTransaction(MessageExt messageExt) { 
            // 编写业务逻辑
             /* 业务逻辑：根据消息携带的不同的Tag属性进行不同的业务操作 回查时，Tag_C的消息会提交，即会被消费者消费到 回查时，Tag_D的消息会回滚，即会被丢弃 回查时，其他Tag属性消息会在多次回查（默认15次）后，最终丢弃 */
            String tags = messageExt.getTags();
            if (StringUtils.contains(tags, "Tag_C")) { 
                return LocalTransactionState.COMMIT_MESSAGE;
            } else if (StringUtils.contains(tags, "Tag_D")) { 
                return LocalTransactionState.ROLLBACK_MESSAGE;
            } else { 
                return LocalTransactionState.UNKNOW;
            }
    
        }
    }

## 4. 事务消息的限制 ##

1.  不支持延迟消息和批量消息
2.  消息回查次数默认15次，为了防止消息回查次数太多次而导致消息累积。可以配置 Broker 的配置文件的 transactionCheckMax 参数来修该回查次数，回查次数超过该参数时， Broker 将丢弃此消息，并在默认情况下同时打印错误日志，可以通过重 AbstractTransactionCheckListener 类来修改这个行为
3.  多久后进行回查？超时时间在Broker 的配置文件的 transactionMsgTimeout 参数进行设置，也可以在发送事务消息时，配置用户属性 CHECK\_IMMUNITY\_TIME\_IN\_SECONDS 来进行设置（该设置优先级高于transactionMsgTimeout 参数）
4.  多次消费，即事务性消息可能不止一次被检查或消费
5.  事务消息的生产者 ID 不能与其他类型消息的生产者 ID 共享，因为事务消息会进行回查

## 5. 总结 ##

**RocketMQ的事务消息模式息只保证了发送者本地事务和发送消息这两个操作的原子性，并没有保证分布式系统下的两个事务的一致性**（下游事务失败或者回滚，上游事务是感知不到的）。所以，事务消息模式只保证了分布式事务的一半的一致性。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h1ZWd1Y2hlbg_size_16_color_FFFFFF_t_70]: /images/20221022/08491fddae95414ebd4bb10f74796282.png