建站
咨詢
售后
建站咨詢
新聞中心
大家好，我是君哥。
今天來聊一聊 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ā)。這里需要分不同場景進(jìn)行分析。
1、只升級(jí)消費(fèi)者
這是最簡單的情況，比如只有消費(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 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 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)于是廣播組：
- 兩個(gè)組都要對(duì)所有的消息進(jìn)行拉取，比如本來使用灰度發(fā)布計(jì)劃切 10% 的流量，但實(shí)際上全部流量都切過去了，只是根據(jù)屬性做了判斷。這讓消費(fèi)端整體承擔(dān)了兩倍的壓力；
- 因?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ù)雜的場景，不僅可以過濾 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 灰度消息的使用方法，場景比較簡單。對(duì)于全鏈路的復(fù)雜灰度場景，可以參考使用阿里的微服引擎 MSE。
新聞標(biāo)題：面試官：使用RocketMQ怎么進(jìn)行灰度發(fā)布？
文章路徑：http://www.5511xx.com/article/dpedeeg.html
