RocketMQ-基礎(chǔ)使用（二）

前置文章：
RocketMQ-基礎(chǔ)使用（一），該文主要涉及MQ基礎(chǔ)、RocketMQ安裝&集群搭建、RocketMQ監(jiān)控平臺。

官方基礎(chǔ)使用樣例，很多基礎(chǔ)內(nèi)容其實官方文檔都有很詳細(xì)的說明。日常使用如果時間充足，還是推薦查看官方文檔。學(xué)習(xí)官方文檔是一個良好的習(xí)慣。

零、本文綱要

一、RocketMQ-基礎(chǔ)使用

  1. 前置文章基礎(chǔ)指令

二、RocketMQ-發(fā)送消息

  1. 發(fā)送同步消息
  2. 發(fā)送異步消息
  3. 發(fā)送單向消息

三、RocketMQ-接收消息

  1. 消息接收
  2. 消息接收-負(fù)載均衡【默認(rèn)】
  3. 消息接收-廣播模式

四、RocketMQ-消息類型

  1. 順序消息
  2. 延遲消息
  3. 批量消息
  4. 過濾消息
  5. 事務(wù)消息

一、RocketMQ-基礎(chǔ)使用

0. 前置文章基礎(chǔ)指令

Ⅰ 啟動RocketMQ的基礎(chǔ)指令

# Start Name Server
nohup sh bin/mqnamesrv &
tail -f ~/logs/rocketmqlogs/namesrv.log

# Start Broker
nohup sh bin/mqbroker -n localhost:9876 &
tail -f ~/logs/rocketmqlogs/broker.log

指定自定義配置文件啟動nohup sh bin/mqbroker -n localhost:9876 -c conf/broker.conf &

Ⅱ 關(guān)閉RocketMQ的基礎(chǔ)指令

# Shutdown Servers
sh bin/mqshutdown broker
sh bin/mqshutdown namesrv

二、RocketMQ-發(fā)送消息

發(fā)送同步消息 / 發(fā)送異步消息 / 發(fā)送單向消息

1. 發(fā)送同步消息

這種可靠性同步地發(fā)送方式使用的比較廣泛，比如：重要的消息通知，短信通知。

  • ① 基礎(chǔ)依賴
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.5.0</version>
</dependency>
  • ② 同步消息代碼
/**
 * 發(fā)送同步消息
 */
public class SyncProducer {

    public static void main(String[] args) throws Exception {
        //1.創(chuàng)建消息生產(chǎn)者producer，并制定生產(chǎn)者組名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.指定Nameserver地址，多個NameServer則用“;”隔開
        producer.setNamesrvAddr("192.168.253.128:9876");
        //3.啟動producer
        producer.start();

        for (int i = 0; i < 10; i++) {
            //4.創(chuàng)建消息對象，指定主題Topic、Tag和消息體
            /**
             * 參數(shù)一：消息主題Topic
             * 參數(shù)二：消息Tag
             * 參數(shù)三：消息內(nèi)容
             */
            Message msg = new Message("base", "tag1", ("Hello RocketMQ [" + i + "]").getBytes());
            //5.發(fā)送消息
            SendResult result = producer.send(msg);
            //發(fā)送狀態(tài)
            SendStatus status = result.getSendStatus();
            String msgId = result.getMsgId();
            int queueId = result.getMessageQueue().getQueueId();
            System.out.printf("發(fā)送狀態(tài)：%s，消息ID：%s，隊列：%d%n", status, msgId, queueId);
            //線程睡1秒
            Thread.sleep(1000);
        }

        //6.關(guān)閉生產(chǎn)者producer
        producer.shutdown();
    }
}

截取控制臺輸出
發(fā)送狀態(tài)：SEND_OK，消息ID：C0A8026AC05818B4AAC28D428B270000，隊列：3

2. 發(fā)送異步消息

異步消息通常用在對響應(yīng)時間敏感的業(yè)務(wù)場景，即發(fā)送端不能容忍長時間地等待Broker的響應(yīng)。

/**
 * 發(fā)送異步消息
 */
public class AsyncProducer {

    public static void main(String[] args) throws Exception {
        //1.創(chuàng)建消息生產(chǎn)者producer，并制定生產(chǎn)者組名
        DefaultMQProducer producer = new DefaultMQProducer("group2");
        //2.指定Nameserver地址，多個NameServer則用“;”隔開
        producer.setNamesrvAddr("192.168.253.128:9876");
        //3.啟動producer
        producer.start();

        for (int i = 0; i < 10; i++) {
            //4.創(chuàng)建消息對象，指定主題Topic、Tag和消息體
            /**
             * 參數(shù)一：消息主題Topic
             * 參數(shù)二：消息Tag
             * 參數(shù)三：消息內(nèi)容
             */
            Message msg = new Message("base", "tag2", ("AsyncMsg [" + i + "]").getBytes());
            //5.發(fā)送異步消息
            producer.send(msg, new SendCallback() {
                /**
                 * 發(fā)送成功的回調(diào)函數(shù)
                 * @param sendResult 發(fā)送結(jié)果
                 */
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println("發(fā)送結(jié)果：" + sendResult);
                }

                /**
                 * 發(fā)送失敗的回調(diào)函數(shù)
                 * @param throwable 發(fā)送異常
                 */
                @Override
                public void onException(Throwable throwable) {
                    System.out.println("發(fā)送異常：" + throwable);
                }
            });
            //線程睡1秒
            Thread.sleep(1000);
        }

        //6.關(guān)閉生產(chǎn)者producer
        producer.shutdown();
    }
}

與同步消息的不同之處在于通過回調(diào)函數(shù)來獲取發(fā)送結(jié)果。

3. 發(fā)送單向消息

這種方式主要用在不特別關(guān)心發(fā)送結(jié)果的場景，比如：日志發(fā)送。

/**
 * 發(fā)送單向消息
 */
public class OneWayProducer {

    public static void main(String[] args) throws Exception, MQBrokerException {
        //1.創(chuàng)建消息生產(chǎn)者producer，并制定生產(chǎn)者組名
        DefaultMQProducer producer = new DefaultMQProducer("group3");
        //2.指定Nameserver地址，多個NameServer則用“;”隔開
        producer.setNamesrvAddr("192.168.253.128:9876");
        //3.啟動producer
        producer.start();

        for (int i = 0; i < 10; i++) {
            //4.創(chuàng)建消息對象，指定主題Topic、Tag和消息體
            /**
             * 參數(shù)一：消息主題Topic
             * 參數(shù)二：消息Tag
             * 參數(shù)三：消息內(nèi)容
             */
            Message msg = new Message("base", "tag3", ("OneWayMsg [" + i + "]").getBytes());
            //5.發(fā)送單向消息
            producer.send(msg);
            //線程睡1秒
            Thread.sleep(1000);
        }

        //6.關(guān)閉生產(chǎn)者producer
        producer.shutdown();
    }
}

三、RocketMQ-接收消息

1. 消息接收

/**
 * 消息的接受者
 */
public class Consumer {

    public static void main(String[] args) throws Exception {
        //1.創(chuàng)建消費者Consumer，制定消費者組名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group3");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.253.128:9876");
        //3.訂閱主題Topic和Tag
        consumer.subscribe("base", "tag1");
        //設(shè)定消費模式：負(fù)載均衡|廣播模式

        //4.設(shè)置回調(diào)函數(shù)，處理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //接收消息內(nèi)容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                                                            ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.println(list);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //5.啟動消費者consumer
        consumer.start();
    }
}

2. 消息接收-負(fù)載均衡【默認(rèn)】

消費者采用負(fù)載均衡方式消費消息，多個消費者共同消費隊列消息，每個消費者處理的消息不同?！灸J(rèn)的消息消費方式】

consumer.setMessageModel(MessageModel.CLUSTERING);

3. 消息接收-廣播模式

消費者采用廣播的方式消費消息，每個消費者消費的消息都是相同的。

consumer.setMessageModel(MessageModel.BROADCASTING);

四、RocketMQ-消息類型

1. 順序消息

  • ① 基礎(chǔ)分析

假定一個訂單的順序流程是：創(chuàng)建、付款、推送、完成。有張三、李四兩人進(jìn)行訂單業(yè)務(wù)。

a、全局有序：
張三所有消息消費完，再消費李四消息，且內(nèi)部有序；
一個Borker，一個MessageQueue；

b、局部有序：
只要保證各自消息內(nèi)部的有序消費，交替消費兩者的消息是可以的；
一個Borker，多個MessageQueue，一個MessageQueue對應(yīng)一個訂單。

所以，一般僅需保證局部有序即可。
實現(xiàn)方式：同一個用戶的一個業(yè)務(wù)消息放到同一個隊列，比如：訂單號相同的消息進(jìn)同一個隊列。

  • ② 代碼實現(xiàn)

Ⅰ 消息生產(chǎn)者producer的核心代碼

/**
 * 參數(shù)一: 消息對象
 * 參數(shù)二: 消息隊列選擇器 MessageQueueSelector
 * 參數(shù)三: 選擇隊列業(yè)務(wù)標(biāo)識，此處為訂單ID
 */
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
    /**
     *
     * @param list      消息隊列
     * @param message   消息對象
     * @param o         業(yè)務(wù)標(biāo)識的參數(shù)
     * @return 消息隊列
     */
    @Override
    public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
        long orderId = (long) o;
        long index = orderId % list.size(); //訂單ID一致，則取模結(jié)果一致，最終選擇的隊列一致
        return list.get((int) index);
    }
}, order.getOrderId());

Ⅱ 消息消費者consumer的核心代碼

此處是通過有序消息監(jiān)聽MessageListenerOrderly來實現(xiàn)的

//4.注冊消息監(jiān)聽器
consumer.registerMessageListener(new MessageListenerOrderly() {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
        for (MessageExt messageExt : list) {
            System.out.println("線程名稱：" + Thread.currentThread().getName() + " → " +
                    "消費消息：" + new String(messageExt.getBody()));
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
});

2. 延遲消息

使用場景：比如電商里，提交了一個訂單就可以發(fā)送一個延時消息，1h后去檢查這個訂單的狀態(tài)，如果還是未付款就取消訂單釋放庫存。

延遲消息使用限制：

// org/apache/rocketmq/store/config/MessageStoreConfig.java
private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

Ⅰ 消息生產(chǎn)者producer的核心代碼

msg.setDelayTimeLevel(2);

Ⅱ 消息消費者consumer的核心代碼

無需調(diào)整。

注意：受限于網(wǎng)絡(luò)情況，實際的延遲往往大于設(shè)置的延遲。

3. 批量消息

批量發(fā)送消息能顯著提高傳遞小消息的性能。

限制：
a、相同的topic；
b、相同的waitStoreMsgOK；
c、不能是延時消息；
d、總大小不應(yīng)超過4MB。

Ⅰ 消息生產(chǎn)者producer的核心代碼

List<Message> messageList = new ArrayList<>();

Ⅱ 消息消費者consumer的核心代碼

無需調(diào)整。

4. 過濾消息

一般過濾消息可通過 TAG / SQL92標(biāo)準(zhǔn) 來進(jìn)行過濾

Ⅰ 消息生產(chǎn)者producer的核心代碼

//方式一：通過Tag過濾的使用方法，消息發(fā)送方不做調(diào)整
//...

//方式二：通過sql過濾的使用方法，使用putUserProperty設(shè)置一些消息屬性
msg.putUserProperty("a", String.valueOf(i));

Ⅱ 消息消費者consumer的核心代碼

//方式一：通過Tag過濾的使用方法，consumer使用" || "分隔訂閱不同的Tag即可
consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");

//方式二：通過sql過濾的使用方法，通過MessageSelector消息選擇器的bySql方法過濾消息
consumer.subscribe("topic_sql_filter", MessageSelector.bySql("num > 5"));

方式二如果報錯：The broker does not support consumer to filter message by SQL92，
則需要在我們對應(yīng)的Broker配置文件內(nèi)做調(diào)整，添加enablePropertyFilter=true，重啟服務(wù)即可生效。

關(guān)于SQL92基礎(chǔ)語法，RocketMQ只定義了一些基本語法來支持這個特性：
數(shù)值比較，比如：>，>=，<，<=，BETWEEN，=；
字符比較，比如：=，<>，IN；
IS NULL 或者 IS NOT NULL；
邏輯符號 AND，OR，NOT；

常量支持類型為：
數(shù)值，比如：123，3.1415；
字符，比如：'abc'，必須用單引號包裹起來；
NULL，特殊的常量
布爾值，TRUE 或 FALSE

注意：只有使用push模式的消費者才能用使用SQL92標(biāo)準(zhǔn)的sql語句。

5. 事務(wù)消息

在【分布式事務(wù)-可靠消息最終一致性】的解決方案內(nèi)使用的就是事務(wù)消息。

事務(wù)消息流程：正常事務(wù)消息的發(fā)送及提交，以及事務(wù)消息的補償【事務(wù)狀態(tài)回查】；

事務(wù)狀態(tài)：
LocalTransactionState.COMMIT_MESSAGE 提交狀態(tài) 允許消費消息；
LocalTransactionState.ROLLBACK_MESSAGE 回滾狀態(tài) 刪除消息，不允許被消費；
LocalTransactionState.UNKNOW 中間狀態(tài) 需要回查事務(wù)。

Ⅰ 消息生產(chǎn)者producer的核心代碼

/**
 * 發(fā)送同步消息
 */
public class Producer {

    public static void main(String[] args) throws Exception {
        //1.創(chuàng)建消息生產(chǎn)者producer，并制定生產(chǎn)者組名
        TransactionMQProducer producer = new TransactionMQProducer("group1");
        //2.指定Nameserver地址，多個NameServer則用“;”隔開
        producer.setNamesrvAddr("192.168.253.128:9876");
        //3.設(shè)置消息事務(wù)的監(jiān)聽器
        producer.setTransactionListener(new TransactionListener() {
            /**
             * 在該方法中執(zhí)行本地的事務(wù)
             * @param message 消息
             * @param o
             * @return 事務(wù)狀態(tài)
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                String messageTags = message.getTags();
                if (StringUtils.equals("TagA", messageTags)) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else if (StringUtils.equals("TagB", messageTags)) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                } else {
                    return LocalTransactionState.UNKNOW;
                }
            }

            /**
             * 該方法進(jìn)行MQ事務(wù)狀態(tài)的回查
             * @param messageExt 消息
             * @return 事務(wù)狀態(tài)
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println("消息的Tag：" + messageExt.getTags());
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        String[] tags = new String[]{"TagA", "TagB", "TagC"};

        //4.啟動producer
        producer.start();
        for (int i = 0; i < 3; i++) {
            //5.創(chuàng)建消息對象，指定Topic、Tag、消息體
            Message msg = new Message("topic_transaction", tags[i],
                    (tags[i] + " Hello transactionMsg " + i).getBytes(StandardCharsets.UTF_8));
            //6.發(fā)送消息
            SendResult sendResult = producer.sendMessageInTransaction(msg, null);
            System.out.println("發(fā)送結(jié)果：" + sendResult);
            Thread.sleep(1000);
        }
        //7.關(guān)閉生產(chǎn)者producer，此處需要回查，所以不關(guān)閉
        //producer.shutdown();
    }
}

Ⅱ 消息消費者consumer的核心代碼

/**
 * 消息的接受者
 */
public class Consumer {

    public static void main(String[] args) throws Exception {
        //1.創(chuàng)建消費者Consumer，制定消費者組名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.253.128:9876");
        //3.訂閱主題Topic和Tag
        consumer.subscribe("topic_transaction", "*");
        //4.注冊消息監(jiān)聽器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                                                            ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt messageExt : list) {
                    System.out.println("消費消息：" + new String(messageExt.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.啟動消費者consumer
        consumer.start();
        System.out.println("消費者啟動了...");
    }
}

注意：事務(wù)消息不支持延時消息和批量消息。

五、結(jié)尾

以上即為RocketMQ-基礎(chǔ)使用（二）的全部內(nèi)容，感謝閱讀。

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容