2018 年，做為架構(gòu)負(fù)責(zé)人，接到一個(gè)架構(gòu)需求：實(shí)現(xiàn)一個(gè)簡單易用的 RocketMQ SDK 。

因?yàn)楦鱾€(gè)團(tuán)隊(duì) RocketMQ 原生客戶端配置起來千奇百怪，有的配置存在風(fēng)險(xiǎn)，各團(tuán)隊(duì)負(fù)責(zé)人都需要一個(gè)簡潔易用的 RocketMQ SDK 。

我立馬調(diào)研相關(guān)開源的方案，當(dāng)時(shí) RocketMQ-Spring 項(xiàng)目并沒有開源，而阿里云的 ONS SDK 是開源的，我只能講目標(biāo)轉(zhuǎn)向 阿里云 ONS 。

通過學(xué)習(xí) ONS 的設(shè)計(jì)方式，我對于 RocketMQ 的客戶端原理有了進(jìn)一步了解，也實(shí)現(xiàn)了公司內(nèi)部使用的 RocketMQ  SDK 。

圖片

項(xiàng)目地址：https://github.com/makemyownlife/platform-rocketmq

之所以說簡單，就是讓用戶（開發(fā)者）使用 SDK 時(shí)，減少心智負(fù)擔(dān)。

舉三個(gè)例子：

1 發(fā)送順序消息

使用原生代碼發(fā)送消息時(shí)，會(huì)使用如下的代碼：

SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        Integer id = (Integer) arg;
        int index = id % mqs.size();
        return mqs.get(index);
    }
}, orderId);

我們可以將 SDK API 簡化為：

SendResult send(final ProducerMessage message, final String shardingKey);

開發(fā)者不需要定義隊(duì)列選擇器，只需要傳遞分片鍵 orderId 即可。

2 單條消息消費(fèi)

使用原來代碼定義消費(fèi)監(jiān)聽器時(shí)，使用如下的代碼：

consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
        // 返回消息消費(fèi)狀態(tài)，ConsumeConcurrentlyStatus.CONSUME_SUCCESS為消費(fèi)成功
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});

監(jiān)聽器內(nèi)部，對于開發(fā)者操作的對象是消息列表 msgs ，很多開發(fā)同學(xué)想只操作一條消息。

于是，我們可以將 SDK API 簡化為：

consumer.subscribe("mytest", new ConsumerListener() {
    @Override
    public ConsumerAction consumer(ConsumerMessage msg) {
        byte[] body = msg.getBody();
        System.out.println("msg:" + new String(body));
        return ConsumerAction.CommitMessage;
    }
});

開發(fā)者在消費(fèi)時(shí)，可以一條一條操作，代碼簡潔了不少。

同時(shí)，很多開發(fā)者在使用普通消費(fèi)、順序消費(fèi)時(shí)，需要返回延時(shí)消費(fèi)的狀態(tài)碼時(shí)，兩種消費(fèi)模式定義的枚舉也不相同。我們將枚舉做了統(tǒng)一：

/**
 * 消費(fèi)消息的返回結(jié)果
 */
public enum ConsumerAction {

    /**
     * 消費(fèi)成功，繼續(xù)消費(fèi)下一條消息
     */
    CommitMessage,
    
    /**
     * 消費(fèi)失敗，告知服務(wù)器稍后再投遞這條消息，繼續(xù)消費(fèi)其他消息
     */
    ReconsumeLater;
}

3 訂閱關(guān)系一致

實(shí)際場景里，訂閱關(guān)系不一致是極容易發(fā)生的事情，就算是高級(jí)別的架構(gòu)師也會(huì)翻車，每次翻車現(xiàn)場都是慘不忍睹。

正確的訂閱關(guān)系見下圖：

圖片

正確的訂閱關(guān)系

代碼邏輯角度來看，每個(gè)消費(fèi)者實(shí)例內(nèi)訂閱方法的主題、 TAG、監(jiān)聽邏輯都需要保持一致。

圖片

當(dāng)訂閱關(guān)系不一致時(shí)，在 Broker 端同一個(gè)消費(fèi)組內(nèi)的各個(gè)消費(fèi)者客戶端的訂閱信息相互被覆蓋，從而導(dǎo)致某個(gè)消費(fèi)者客戶端無法拉取到新的消息。

怎么解決呢 ？

我當(dāng)時(shí)想起了阿里技術(shù)專家沈詢的一句話：

世界上解決一個(gè)計(jì)算機(jī)問題最簡單的方法：“恰好”不需要解決它 ！

公司內(nèi)部出現(xiàn)訂閱關(guān)系一致99%的問題是：消費(fèi)者組一致的前提下，主題相同，但 TAG 不相同。

基于此，我的設(shè)計(jì)思路就明確了：不開放訂閱 TAG 的權(quán)限！

沒想到吧，我就是這么粗暴。

按照這種設(shè)計(jì)思路，雖然開始有的程序員會(huì)有質(zhì)疑，但你和他梳理好消費(fèi)者組的定義，以及做好領(lǐng)域劃分，對業(yè)務(wù)來講，反而清晰了。