大家好，我是君哥。

今天來聊一聊 RocketMQ 的灰度方案。

灰度發(fā)布是指在黑與白之間，平滑過渡的一種發(fā)布方式。在大流量的系統(tǒng)中，如果一次升級(jí)改造范圍比較大，或者影響內(nèi)容不太確定，一般會(huì)采用切量的方式進(jìn)行升級(jí)，這樣可以減少生產(chǎn)變更帶來的影響。

如上圖，對(duì) ServiceA 這個(gè)服務(wù)進(jìn)行升級(jí)，采用灰度發(fā)布，先升級(jí) Server5，一周后如果沒有問題，升級(jí) Server4 和 Server 3，再運(yùn)行一周沒有問題，把剩下兩個(gè)節(jié)點(diǎn)都升級(jí)。

上面的案例是一個(gè) RPC 的調(diào)用。但如果使用消息隊(duì)列該怎么做呢？使用消息隊(duì)列，并不能使用網(wǎng)關(guān)來進(jìn)行流量轉(zhuǎn)發(fā)。這里需要分不同場(chǎng)景進(jìn)行分析。

1、只升級(jí)消費(fèi)者

這是最簡(jiǎn)單的情況，比如只有消費(fèi)者修改了消費(fèi)邏輯，就是 RPC 調(diào)用的情況類似，我們只要把消費(fèi)者進(jìn)行灰度發(fā)布就可以。如下圖：

2、生產(chǎn)者也升級(jí)

下面是一個(gè)訂單的實(shí)體類，我們新加了一個(gè)屬性，訂單生成時(shí)間：

public class Order {
    private Long id;
    private Long userId;
    private Long productId;
    private Integer count;
    private BigDecimal payAmount;
    /**訂單狀態(tài)：0：創(chuàng)建中；1：已完結(jié)*/
    private Integer status;
    /**新加屬性，訂單生成時(shí)間*/
    private String createTime;
}

消費(fèi)端的改造是需要對(duì) createTime 這個(gè)屬性進(jìn)行處理。

（1）消費(fèi)端過濾

在生產(chǎn)者的 Order 類中增加 createTime 屬性，如果我們直接使用 createTime 屬性來過濾，消費(fèi)者并不能實(shí)現(xiàn)灰度，因?yàn)樗械南M(fèi)者都可能會(huì)拉取到帶有 createTime 屬性的消息。

RocketMQ 中 Message 的定義如下：

public class Message implements Serializable {
    private String topic;
    private int flag;
    private Map<String, String> properties;
    private byte[] body;
    private String transactionId;
}

可以在 properties 屬性中增加一個(gè)灰度標(biāo)識(shí)，比如生產(chǎn)者發(fā)送消息的時(shí)候封裝如下：

Message msg = buildMessage(topic);
msg.putUserProperty("gray", "true");

注意：也可以在 SendMessageHook 這個(gè)鉤子函數(shù)中定義。通過這種方式可以在消費(fèi)端新增加一個(gè)灰度 Consumer Group，用來對(duì)灰度消息則進(jìn)行消費(fèi)。如下圖：

對(duì)于灰度 Consumer Group 判斷到 gray 屬性是 true 時(shí)進(jìn)行消費(fèi)，而對(duì)于普通 Consumer Group，判斷到 gray 屬性不等于 true 時(shí)再進(jìn)行消費(fèi)。這里可以借助 RocketMQ 客戶端的 FilterMessageHook，代碼如下：

defaultMQPushConsumerImpl.registerFilterMessageHook(new FilterMessageHook() {
 @Override
 public String hookName(){
  return "filterHook";
 }

 @Override
 public void filterMessage(FilterMessageContext context){
  List<MessageExt> messages = context.getMsgList();
                context.setMsgList(messages.stream().filter(m -> StringUtils.equals(m.getProperty("gray"),"true"))
                        .collect(Collectors.toList()));
 }
});

?不過這樣會(huì)有兩個(gè)問題，灰度和正常的兩個(gè) Consumer Group 相當(dāng)于是廣播組：

1. 兩個(gè)組都要對(duì)所有的消息進(jìn)行拉取，比如本來使用灰度發(fā)布計(jì)劃切 10% 的流量，但實(shí)際上全部流量都切過去了，只是根據(jù)屬性做了判斷。這讓消費(fèi)端整體承擔(dān)了兩倍的壓力；
2. 因?yàn)閮蓚€(gè)消費(fèi)者組都要去 Broker 拉取消息，Broker 的壓力也增加了一倍。

（2）Broker 過濾

使用 tag 過濾

如果一個(gè) Consumer 不訂閱一個(gè) Topic 中的全部消息，可以通過 Tag 來過濾。比如一個(gè) Consumer 訂閱了 TopicA 這個(gè) Topic 中的 Tag1 和 Tag2 這兩個(gè) tag，那這個(gè) Consumer 的訂閱關(guān)系如下圖：

SubscriptionData 這個(gè)對(duì)象封裝了 Topic、tag 以及所訂閱 tag 的 hashcode 集合。

Consumer 發(fā)送拉取消息請(qǐng)求時(shí)，會(huì)把訂閱關(guān)系傳給 Broker（Broker 解析成 SubscriptionData 對(duì)象），Broker 使用 consumequeue 獲取消息時(shí)，首先判斷判斷最后 8 個(gè)字節(jié)的 tag hashcode 是否在 SubscriptionData 的 codeSet 中，如果不在就跳過，如果存在把消息返回給 Consumer。如下圖：

這樣可以在灰度 Producer 發(fā)送消息時(shí)加上 Tag，如下代碼：

Message msg = new Message();
msg.setBody("Test");
msg.setTopic("Topic");
msg.setTags("Gray");

而在灰度消費(fèi)者訂閱 Gray 這個(gè) tag。這樣就避免了 2.1 節(jié)中消息全量拉取的問題。

使用 SQL92 過濾

使用 SQL92 過濾，可以應(yīng)對(duì)更加復(fù)雜的場(chǎng)景，不僅可以過濾 Tag，還可以過濾 UserProperty。

比如下面是一個(gè)生產(chǎn)者的代碼：

Message msg = new Message();
msg.setTopic("testTopic");
msg.setTags("tag1");
msg.putUserProperty("gray","true");

這樣消費(fèi)者初始化的時(shí)候，可以定義使用 SQL92 過濾，代碼如下：

consumer.subscribe("testTopic",
            MessageSelector.bySql("(TAGS is not null and TAGS in TAGS='''''tag1''''')" +
                "and (gray is not null gray='true')"));

下面是 bySql 的源代碼：

public static MessageSelector bySql(String sql){
 return new MessageSelector(ExpressionType.SQL92, sql);
}

3、總結(jié)

本文介紹了 RocketMQ 灰度消息的使用方法，場(chǎng)景比較簡(jiǎn)單。對(duì)于全鏈路的復(fù)雜灰度場(chǎng)景，可以參考使用阿里的微服引擎 MSE。