RocketMQ 在使用上的一些排坑和优化

public interface MessageQueueSelector {
    MessageQueue select(final List<MessageQueue> mqs, final Message msg, final Object arg);
}

DefaultMQProducer producer = // ...;

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

public interface AllocateMessageQueueStrategy {

    /**
     * Allocating by consumer id
     *
     * @param consumerGroup 当前 consumer group
     * @param currentCID 当前 consumer id
     * @param mqAll 当前 topic 的所有 queue 列表
     * @param cidAll 当前 consumer group 下所有的 consumer id set 集合
     * @return 根据策略给当前 consumer 分配的 queue 列表
     */
    List<MessageQueue> allocate(
        final String consumerGroup,
        final String currentCID,
        final List<MessageQueue> mqAll,
        final List<String> cidAll
    );

    /**
     * 策略算法名
     */
    String getName();
}

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MQConstant.MQ_CONSUMER_GROUP_NAME, null,
        new AllocateMessageQueueStrategy() {
            @Override
            public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll, List<String> cidAll) {
                List<MessageQueue> list = new ArrayList<>();
                list.add(mqAll.get(envIndex-1));
                return list;
            }

            @Override
            public String getName() {
                return "env-based";
            }
        });

expireTime:retryCount:typeName:uuid

val RETRY_TIME_STEP_ARRAY = arrayOf(
    3, 5, 30, 60, 120, 300, 480, 600, 900, 1800
)

val expire = System.currentTimeMillis() / 1000 + (RETRY_TIME_STEP_ARRAY.getOrNull(retryCount) ?: 10)

private fun insert(msg: ByteArray, retryCount: Int, typeName: String) {
    val key = genKey(retryCount, typeName)
    rocksDB.put(mqRetryCFHandler, WRITE_OPTIONS_SYNC, key.toByteArray(), msg)
}

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

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

    rocksDB.newIterator(mqRetryCFHandler, READ_OPTIONS).use {
        it.seek("$lastSeekTime".toByteArray()) // seek 到以 lastSeekTime 开头的 key 的地方
        while (it.isValid) {
            val value = it.value()
            blockingQueue.put(String(it.key()) to value) // 放入一个固定大小的阻塞队列中
            it.next()
        }
    }
    ++lastSeekTime
}

private fun startConsume() {
    repeat(THREAD_NUM) {
        thread {
            while (true) {
                val list = drain() // 批量从 blockingQueue 中取数据
                list.forEach {
                    try {
                        val typeName = getTypeName(it.first)
                        val handler = getHandler(typeName) ?: return@forEach
                        val success = handler.handler(it.second)
                        // 如果不成功，则重新写入 RocksDB
                        if (!success) {
                            val currentRetryCount = getRetryCountFromKey(it.first) + 1
                            val maxRetryCount = handler.retryCount
                            if (currentRetryCount >= RETRY_TIME_STEP_ARRAY_SIZE || currentRetryCount >= maxRetryCount) {
                                val msgString = getStringFromBytes(it.second)
                                logger.info("send reach limit, retry count:$currentRetryCount,default count:$RETRY_TIME_STEP_ARRAY_SIZE,custom count:$maxRetryCount, msg: $msgString")
                                exceptionHandle.handler("retry $currentRetryCount fail,msg:$msgString")
                                return@forEach
                            }
                            insert(it.second, currentRetryCount, typeName)
                        }
                    } catch (ex: Throwable) {
                        exceptionHandle.handler("key: $it.first ,error: ${ex.message}")
                        Thread.sleep(30)
                    }
                }
            }
        }
    }
}