RocketMQ的傳送模式和消費模式
前言
小夥伴們大家好啊,王子又來和大家一起閒談MQ技術了。
通過之前文章的學習,我們已經對RocketMQ的基本架構有了初步的瞭解,那今天王子就和大家一起來點實際的,用程式碼和大家一起看看RocketMQ的幾種傳送模式和消費模式。好了,讓我們開始吧。
RocketMQ的環境準備
本次我們的環境搭建是基於docker的,需要小夥伴們準備一臺CentOS7的虛擬機器(虛擬機器的安裝這裡就不做介紹了)。
王子使用的是安裝在VMware中的CentOS7虛擬機器系統。正式安裝環境之前,為了測試方便,我們關閉了centos系統的防火牆。然後我們需要在這臺虛擬機器中安裝docke和docker-compose,具體的安裝步驟大家可以參考
安裝過docker和docker-compose後,我們就要開始安裝RocketMQ的環境了,具體的安裝步驟可以參考https://gitee.com/lm970585581/docker-rocketmq中的一鍵部署docker-compose部分。
步驟都很簡單,基本上算是一鍵式傻瓜安裝了,但是,王子在安裝的過程中卻被一個問題困擾住了很久。
下邊給大家說一下遇到的問題和解決的辦法,這是重點哦。
王子按照步驟一步一步的執行,到最後一步執行start.sh的時候悲劇發生了。
如果一切正常,執行了這一步驟後,docker中一共會啟動三個容器,分別是nameserver、broker、和rocketMQ的管控臺,但是王子執行docker ps檢視正在執行的容器的時候發現,居然只有兩個容器在執行,broker不見了。
於是開啟localhost:8180(管控臺地址),發現管控臺中是空的。
於是我們開始檢視broker容器的啟動日誌,使用的命令是:docker logs -f -t --tail 行數 容器名
發現的報錯資訊是 java.io.FileNotFoundException: /etc/rocketmq/broker.conf (Permission denied)
到這裡真實丈二的和尚摸不著頭腦,一臉懵逼有木有。
進過一番努力的查詢,終於找到了原因,由於centos7中安全模組selinux把許可權禁掉了,所以導致不能讀取broker.conf檔案,具體這個安全模組是做什麼的王子也沒有仔細研究,我們目前是用不到,於是在linux中使用setenforce 0命令把它關閉了。
再重新啟動broker容器,發現問題解決了。
好了環境我們部署完成之後,進入我們今天的主要內容,RocketMQ的傳送模式和消費模式有哪幾種。
RocketMQ的傳送
同步傳送
我們先來看一段生產者的程式碼:
import com.alibaba.rocketmq.client.exception.MQClientException; import com.alibaba.rocketmq.client.producer.DefaultMQProducer; import com.alibaba.rocketmq.client.producer.SendResult; import com.alibaba.rocketmq.common.message.Message; import com.alibaba.rocketmq.remoting.common.RemotingHelper; public class RocketMQProducer { // RocketMQ的生產者類 private static DefaultMQProducer producer; static { // 構建生產者物件,指定生產組 producer=new DefaultMQProducer("test_group"); // 設定NameServer的地址,拉取路由資訊 producer.setNamesrvAddr("192.168.220.110:9876"); try { // 啟動生產者 producer.start(); } catch (MQClientException e) { e.printStackTrace(); } } public static void send(String topic,String message) throws Exception { // 構建訊息物件 Message msg=new Message(topic, "",//這裡存放的Tag 我們之後會講解 message.getBytes(RemotingHelper.DEFAULT_CHARSET)); SendResult send = producer.send(msg); System.out.println(send); } public static void main(String[] args) { try { send("test","hello world!"); } catch (Exception e) { e.printStackTrace(); } } }
上面的程式碼片段就是生產者傳送訊息到RocketMQ裡去的程式碼,其實這種方式就是所謂的同步傳送訊息到MQ
那麼什麼叫同步傳送訊息到MQ裡去?
所謂同步,意思就是你通過這行程式碼傳送訊息到MQ去,SendResult sendResult = producer.send(msg),然後你會卡在這裡,程式碼不能往下走了
你要一直等待MQ返回一個結果給你,你拿到了結果之後,你的程式才會繼續向下執行。
這個就是所謂的同步傳送模式。
非同步傳送
接著我們來看一下非同步傳送的程式碼:
import com.alibaba.rocketmq.client.exception.MQClientException; import com.alibaba.rocketmq.client.producer.DefaultMQProducer; import com.alibaba.rocketmq.client.producer.SendCallback; import com.alibaba.rocketmq.client.producer.SendResult; import com.alibaba.rocketmq.common.message.Message; import com.alibaba.rocketmq.remoting.common.RemotingHelper; public class RocketMQProducer { // RocketMQ的生產者類 private static DefaultMQProducer producer; static { // 構建生產者物件,指定生產組 producer=new DefaultMQProducer("test_group"); // 設定NameServer的地址,拉取路由資訊 producer.setNamesrvAddr("192.168.220.110:9876"); try { // 啟動生產者 producer.start(); } catch (MQClientException e) { e.printStackTrace(); } // 設定非同步傳送失敗的時候不重試 producer.setRetryTimesWhenSendAsyncFailed(0); } public static void send(String topic,String message) throws Exception { // 構建訊息物件 Message msg=new Message(topic, "",//這裡存放的Tag 我們之後會講解 message.getBytes(RemotingHelper.DEFAULT_CHARSET)); producer.send(msg, new SendCallback() { public void onSuccess(SendResult sendResult) { System.out.println(sendResult); } public void onException(Throwable throwable) { System.out.println(throwable); } }); } public static void main(String[] args) { try { send("test","hello world!"); } catch (Exception e) { e.printStackTrace(); } } }
意思就是訊息傳送後,程式碼繼續向下執行,等到mq返回結果的時候,如果返回成功,就會呼叫回撥函式onSuccess方法,返回失敗就會呼叫onException方法。
這就是非同步傳送,它的特點就是不會阻塞程式,訊息返回結果後再呼叫回撥函式。
單向傳送
還有一種傳送方式,叫做單向傳送,那麼什麼是單向傳送呢?
程式碼如下:
producer.sendOneway(msg);
它的意思就是生產者傳送訊息給MQ,傳送後程序繼續向下執行,不會阻塞,而且不會再管MQ的返回值。
也就是說傳送過後就不關它的事了。
RocketMQ的傳送方式就介紹到這裡,關於具體的使用場景我們之後的文章再討論,現在只要清楚有這些方式就可以了。
RocketMQ的消費
Push消費
我們來看一下push消費的程式碼
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer; import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently; import com.alibaba.rocketmq.client.exception.MQClientException; import com.alibaba.rocketmq.common.message.MessageExt; import java.io.UnsupportedEncodingException; import java.util.List; import static com.alibaba.rocketmq.remoting.common.RemotingHelper.DEFAULT_CHARSET; public class RocketMQConsumer { public static void main(String[] args) throws MQClientException { // 建立push消費者例項,指定消費者組 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_group"); // 設定NameServer的地址,拉取路由資訊 consumer.setNamesrvAddr("192.168.220.110:9876"); // 訂閱test Topic , 第二個引數是Tag consumer.subscribe("test",null); // 註冊消費者監聽器,接收到訊息就會呼叫這個方法 consumer.registerMessageListener(new MessageListenerConcurrently() { public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { // 在這裡進行訊息的處理 for (MessageExt t : msgs) { try { System.out.println(new String(t.getBody(), DEFAULT_CHARSET)); } catch (UnsupportedEncodingException e) { e.printStackTrace(); } } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); // 啟動消費者例項 consumer.start(); System.out.println("----------Consumer Start-----------"); } }
大家注意裡面Consumer的類名:DefaultMQPushConsumer。
從類名中我們可以提取出來一個關鍵的資訊:Push。其實從這裡我們就能看出來,當前我們使用的訊息消費實際上是Push模式。
那麼什麼是Push消費模式呢?
簡單來講就是Broker會主動把訊息傳送給你的消費者,你的消費者是被動的接收Broker推送給過來的訊息,然後進行處理。
這個就是所謂的Push模式,意思就是Broker主動推送訊息給消費者。
Pull消費
接下來我們簡單說一下Pull消費的方式。
Pull消費其實理解起來也很容易,就是Broker不在主動推送訊息給消費者了,而是消費者主動傳送請求從Broker中拉取訊息。
至於什麼時候用Push模式,什麼時候用Pull模式,我們以後再聊這個話題。
總結
今天我們一起使用Docker快速部署了RocketMQ的環境,又為小夥伴們分享了部署環境時遇到的坑和解決辦法,希望小夥伴們不要再遇到這個問題的時候無從下手。
之後我們又一起寫了生產者和消費者的程式碼,聊了聊RocketMQ的幾種傳送模式和消費模式。相信小夥伴們應該會有所收穫。
那今天的分享就到這裡,希望小夥伴們繼續支援我,那麼下次的分享不見不散。
往期文章推薦:
中介軟體專輯:
演算法專輯: