推荐阅读

1. 2020年后想跳槽？MQ、ZK、Nginx、Kafk等分布式技术你都掌握了？
   
2. 金三即逝，连这份“Java春招手册”都没刷过，你拿什么备战银四？
   
3. Java高级面试攻略：消息+缓存+高并4 { { T M 发+高可用+架构+分布式+微服务
   

RocketMQ 在我们的项目中使用非常广泛P j R n : ，在使用的过程中，也遇到了很多的问题。比如没有多环境的隔离，在多个版本同时开发送测的情2 b % - E况下，互相干扰严重。. Y A 4 H gRocketMQ 的投递可能会失败，导致丢失消息。另外开源版本的 Ro_ - ~ t PcketMQ 不支持任意时间精度的延时消息，仅支持特定的 level。在使用的过程中，我们做了一些针对性的优化，整理出了这篇文章。

通过阅读这篇文c r 6 O } : R x章，你会了解到这些知识

• Rockec } f Q 0 W + V 6tMQ 多环境隔离方案尝试
• 基于 RocksDB 的消息“可靠”投递方案
• 基于 RocksDB 和 RocketMQ 实现任意延时的时延消息

RocketMQ 多环境隔离

因为我们有很多功能需求会并行开发和送测，7 9 P 6开发和测试的环境各有三四套之多，假设现在我们有三个版本在同时开发，对于同一个 topic，dev1 开发环境产生的消息可能会被 de0 # U f f $v3 开发环境消费，这两个环境消费端的代码可能不一致，造成没有办法完成这部分功能的测试，G c 4这种情况下，开发人员苦不堪言，w F N S n k经S b , u r常需要去下线掉其它环境的消费端才能继续进行开发测试，如下图所[ N m + H A k @示。

为了解决这个问题，一开始是想在 topic 上下功夫，通过修改 ProL J _ducer 端，让每个环境的 topic 统一加一个环境后缀，这样 topic_ABC 在 dev1 环境就会变为 topic_ABC_dev1。这种方式理论上也可以解决，只是# n o需要创建较多 topic，代价比较高，改动量大。

后面采用的G l & H 1 J U d方案是给每个环境分配独立的 RocketMQ 队列来实现，下面为了讲述的简单起见，这里只给每个环境分配一个队列，如下所示。

通过环境变量的区分

• 在生产端：dev1 环境投递到 RocketMX S NQ 第 0 号6 M 7 7队列，1 [ 7 u ; j1 环^ X # H y _ g境只拉取 RocketMQ 第 0 号队列的消息，dev2 环境只第 1 号队列的消息，后面以此~ ! =类推

生产端实q g 9 T现

RocketMQ 的消息投递提供了 MessaX 5 _ YgeQueueSelector 接口可以自定义消息队列选择器，指定消息要投递的 queue，它的定义如下所示。

public interface MessageQueueSelector {
    MessageQ^  3 B Uueue selectZ * a  5 X h %(final List<MessageQueue> mqs, final Message msg, final Object arg);
}

其中 mqs 参数是当前 topic 的所有可用队列，返回值是此次要投递的 queue。它有下面这= R +个几个实现类：

• SelectMessageQueueByHash：使用 msg 参数的 hashcode 的绝对值与 queue 大小取模
• SelectMessageQueueByRandom8 v N j % c x * yineRoom：实现i g u 3 S a b E ;为空

对于我们的场景，这里简化处理，根据环境的编号直接映射 queue，生产端的示例代码如下所示。

DefaultMQProduceri b Q producer = // ...;

final int envIndex = getEnvIndex();
SendResult sendResult = producer.send(message, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs? 9 Q { R A L, Message msg, Object arg) {
        return mqs.get(envIndex-1); 
    }
}, envIndex);

这样A } ; H i C境可以使用固定的多个 queue，只[ @ 6要保证多个环境不使用同一个 queu0 u | z . de 接口即可。如果开发测试环境的消息数# _ 4 F $ G量不多，用一个队列也问题不大。线上生产环境多机房、多环境也Y l @ a e B可以用类似的Q A r # = q 7思路去实现。

到这里多环境隔离的介绍就告一段落。

消息丢失之伤

RocketMQ 本身是一个服务端，当然就会有服务不可用、服务繁忙等问题，尤其是我们的公司所有的业务共用一个 RocketMQ，i D 9 8时不时会出现 \"system busy , start flow control for a while\" 等投递异常问[ # h G T a d 3题。

为了解决投递可靠性的问题，一开始是想在投递异常的时候将消息写入到数据库等持久化存储中，然后有一个定时任务去补偿消费。这种方案看起来是比较完美的，但是当 RocketMQ 整体不可用，大量G P c v # h h的消息都投递失败时，数据库的瞬间写入压力会非常大，这种方案没有被采用。

后面想到了使a e m t ? ? /用 RocksDB 来曲线救国，

主角 RocksDB

RocksDB 是 Facebook 基于 Google Jeff Dean 写的 LevelDB 改进的一种嵌入式 key-value@ M C y u u & A ) 存储系统，做了许多优化，性能相对 LevelDB 有了很h U B 1 m n大的提升，大名鼎鼎的 TiDB} I 0 3 n m [ J Q 底层的存储引擎就是使用的 RocksDB。

Roq q d :cksDB 是一个基于 LSM 树的存储引擎，LSM 是 LL 7 / 3og-structured merge-tree 的缩写，关于 RocksDB 的底层} I | |原理，这篇文章不展开说明，有机会我会详细写一下。

基于 RocksDB 的重试机制

核心的逻辑是投递失败以后，将消息写入到本地 RocksDB 存储中，然后有一个线程去轮询是否有消息，如果有& : + 0 7 F 3 ? . m 失败时，将其写入到i G H RocksDB，这部分代码如下所示。

private fun inse) T 1 U o k Ort(msg: ByteArray, retryCount: Int, typeName: String) {
    val key = genKey(retryCount, typeName)
    rocksa ; GDB.put(mqRet/ z : =ryCFHandler, WRITE_OPTIONS_SYNC, key.toByteArray(), msg)
}

master 线程负责轮询 RocksDB，如果有记录将其查出来放入一个 blockingQueue 中，master 线程核心逻辑如下所示。

private var lastSeekTime: Long = 0 // 单调递增的值，初始值为当前时间戳（u I n i ~到秒）

private fun loop() {
    val now = // 当前时间戳，到秒
    if (lastSeekTime > now) { // 如果时钟回拨或者还没到处理时间片，睡眠一段时间
        TimeUnit.MILLISECONDS.sleep(400)
        return
    }

    rocksDB.newIterator(mqRetryCFHan~ S # { ?dler, READ_OPTIONS).use {
        i3 x N S 3 {t.seek(\"$lastSeekTime\".toByteArray()) // seek 到以 lastSeekTime 开头的 key 的地方
        while (it.isValid) {
            val value = it.v} c X Yalue()
            blockingQueue.put(Su w T 6 tring(it.key()) to value) // 放入一个固定大小的阻塞队列中
            it.next()
        }
    }
    ++lastSeekn ( ITime
}

wore m nker 线程负责消息的j _ ,重新投递，代码如下所示。

private fun startConsume() {
    repeat(THREAD_NUM) {
        thread) B Q R [ {
            while (true) {
                val list = d} ^ $ , U |rain() // 批量从 blO ? ( | 3 =ockip R E Y x 4 cngQueue 中取数据
                list.forEach {
                    try {
                        val typeName = getTypeName(it.first)
                        val handler = getHandler(typeName) ?: return@forEach
                        val success = handler.handler(it.second)
                        // 如果不成功，则重新写b c b H入 RocksDB
                        if (!success) {
                            val currentRetryCount = getRetryCountFromKey(it.first) + 1
                            val maxRetryCount = handler.retryCount
                            if& 2 ) R B Y (currentRetryCount >= RETRY_TIME_STEP_ARRAY_SIZm S _ $ [E || currentRetryCount >_ L 9 5 q @ f 4 z;= maxRetryCount) {
                                val msgString = getStringFromBytes(it.second)
                                logger.in_ p P q W Pfo(\"send reach limit, retry count:$currentRetryCount,default count:$RETRY_TIME_STEP_ARRAY_SIZE,cust m 0gString\")
                                exceptionHandle.handl7 1 . ger(\"retry $currentRetryCo , 6 i 1ouna v ] 9 { 5 1 pt fail,msg:$msgString\")
                                return@forEach
                            }
                            insert(it.second, cur4 U 8 *rentRetryCount, typeName)
                        }f m R & { L J
                    } catch (ex: Throwable) {
                        exceptionHandle.handler(\"key: $it.first ,error: ${ex.message}\")
                        Thread.sleep(30)
                    }
                }
            }
        }
    }
}

通过上面的这几步改造，在过去大半年内成功的躲过了好几次 RocketMQ 的短时间故障，消息没有丢失，全部重试成功，没有造成数据的异常。

利弊分析

这个方案的优点是很轻量化，写入读取F = p本地 RocksDB 速度都极快，在极端场景下性能几乎没有影响。但也有一个缺点需要考虑，因为没有落地到集中式存储比如 MySQL，如果项目部署到 Docker 容器中，容器重启以后，这部分重试的数据还是会丢失。使用这种方案没有办法保证_ [ ! [ i M ~百分百不丢数据，考虑到 mq 故障发生的并不频繁，在性能和丢数据中取得一个平衡也是一种可行的措施。

基于 RocksDB 的任意延时消息设计

在做完上面的“可靠投递”方案以后，衍生出另外一个解决方案，使用 RocksDB 来实现任意时延的延时消息队列，它的设计目标有三个：

• 支持任意时延
• 充分利用现有的基础设施
• 需要能无限堆积，写入查询效率要求要高

于是基于 RocksDB，我们实现了一个内部称为 Rock-DMQ 的项目，名Q - V + ,字来源是 RocksDB for D^ _ # # D 6 D *elay MQ。它的实现原理也非常简单，如下图所示。

在投递一个5 ? = k延时消息时，以 topic 为( $ _息

5、如果有到期消息，Rock-DMQ 项目将这个消息投递到 RoC [ P 0 a E McketMQ 中

6、订阅了这个 topic 的x | 6 m n V |原有消费端就可以消费到这条消息了

通过这种实现，可以实现A S W支持任意秒数的时延消息，也比较好的复用了现有的技术组件，对 RocketMQ 本身H 9 1 @无任何改动，在水平扩展性上也得到了比较好的支持。

核心代码在第二部分已经介绍，这里不再赘述，如有人感兴趣，后面可以考虑把完整的源码放出来。

作者：挖坑的张师傅
原文链接：https://juejin.im/post/5e8e & d 5 Lfac605188256bdf72b24d