RocketMQ事务消息提交延迟导致业务阻塞的性能调优实战 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

RocketMQ 事务消息提交延迟导致业务阻塞的性能调优实战

大家好！今天我们来聊聊在使用 RocketMQ 事务消息时，可能遇到的一个棘手问题：事务消息提交延迟导致业务阻塞，以及如何进行性能调优。这个问题如果不重视，可能会导致整个系统的性能瓶颈，甚至出现严重的业务故障。

1. 事务消息的基本原理回顾

在深入分析问题之前，我们先简单回顾一下 RocketMQ 事务消息的基本原理。事务消息的核心是为了保证分布式事务的最终一致性。其主要流程如下：

发送 Half 消息 (Prepare 消息): Producer 先发送一条半消息 (Half Message) 到 Broker。Half 消息对 Consumer 不可见。
执行本地事务: Producer 端执行本地事务。
Commit/Rollback:
- 如果本地事务执行成功，Producer 向 Broker 发送 Commit 消息，Broker 将 Half 消息标记为可投递，Consumer 就能消费到这条消息。
- 如果本地事务执行失败，Producer 向 Broker 发送 Rollback 消息，Broker 删除 Half 消息。
回查 (Check): 如果 Broker 长时间未收到 Commit/Rollback 消息，会主动向 Producer 发起回查请求，询问事务状态。 Producer 根据本地事务状态，返回 Commit/Rollback。

2. 问题分析：事务消息提交延迟的原因

事务消息提交延迟，本质上是 Producer 发送 Commit/Rollback 消息延迟。导致延迟的原因有很多，可以从以下几个方面进行分析：

网络延迟: 这是最常见的原因。Producer 和 Broker 之间的网络不稳定、带宽受限，都会导致消息发送延迟。
Broker 压力过大: Broker 资源不足，例如 CPU 负载过高、磁盘 I/O 瓶颈，会导致处理 Commit/Rollback 消息缓慢。
Producer 线程池阻塞: Producer 端用于发送 Commit/Rollback 消息的线程池被阻塞，无法及时发送消息。
本地事务执行时间过长: 如果本地事务执行时间过长，会导致 Producer 迟迟无法发送 Commit/Rollback 消息。
回查机制影响: 频繁的回查操作也会增加 Broker 的负载，甚至导致阻塞。
RocketMQ 配置不合理: 某些配置项，例如消息刷盘策略、线程池大小等，配置不合理也会影响性能。
MQ Client版本问题: 某些版本的Client可能存在Bug。
同步发送阻塞: 事务消息默认是同步发送的，如果Broker处理慢，Producer会被阻塞。

3. 定位问题的工具和方法

在排查问题时，我们需要借助一些工具和方法来定位瓶颈：

RocketMQ Dashboard: RocketMQ Dashboard 提供了丰富的监控数据，可以查看 Broker 的 CPU、内存、磁盘 I/O 等指标，以及消息的堆积情况、消费进度等。
日志分析: 分析 Producer 和 Broker 的日志，可以找到异常信息，例如网络错误、超时等。
JVM 监控: 使用 jstack、jstat 等工具，监控 Producer 端的 JVM 状态，例如线程状态、GC 情况等。
链路追踪: 使用 SkyWalking、Jaeger 等链路追踪工具，可以追踪整个事务消息的流程，找到延迟的环节。
压测: 通过压测工具模拟高并发场景，可以快速发现性能瓶颈。

4. 性能调优实战：解决阻塞的具体步骤

接下来，我们针对上面分析的原因，给出一些具体的性能调优方案。

4.1. 优化网络

优化网络环境: 确保 Producer 和 Broker 之间的网络稳定、带宽充足。可以考虑使用专线连接，或者优化网络配置。
减少网络传输量: 尽量减少消息的大小，例如使用压缩算法对消息进行压缩。
调整 TCP 参数: 调整 TCP 的一些参数，例如 tcp_keepalive_time、tcp_syn_retries 等，可以提高网络连接的稳定性。

4.2. 优化 Broker 性能

Broker 扩容: 如果 Broker 的资源不足，可以考虑扩容 Broker 集群，增加 CPU、内存、磁盘等资源。
优化 Broker 配置:
- 刷盘策略: 根据业务场景选择合适的刷盘策略。SYNC_FLUSH 模式可以保证消息的可靠性，但性能较低；ASYNC_FLUSH 模式性能较高，但可能存在数据丢失的风险。
- 线程池大小: 调整 Broker 的线程池大小，例如 storeThreadPoolNums、pullThreadPoolNums 等，可以提高 Broker 的并发处理能力。
- 内存管理: 合理配置 Broker 的内存管理参数，例如 maxMessageSize、flushCommitLogLeastPages 等，可以减少内存碎片，提高内存利用率。
监控 Broker 状态: 定期监控 Broker 的 CPU、内存、磁盘 I/O 等指标，及时发现并解决性能问题。

4.3. 优化 Producer 性能

优化本地事务: 尽量缩短本地事务的执行时间，例如优化 SQL 查询、减少远程调用等。
调整 Producer 线程池: 调整 Producer 端用于发送 Commit/Rollback 消息的线程池大小，确保有足够的线程来处理 Commit/Rollback 请求。

异步发送 Commit/Rollback 消息: 将 Commit/Rollback 消息的发送改为异步方式，可以避免阻塞本地事务的执行。可以通过 TransactionMQProducer.sendMessageInTransaction() 方法的返回值 LocalTransactionState 来判断事务状态，并异步发送 Commit/Rollback 消息。

public class TransactionProducer {

    public static void main(String[] args) throws MQClientException, InterruptedException {

        TransactionMQProducer producer = new TransactionMQProducer("transaction_group");
        producer.setNamesrvAddr("127.0.0.1:9876");

        ExecutorService executorService = Executors.newFixedThreadPool(5);
        producer.setExecutorService(executorService);

        TransactionListener transactionListener = new TransactionListenerImpl();
        producer.setTransactionListener(transactionListener);

        producer.start();

        String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
        for (int i = 0; i < 10; i++) {
            try {
                Message msg =
                    new Message("TopicTest", tags[i % tags.length], "KEY" + i,
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.sendMessageInTransaction(msg, null);
                System.out.printf("%s%n", sendResult);

                Thread.sleep(10);
            } catch (MQClientException | UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }

        Thread.sleep(Integer.MAX_VALUE);
        producer.shutdown();
    }
}

class TransactionListenerImpl implements TransactionListener {

    private AtomicInteger transactionIndex = new AtomicInteger(0);

    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        int value = transactionIndex.getAndIncrement();
        int status = value % 3;
        localTrans.put(msg.getTransactionId(), status);
        System.out.println("执行本地事务, transactionId:" + msg.getTransactionId() + ", status:" + status);

        // 模拟本地事务执行，根据status返回不同的状态
        if (status == 0) {
            // Return commit message
            return LocalTransactionState.COMMIT_MESSAGE;
        } else if (status == 1) {
            // Return rollback message
            return LocalTransactionState.ROLLBACK_MESSAGE;
        } else {
            // Return unknown
            return LocalTransactionState.UNKNOW;
        }
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        Integer status = localTrans.get(msg.getTransactionId());
        if (null != status) {
            switch (status) {
                case 0:
                    return LocalTransactionState.COMMIT_MESSAGE;
                case 1:
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                case 2:
                    return LocalTransactionState.UNKNOW;
            }
        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

进一步优化，在 executeLocalTransaction 方法中，使用单独的线程池异步发送 Commit/Rollback:

class TransactionListenerImpl implements TransactionListener {

    private AtomicInteger transactionIndex = new AtomicInteger(0);

    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();

    private final ExecutorService commitRollbackExecutor = Executors.newFixedThreadPool(10);

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        int value = transactionIndex.getAndIncrement();
        int status = value % 3;
        localTrans.put(msg.getTransactionId(), status);
        System.out.println("执行本地事务, transactionId:" + msg.getTransactionId() + ", status:" + status);

        // 模拟本地事务执行，根据status返回不同的状态
        // 使用线程池异步发送 Commit/Rollback
        commitRollbackExecutor.submit(() -> {
            if (status == 0) {
                // Return commit message
                // 实际场景中，这里需要调用 producer.endTransaction(msg, LocalTransactionState.COMMIT_MESSAGE)
                System.out.println("异步发送 Commit 消息, transactionId:" + msg.getTransactionId());
            } else if (status == 1) {
                // Return rollback message
                // 实际场景中，这里需要调用 producer.endTransaction(msg, LocalTransactionState.ROLLBACK_MESSAGE)
                System.out.println("异步发送 Rollback 消息, transactionId:" + msg.getTransactionId());
            } else {
                // Return unknown
                System.out.println("返回 Unknown 状态, transactionId:" + msg.getTransactionId());
            }
        });

        return LocalTransactionState.UNKNOW; // 这里返回 UNKNOW，依赖回查机制
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        Integer status = localTrans.get(msg.getTransactionId());
        if (null != status) {
            switch (status) {
                case 0:
                    return LocalTransactionState.COMMIT_MESSAGE;
                case 1:
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                case 2:
                    return LocalTransactionState.UNKNOW;
            }
        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

注意: 上面的代码只是示例，实际场景中需要根据具体的业务逻辑来处理 Commit/Rollback 消息的发送。 producer.endTransaction() 是关键。

批量发送消息: 如果需要发送大量的事务消息，可以考虑使用批量发送，减少网络开销。
调整消息发送超时时间: 调整 sendMessageInTransaction 方法的超时时间，避免长时间阻塞。
使用高性能序列化方式: 使用 Protobuf、Thrift 等高性能序列化方式，可以减少消息的大小，提高传输效率。

4.4. 优化回查机制

减少回查频率: 尽量缩短本地事务的执行时间，减少 Broker 发起回查的频率。可以通过优化本地事务逻辑，或者使用缓存等技术来提高性能。
优化回查逻辑: 优化 Producer 端的回查逻辑，确保回查操作能够快速完成。

调整回查参数: 调整 RocketMQ 的回查参数，例如 transactionTimeout、transactionCheckInterval 等，可以控制回查的频率和超时时间。

参数名	描述	默认值	建议
`transactionTimeout`	事务消息的超时时间，超过这个时间Broker会发起回查。	60 秒	根据实际业务场景调整。如果本地事务执行时间较长，可以适当增加超时时间。
`transactionCheckInterval`	Broker回查Producer的间隔时间。	1 分钟	根据业务调整，不建议过短。如果频繁回查，会增加Producer的负担。
`checkRequestHoldMax`	Broker允许hold的最大回查请求数量，超过这个数量则拒绝新的回查请求。	20000	如果并发量很高，回查次数很多，可以适当增加这个值。需要注意，增加这个值会占用更多的Broker资源。
`maxCheckTimes`	允许回查的最大次数，超过这个次数则丢弃消息。	15	根据业务调整。如果本地事务确实存在问题，多次回查仍然无法确定状态，则应该丢弃消息，避免无限回查。

避免长时间的UNKNOW状态: 尽量避免事务长时间处于 UNKNOW 状态，确保最终能够 Commit 或 Rollback。

4.5. 其他优化

升级 RocketMQ 版本: 新版本的 RocketMQ 通常会包含性能优化和 Bug 修复，升级到最新版本可以获得更好的性能。
调整 JVM 参数: 调整 Producer 和 Broker 的 JVM 参数，例如堆大小、GC 策略等，可以优化 JVM 的性能。
监控和报警: 建立完善的监控和报警机制，可以及时发现并解决性能问题。

5. 代码示例：异步发送 Commit/Rollback 消息的实现

以下是一个简单的代码示例，展示如何使用异步方式发送 Commit/Rollback 消息：

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;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class MyTransactionListener implements TransactionListener {

    private final ExecutorService executorService = Executors.newFixedThreadPool(5);

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 执行本地事务
        boolean success = doLocalTransaction(msg);

        // 异步发送 Commit/Rollback 消息
        executorService.submit(() -> {
            try {
                if (success) {
                    // 发送 Commit 消息
                    // producer.endTransaction(msg, LocalTransactionState.COMMIT_MESSAGE);  //  这里需要 producer 实例
                    System.out.println("Commit 消息发送成功");
                } else {
                    // 发送 Rollback 消息
                    // producer.endTransaction(msg, LocalTransactionState.ROLLBACK_MESSAGE); //  这里需要 producer 实例
                    System.out.println("Rollback 消息发送成功");
                }
            } catch (Exception e) {
                System.err.println("发送 Commit/Rollback 消息失败: " + e.getMessage());
            }
        });

        return LocalTransactionState.UNKNOW; // 返回 UNKNOW，等待回查
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 回查本地事务状态
        boolean success = checkLocalTransactionStatus(msg);
        if (success) {
            return LocalTransactionState.COMMIT_MESSAGE;
        } else {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }

    private boolean doLocalTransaction(Message msg) {
        // 模拟本地事务
        try {
            Thread.sleep(100); // 模拟耗时操作
            return true; // 假设事务成功
        } catch (InterruptedException e) {
            return false; // 事务失败
        }
    }

    private boolean checkLocalTransactionStatus(MessageExt msg) {
        // 模拟回查本地事务状态
        return true; // 假设事务成功
    }
}

注意: 这个代码只是一个示例，实际应用中需要根据具体的业务逻辑来修改。关键在于 executeLocalTransaction 方法中，使用 ExecutorService 异步发送 Commit/Rollback 消息。

6. 避免同步发送的阻塞

事务消息默认是同步发送的，这意味着 Producer 会阻塞等待 Broker 的响应。在高并发场景下，如果 Broker 处理缓慢，Producer 可能会被长时间阻塞，影响整体性能。

解决方案:

使用异步发送: 将事务消息的发送改为异步方式，可以避免 Producer 被阻塞。 RocketMQ 并没有直接提供事务消息的异步发送接口。但是可以通过上面提到的使用单独线程池发送 Commit/Rollback 的方式来间接实现异步发送，从而减少阻塞。

7. 总结：优化事务消息的性能，保障业务流畅运行

通过对网络、Broker、Producer、回查机制等方面的优化，我们可以有效地解决 RocketMQ 事务消息提交延迟导致业务阻塞的问题。关键在于：

定位瓶颈: 使用合适的工具和方法，找到性能瓶颈所在。
对症下药: 根据瓶颈原因，采取相应的优化措施。
监控和报警: 建立完善的监控和报警机制，及时发现并解决问题。

通过这些措施，我们可以提高 RocketMQ 事务消息的性能，保障业务的流畅运行。

RocketMQ 事务消息提交延迟导致业务阻塞的性能调优实战

发表回复 取消回复

发表回复取消回复