前言
小伙伴們大家好啊，小阿Giao來和大家一起閑談MQ技術了，
通過之前文章的學習，我們已經對RocketMQ的基本架構有了初步的了解，那今天小阿Giao就和大家一起來點實際的，用代碼和大家一起看看RocketMQ的幾種發送模式和消費模式，好了，讓我們開始吧，
RocketMQ的環境準備
本次我們的環境搭建是基于docker的，需要小伙伴們準備一臺CentOS7的虛擬機(虛擬機的安裝這里就不做介紹了)，
小阿Giao使用的是安裝在VMware中的CentOS7虛擬機系統，正式安裝環境之前，為了測驗方便，我們關閉了centos系統的防火墻，然后我們需要在這臺虛擬機中安裝docke和docker-compose，具體的安裝步驟大家可以參考https://www.runoob.com/docker/centos-docker-install.html，一些常用的docker命令也可以在這個網站里查看，還是很方便的，小阿Giao在這里分享給大家，
安裝過docker和docker-compose后，我們就要開始安裝RocketMQ的環境了，具體的安裝步驟可以參考https://gitee.com/lm970585581/docker-rocketmq中的一鍵部署docker-compose部分，
步驟都很簡單，基本上算是一鍵式傻瓜安裝了，但是，小阿Giao在安裝的程序中卻被一個問題困擾住了很久，
下邊給大家說一下遇到的問題和解決的辦法，這是重點哦，
小阿Giao按照步驟一步一步的執行，到最后一步執行start.sh的時候悲劇發生了，
如果一切正常，執行了這一步驟后，docker中一共會啟動三個容器，分別是nameserver、broker、和rocketMQ的管控臺，但是小阿Giao執行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的幾種發送模式和消費模式，相信小伙伴們應該會有所識訓，
那今天的分享就到這里，希望小伙伴們繼續支持我，那么下次的分享不見不散，
轉載請註明出處，本文鏈接：https://www.uj5u.com/yidong/128617.html
標籤：其他