Spring boot實戰專案整合阿里雲RocketMQ （非開源版）訊息佇列實現傳送普通訊息，延時訊息 --附程式碼
阿新 • • 發佈：2019-08-17
一.為什麼選擇RocketMQ訊息佇列？
- 首先RocketMQ是阿里巴巴自研出來的，也已開源。其效能和穩定性從雙11就能看出來，借用阿里的一句官方介紹：歷年雙 11 購物狂歡節零點千萬級 TPS、萬億級資料洪峰，創造了全球最大的業務訊息併發以及流轉紀錄（日誌類訊息除外）；
- 在始終保證高效能前提下，支援億級訊息堆積，不影響叢集的正常服務，在削峰填谷（蓄洪）、微服務解耦的場景下尤為重要；這，就能說明RocketMQ的強大。
二.RocketMQ的特點和優勢（可跳過看三的整合程式碼）
- 削峰填谷（主要解決諸如秒殺、搶紅包、企業開門紅等大型活動時皆會帶來較高的流量脈衝，或因沒做相應的保護而導致系統超負荷甚至崩潰，或因限制太過導致請求大量失敗而影響使用者體驗，海量訊息堆積能力強）
- 非同步解耦（高可用鬆耦合架構設計，對高依賴的專案之間進行解耦，當下遊系統出現宕機，不會影響上游系統的正常執行，或者雪崩）
- 順序訊息（順序訊息即保證訊息的先進先出，比如證券交易過程時間優先原則，交易系統中的訂單建立、支付、退款等流程，航班中的旅客登機訊息處理等）
-
分散式事務訊息（確保資料的最終一致性，大量引入 MQ 的分散式事務，既可以實現系統之間的解耦，又可以保證最終的資料一致性，減少系統間的互動）
三.SpringBoot 整合RocketMQ（商業雲端版）
- 首先去阿里雲控制檯建立所需訊息佇列資源，包括訊息佇列 RocketMQ 的例項、Topic、Group ID (GID)，以及鑑權需要的 AccessKey（AK）。
- 在springboot專案pom.xml新增需要的依賴 ons-client v1.8.0.Final
<!-- RocketMQ --> <dependency> <groupId>com.aliyun.openservices</groupId> <artifactId>ons-client</artifactId> <version>1.8.0.Final</version> </dependency>
- 在對應環境的application-xx.properties檔案配置引數
##-------鑑權需要的 AccessKey（AK）--- rocketmq.accessKey=xxAxxxxxxxxxx rocketmq.secretKey=xxxxxxxxxiHxxxxxxxxxxxxxx
## 例項TCP 協議公網接入地址 rocketmq.nameSrvAddr=http://MQ_INST_***********85_BbM********************yuncs.com:80 #普通訊息topic rocketmq.topic=common rocketmq.groupId=GID-message rocketmq.tag=* #定時/延時訊息 rocketmq.timeTopic=time-lapse rocketmq.timeGroupId=GID-message rocketmq.timeTag=*
- 封裝MQ配置類：MqConfig
/** * MQ配置載入 * @author laifuwei */ @Configuration @ConfigurationProperties(prefix = "rocketmq") public class MqConfig { private String accessKey; private String secretKey; private String nameSrvAddr; private String topic; private String groupId; private String tag; private String timeTopic; private String timeGroupId; private String timeTag; public Properties getMqPropertie() { Properties properties = new Properties(); properties.setProperty(PropertyKeyConst.AccessKey, this.accessKey); properties.setProperty(PropertyKeyConst.SecretKey, this.secretKey); properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, this.nameSrvAddr); //設定傳送超時時間，單位毫秒 properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "4000"); return properties; } public String getAccessKey() { return accessKey; } public void setAccessKey(String accessKey) { this.accessKey = accessKey; } public String getSecretKey() { return secretKey; } public void setSecretKey(String secretKey) { this.secretKey = secretKey; } public String getNameSrvAddr() { return nameSrvAddr; } public void setNameSrvAddr(String nameSrvAddr) { this.nameSrvAddr = nameSrvAddr; } public String getTopic() { return topic; } public void setTopic(String topic) { this.topic = topic; } public String getGroupId() { return groupId; } public void setGroupId(String groupId) { this.groupId = groupId; } public String getTag() { return tag; } public void setTag(String tag) { this.tag = tag; } public String getTimeTopic() { return timeTopic; } public void setTimeTopic(String timeTopic) { this.timeTopic = timeTopic; } public String getTimeGroupId() { return timeGroupId; } public void setTimeGroupId(String timeGroupId) { this.timeGroupId = timeGroupId; } public String getTimeTag() { return timeTag; } public void setTimeTag(String timeTag) { this.timeTag = timeTag; } }View Code
- 給訊息生產者注入配置資訊，
ProducerBean用於將Producer整合至Spring Bean中/** * MQ配置注入生成訊息例項 */ @Configuration public class ProducerClient { @Autowired private MqConfig mqConfig; @Bean(initMethod = "start", destroyMethod = "shutdown") public ProducerBean buildProducer() { //ProducerBean用於將Producer整合至Spring Bean中 ProducerBean producer = new ProducerBean(); producer.setProperties(mqConfig.getMqPropertie()); return producer; } }
- 為了方便使用，我封裝了一個傳送訊息的類，訊息的Message引數和配置，看程式碼註釋，很容易理解
/** * MQ傳送訊息助手 * @author laifuwei */ @Component public class ProducerUtil { private Logger logger = LoggerFactory.getLogger(ProducerUtil.class); @Autowired private MqConfig config; @Autowired private ProducerBean producer; /** * 同步傳送訊息 * @param msgTag 標籤，可用於訊息小分類標註 * @param messageBody 訊息body內容，生產者自定義內容 * @param msgKey 訊息key值，建議設定全域性唯一，可不傳，不影響訊息投遞 * @return success:SendResult or error:null */ public SendResult sendMsg(String msgTag,byte[] messageBody,String msgKey) { Message msg = new Message(config.getTopic(),msgTag,msgKey,messageBody); return this.send(msg,Boolean.FALSE); } /** * 同步傳送定時/延時訊息 * @param msgTag 標籤，可用於訊息小分類標註，對訊息進行再歸類 * @param messageBody 訊息body內容，生產者自定義內容，二進位制形式的資料 * @param msgKey 訊息key值，建議設定全域性唯一值，可不設定，不影響訊息收發 * @param delayTime 服務端傳送訊息時間，立即傳送輸入0或比更早的時間 * @return success:SendResult or error:null */ public SendResult sendTimeMsg(String msgTag,byte[] messageBody,String msgKey,long delayTime) { Message msg = new Message(config.getTimeTopic(),msgTag,msgKey,messageBody); msg.setStartDeliverTime(delayTime); return this.send(msg,Boolean.FALSE); } /** * 傳送單向訊息 */ public void sendOneWayMsg(String msgTag,byte[] messageBody,String msgKey) { Message msg = new Message(config.getTopic(),msgTag,msgKey,messageBody); this.send(msg,Boolean.TRUE); } /** * 普通訊息傳送發放 * @param msg 訊息 * @param isOneWay 是否單向傳送 */ private SendResult send(Message msg,Boolean isOneWay) { try { if(isOneWay) { //由於在 oneway 方式傳送訊息時沒有請求應答處理，一旦出現訊息傳送失敗，則會因為沒有重試而導致資料丟失。 //若資料不可丟，建議選用同步或非同步傳送方式。 producer.sendOneway(msg); success(msg, "單向訊息MsgId不返回"); return null; }else { //可靠同步傳送 SendResult sendResult = producer.send(msg); //獲取傳送結果，不拋異常即傳送成功 if (sendResult != null) { success(msg, sendResult.getMessageId()); return sendResult; }else { error(msg,null); return null; } } } catch (Exception e) { error(msg,e); return null; } } //對於使用非同步介面，可設定單獨的回撥處理執行緒池，擁有更靈活的配置和監控能力。 //根據專案需要，伺服器配置合理設定執行緒數，執行緒太多有OOM 風險， private ExecutorService threads = Executors.newFixedThreadPool(3); //僅建議執行輕量級的Callback任務，避免阻塞公共執行緒池 引起其它鏈路超時。 /** * 非同步傳送普通訊息 * @param msgTag * @param messageBody * @param msgKey */ public void sendAsyncMsg(String msgTag,byte[] messageBody,String msgKey) { producer.setCallbackExecutor(threads); Message msg = new Message(config.getTopic(),msgTag,msgKey,messageBody); try { producer.sendAsync(msg, new SendCallback() { @Override public void onSuccess(final SendResult sendResult) { assert sendResult != null; success(msg, sendResult.getMessageId()); } @Override public void onException(final OnExceptionContext context) { //出現異常意味著傳送失敗，為了避免訊息丟失，建議快取該訊息然後進行重試。 error(msg,context.getException()); } }); } catch (ONSClientException e) { error(msg,e); } } //--------------日誌列印---------- private void error(Message msg,Exception e) { logger.error("傳送MQ訊息失敗-- Topic:{}, Key:{}, tag:{}, body:{}" ,msg.getTopic(),msg.getKey(),msg.getTag(),new String(msg.getBody())); logger.error("errorMsg --- {}",e.getMessage()); } private void success(Message msg,String messageId) { logger.info("傳送MQ訊息成功 -- Topic:{} ,msgId:{} , Key:{}, tag:{}, body:{}" ,msg.getTopic(),messageId,msg.getKey(),msg.getTag(),new String(msg.getBody())); } }
- 前面已經配置好了將Producer整合至Spring Bean中，直接注入Producer，在業務系統需要的地方呼叫來發送訊息即可
//普通訊息的Producer 已經註冊到了spring容器中，後面需要使用時可以 直接注入到其它類中 @Autowired private ProducerBean producer; /** * 演示方法，可在自己的業務系統方法中進行傳送訊息 */ public String mqTest() { /* 使用前面封裝的方法，傳入對應的引數即可傳送訊息 * msgTag 標籤，可用於訊息小分類標註 * messageBody 訊息body內容，生產者自定義內容,任何二進位制資料，生產者和消費者協定資料的序列化和反序列化 * msgKey 訊息key值，建議設定全域性唯一，可不傳，不影響訊息投遞 */ //body內容自定義 JSONObject body = new JSONObject(); body.put("userId", "this is userId"); body.put("notice", "同步訊息"); //同步傳送訊息 producer.sendMsg("userMessage", body.toJSONString().getBytes(), "messageId"); //單向訊息 producer.sendOneWayMsg("userMessage", "單向訊息".getBytes(), "messageId"); //非同步訊息 producer.sendAsyncMsg("userMessage", "非同步訊息".getBytes(), "messageId"); //定時/延時訊息,當前時間的30秒後推送。時間自己定義 producer.sendTimeMsg("userMessage", "延時訊息".getBytes(), "messageId", System.currentTimeMillis()+30000); //順序訊息（全域性順序 / 分割槽順序）、分散式事務訊息 目前沒用到，可看官網說明操作 return "ok"; }
- 接下來是訊息消費者的配置和接收訊息（一般在下游系統或者相關聯的系統），接收訊息的專案照舊，新增依賴jar包 ons-client v1.8.0.Final 、配置mq引數連結（mq的配置檔案引數要和生產者專案配置的一樣）、新增MqConfig類（上面有寫）
- 注入配置、訂閱訊息、新增訊息處理的方法
@Configuration public class ConsumerClient { @Autowired private MqConfig mqConfig; //普通訊息監聽器，Consumer註冊訊息監聽器來訂閱訊息. @Autowired private MqMessageListener messageListener; //定時訊息監聽器，Consumer註冊訊息監聽器來訂閱訊息. @Autowired private MqTimeMessageListener timeMessageListener; @Bean(initMethod = "start", destroyMethod = "shutdown") public ConsumerBean buildConsumer() { ConsumerBean consumerBean = new ConsumerBean(); //配置檔案 Properties properties = mqConfig.getMqPropertie(); properties.setProperty(PropertyKeyConst.GROUP_ID, mqConfig.getGroupId()); //將消費者執行緒數固定為20個 20為預設值 properties.setProperty(PropertyKeyConst.ConsumeThreadNums, "20"); consumerBean.setProperties(properties); //訂閱訊息 Map<Subscription, MessageListener> subscriptionTable = new HashMap<Subscription, MessageListener>(); //訂閱普通訊息 Subscription subscription = new Subscription(); subscription.setTopic(mqConfig.getTopic()); subscription.setExpression(mqConfig.getTag()); subscriptionTable.put(subscription, messageListener); //訂閱定時/延時訊息 Subscription subscriptionTime = new Subscription(); subscriptionTime.setTopic(mqConfig.getTimeTopic()); subscriptionTime.setExpression(mqConfig.getTimeTag()); subscriptionTable.put(subscriptionTime, timeMessageListener); consumerBean.setSubscriptionTable(subscriptionTable); return consumerBean; } } - 對訊息監聽類進行實現，處理接收到的訊息
/** * 定時/延時MQ訊息監聽消費 * @author laifuwei */ @Component public class MqTimeMessageListener extends AutowiredService implements MessageListener { private Logger logger = LoggerFactory.getLogger(this.getClass()); //實現MessageListtener監聽器的消費方法 @Override public Action consume(Message message, ConsumeContext context) {logger.info("接收到MQ訊息 -- Topic:{}, tag:{},msgId:{} , Key:{}, body:{}",
message.getTopic(),message.getTag(),message.getMsgID(),message.getKey(),new String(message.getBody()));try {
String msgTag = message.getTag();//訊息型別 String msgKey = message.getKey();//業務唯一id switch (msgTag) { //----通過生產者傳的tag標籤進行訊息分類和過濾處理 case "userMessage": //通過唯一key的，查詢需要處理的資料, GroupChatMessage chatMessage = chatMessageService.getById(msgKey); //由於RocketMQ能重複推送訊息，處理訊息的時候做好資料的冪等，防止重複處理 if( chatMessage.isRead() ) { break; } //驗證通過，處理業務 //do something break; } //消費成功，繼續消費下一條訊息 return Action.CommitMessage; } catch (Exception e) { logger.error("消費MQ訊息失敗！ msgId:" + message.getMsgID()+"----ExceptionMsg:"+e.getMessage()); //消費失敗，告知伺服器稍後再投遞這條訊息，繼續消費其他訊息 return Action.ReconsumeLater; } } }
四.最後執行消費者專案和生產者專案，呼叫生產者專案傳送訊息驗證效果：
- 生產者傳送訊息結果日誌：訊息傳送正常
2019-08-17 15:11:06.837 INFO 9996 --- [nio-8080-exec-9] com.dyj.shop.mq.ProducerUtil : 傳送MQ訊息成功 -- Topic:common ,msgId:C0A86532270C2A139A5555A7E5DD0000 , Key:messageId, tag:userMessage, body:{"userId":"this is userId","notice":"同步訊息"} 2019-08-17 15:11:06.841 INFO 9996 --- [nio-8080-exec-9] com.dyj.shop.mq.ProducerUtil : 傳送MQ訊息成功 -- Topic:common ,msgId:單向訊息MsgId不返回 , Key:messageId, tag:userMessage, body:單向訊息 2019-08-17 15:11:06.901 INFO 9996 --- [pool-6-thread-1] com.dyj.shop.mq.ProducerUtil : 傳送MQ訊息成功 -- Topic:common ,msgId:C0A86532270C2A139A5555A7E6630004 , Key:messageId, tag:userMessage, body:非同步訊息 2019-08-17 15:11:07.060 INFO 9996 --- [nio-8080-exec-9] com.dyj.shop.mq.ProducerUtil : 傳送MQ訊息成功 -- Topic:time-lapse ,msgId:C0A86532270C2A139A5555A7E69F0006 , Key:messageId, tag:userMessage, body:定時/延時訊息
- 消費者接收到訊息，可以看到普通訊息的傳送時間和接收到訊息的時間，就相差幾毫秒，值得注意的是：延時訊息按照生產者定義的30秒後消費者才收到。這就是延時訊息的好玩之處
2019-08-17 15:11:06.881 INFO 10942 --- [MessageThread_7] com.dyj.timer.mq.MqMessageListener : 接收到MQ訊息. Topic :common, tag :userMessage msgId : C0A86532270C2A139A5555A7E5DD0000, Key :messageId, body:{"userId":"this is userId","notice":"同步訊息"} 2019-08-17 15:11:06.934 INFO 10942 --- [MessageThread_8] com.dyj.timer.mq.MqMessageListener : 接收到MQ訊息. Topic :common, tag :userMessage msgId : C0A86532270C2A139A5555A7E6550002, Key :messageId, body:單向訊息 2019-08-17 15:11:06.947 INFO 10942 --- [MessageThread_9] com.dyj.timer.mq.MqMessageListener : 接收到MQ訊息. Topic :common, tag :userMessage msgId : C0A86532270C2A139A5555A7E6630004, Key :messageId, body:非同步訊息 2019-08-17 15:11:36.996 INFO 10942 --- [essageThread_10] com.dyj.timer.mq.MqTimeMessageListener : 接收到MQ訊息. Topic :time-lapse, tag :userMessage msgId : cd900e16f7cba68369ec498ae2f9dd6c, Key :messageId, body:定時/延時消
寫在最後：有不妥或有興趣的可以下方留言，多謝指教(#^