上一篇文章后端多環境治理的實踐（一）我們介紹了什么是后端多環境治理，而且用springcloud實現了微服務調用的多環境治理。
實現了之后，我們微服務的流量調度如下圖：
但是mq的消費怎么辦，mq的消費是沒有辦法做自定義的負載均衡，消費邏輯是由mq來決定的，經過mq之后，流量調度就會失效。
default版本的score和v1版本的score會隨機消費mq消息。
一、公司的解決方案
先說一下，公司用的是rocketmq。
而且公司有專門的mq團隊，他們對mq做了各種的魔改，其中一個魔改就是所有的mq消息者不會直接連接mq，而且經過一個mq_proxy再連接到mq。
所以公司的mq的消費可以自定義自己的路由規則。
具體實現如下：
-
基准消息和feature消息復用topic。生產者會根據context，在消息的header里添加env屬性。
-
當feature consumer啟動時，會在proxy根據version注冊對應的filter
-
當feature consumer拉取消息，proxy會根據注冊的filter，將對應version的消息過濾出來加到本地緩存里
-
feature consumer拉取消息，會從proxy的本地緩存里poll消息
大家知道一下原理就好，對於一般的公司來說，不太有能力去修改mq源代碼，也不太可能有專門的基建團隊去打造一個mq_proxy中間件出來。
二、我的解決方案
我沒有能力魔改m，也沒能力造一個mq_proxy。
但是只要理解了解決方案，就能實現多環境的mq消費，只是實現的方式會low一點。
要實現mq的多環境流量調度，只要做四件事：
1、**每個微服務注冊啟動的時候，要帶上一個標記，標記自己當前的版本，消息發送時要把這個標記放到mq_message的header里面。
2、每個微服務啟動的時候，要向某個地方注冊一下，標記現在存在一個某個版本的消費者。
3、所有的mq消費要經過統一的代理。
4、消費代理者執行如下邏輯：“如果消息的版本與本地版本相同，則本地消費，如果消息的版本與本地版本不相同，則判斷是否存在對應版本的消費者，如果存在則轉發給對應版本的消費者，如果不存在則本地消費”。
三、具體代碼實踐
下面以rabbitmq為例子去實現多環境mq消費
最終實現效果如下：
oder-base發出的消息，user-base消費。
order-v1發出的消息，user-v1消費。
oder-v2發出的消息，user-base消費。
1、**每個微服務注冊啟動的時候，要帶上一個標記，標記自己當前的版本，消息發送時要把這個標記放到mqmessage的header里面。
上一期我們已經給每個實例標記了一個版本號，只要把這個版本號丟到mq的header里面，就能給這個message標記他的version。
2、每個微服務啟動的時候，要向某個地方注冊一下，標記現在存在一個某個版本的消費者。
這里實現的方法有很多，可以用zookeeper、redis，這里我利用rabbitmq的自動刪除隊列的特性，如果某個隊列是自動刪除隊列，而且曾經有消費者連接過他，當沒有消費者消費該queue時，該queue會自動刪除。
所以只要我們的featrue消費者啟動的時候，向rabbitmq注冊某個featrue特有的queue，根據這個特性，只要我們看某條queue是否存在，就知道是否有featrue消費者存在了。
3、所有的mq消費要經過統一的代理。
公司是通過統一的proxy做的，而我則可以通過sdk的方式，自定義原生的mq消費者的方式去做mq消費者的代理。
先寫一個抽象消費類
創建一個MyQueuelister去實現這個抽象類
spring容器啟動的時候，找出所以繼承了RabbitmqLister的bean。
然后手動注冊消費者
@Slf4j
@Configuration
public class RabbitmqListerConfig implements InitializingBean{
@Autowired
private List<RabbitmqLister> rabbitmqListerList;
@Autowired
private ConnectionFactory connectionFactory;
@Autowired
private Environment environment;
@Autowired
private ApplicationContext applicationContext;
@Override
public void afterPropertiesSet() throws Exception {
rabbitmqListerList.forEach(r->{
try {
String queueName = r.getQueueName();
Connection connection = connectionFactory.createConnection();
Channel channel = connection.createChannel(false);
RabbitAdmin rabbitAdmin = applicationContext.getBean(RabbitAdmin.class);
String version = environment.getProperty(Constont.VERSION);
if (!StringUtils.equals(version,Constont.DEFAULT_VERSION)){
queueName = queueName +"-"+version;
Queue queue = new Queue(queueName, true, false, true);
rabbitAdmin.declareQueue(queue);
}
Consumer consumerB = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body)
throws IOException {
try {
Map<String, Object> headers = properties.getHeaders();
String msgVersion = headers.get(Constont.VERSION)==null?Constont.DEFAULT_VERSION:headers.get(Constont.VERSION).toString();
String version = environment.getProperty(Constont.VERSION);
ObjectMapper objectMapper = new ObjectMapper();
String o = objectMapper.readValue(body, String.class);
if(StringUtils.equals(version,msgVersion)){
r.handler(o,properties);
channel.basicAck(envelope.getDeliveryTag(),false);
return;
}
String queueName = Constont.MY_QUEUE + "-" + msgVersion;
QueueInformation queueInfo = rabbitAdmin.getQueueInfo(queueName);
if (queueInfo == null || queueInfo.getConsumerCount() ==0) {
r.handler(o,properties);
channel.basicAck(envelope.getDeliveryTag(),false);
return;
}
channel.basicPublish("",queueName,properties,body);
channel.basicAck(envelope.getDeliveryTag(),false);
}catch (Exception e){
log.error("",e);
}
}
};
channel.basicConsume(queueName,consumerB);
} catch (IOException e) {
log.error("",e);
}
});
}
}
4、消費代理者執行如下邏輯：“如果消息的版本與本地版本相同，則本地消費，如果消息的版本與本地版本不相同，則判斷是否存在對應版本的消費者，如果存在則轉發給對應版本的消費者，如果不存在則本地消費”。
通過判斷是有個featrue-queue來判斷是否有fetrue-cosumer，如果有則把消費投遞到featrue-queue。
具體代碼實現細節在3里面。
四、驗證
在order服務，我們編程一個接口用來發送消息。
@GetMapping("/sendMq")
1、打包
mvn clean install -DskipTests
2、使用docker啟動rabbitmq
docker run -d --hostname my-rabbit --name rabbit -p 15672:15672 -p 5672:5672 rabbitmq:management
3、啟動eureka、gateway和3個版本的order服務
nohup java -jar -Dserver.port=8761 eureka/target/eureka-0.0.1-SNAPSHOT.jar >null 2>&1 &
nohup java -jar -Dserver.port=5000 gateway/target/gateway-0.0.1-SNAPSHOT.jar >null 2>&1 &
nohup java -jar -Dserver.port=8001 order/target/order-0.0.1-SNAPSHOT.jar >null 2>&1 &
nohup java -jar -Dversion=v1 -Dserver.port=8002 order/target/order-0.0.1-SNAPSHOT.jar >null 2>&1 &
nohup java -jar -Dversion=v2 -Dserver.port=8003 order/target/order-0.0.1-SNAPSHOT.jar >null 2>&1 &
4、開另外兩個終端，去啟動user服務，這里直接啟動用來看日志打印。
java -jar -Dserver.port=9001 user/target/user-0.0.1-SNAPSHOT.jar
java -jar -Dversion=v1 -Dserver.port=9002 user/target/user-0.0.1-SNAPSHOT.jar
5、發送請求驗證
curl --location --request GET 'localhost:5000/order/sendMq?msg=djb'
然后查看default的user服務
curl --location --request GET 'localhost:5000/order/sendMq?msg=djbv1' \
然后查看v1的user服務
curl --location --request GET 'localhost:5000/order/sendMq?msg=djbv2' \
然后查看default的user服務
五、最后
我們已經解決了后端多版本的rpc調用和mq消費調用了。
其實還有一個場景就是配置的多版本。
有些時候，我們需要defalut版本的實例和v1版本的實例使用不同的配置。
而springcloud-config是沒辦法完成這個需求的。
下一篇文章，我將會介紹后端多環境的情況下，怎么做到不同版本的實例使用不同版本的配置。
代碼地址：
關注“從零開始的it轉行生”，回復“多環境”獲取