RocketMQ的發送模式和消費模式

本文轉載自查看原文 2020-09-13 21:30 1348 MQ

前言

小伙伴們大家好啊，王子又來和大家一起閑談MQ技術了。

通過之前文章的學習，我們已經對RocketMQ的基本架構有了初步的了解，那今天王子就和大家一起來點實際的，用代碼和大家一起看看RocketMQ的幾種發送模式和消費模式。好了，讓我們開始吧。

RocketMQ的環境准備

本次我們的環境搭建是基於docker的，需要小伙伴們准備一台CentOS7的虛擬機(虛擬機的安裝這里就不做介紹了)。

王子使用的是安裝在VMware中的CentOS7虛擬機系統。正式安裝環境之前，為了測試方便，我們關閉了centos系統的防火牆。然后我們需要在這台虛擬機中安裝docke和docker-compose，具體的安裝步驟大家可以參考https://www.runoob.com/docker/centos-docker-install.html，一些常用的docker命令也可以在這個網站里查看，還是很方便的，王子在這里分享給大家。

安裝過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-----------");
    }
}