RocketMQ 4.7.1 環境搭建、集羣、SpringBoot整合MQ

時間 2020-10-31

標籤 html java linux git github web 面試 redis 算法 spring 欄目負載均衡简体版

原文原文鏈接

導讀

　　以前學過ActiveMQ可是併發量不是很大點我直達，因此又學阿里開源的RocketMQ，聽說隊列能夠堆積億級別。下面是網上找的消息隊列對比圖，僅供參考html

部署

官網

點我直達java

前置條件

推薦使用64位操做系統，建議使用Linux / Unix / Mac；
64位JDK 1.8+;
Maven 3.2.x;
Git;
適用於Broker服務器的內存4G +可用磁盤

下載

地址：https://downloads.apache.org/rocketmq/4.7.1/rocketmq-all-4.7.1-source-release.zip linux

百度雲盤：git

連接: https://pan.baidu.com/s/1luq_MwxSn8k_bugrnQSJWg  密碼: varj

安裝依賴項

jdk：點我直達
maven：點我直達
git安裝：yum install -y git

export JAVA_HOME=/opt/soft/jdk1.8.0_202
export PATH=$JAVA_HOME/bin:$PATH
export CLASPATH=.:$JAVA_home/lib/dt.jar:$JAVA_HOME/lib/tools.jar
export JAVA_HOME PATH CLASSPATH
export MAVEN_HOME=/opt/soft/apache-maven-3.6.3
export PATH=$PATH:$MAVEN_HOME/bin

mq上傳至linux

解壓

maven編譯

啓動NameServer

後臺啓動方式

nohup sh bin/mqnamesrv &

NameServer啓動時內存不足(問題解決)

找到runserver.sh 修改JAVA_OPT

vim /bin/runserver.sh配置

啓動Broker

nohup sh bin/mqbroker -n localhost:9876 &

語法：nohup sh bin/mqbroker -n NameServer服務ip地址

Broker內存不足（問題解決）

找到runbroker.sh 修改JAVA_OPT

vim /bin/runbroker.sh配置

服務都啓動成功

模擬消費

export NAMESRV_ADDR=localhost:9876

sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer

sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer

開2個控制檯，鏈接通一臺linuxgithub

注意

　　NameServer默認端口號：9876；broker默認端口號：10911web

可視化控制檯

官網地址

點我直達面試

百度雲盤redis

連接: https://pan.baidu.com/s/1mdEGkq-JBTy1wtNmFPkmDg  密碼: v6bq

解壓

安裝編譯

進入：/opt/soft/rocketmq-externals-master/rocketmq-console
編譯： mvn clean package -Dmaven.test.skip=true

修改appliccation.properties的rocketmq.config.namesrvAddr算法

編譯打包spring

啓動

　　進入target目錄，啓動java -jar

守護進程啓動： nohup java -jar rocketmq-console-ng-2.0.0.jar &

SpringBoot整合RocketMQ(生產者)

建立SpringBoot項目

點我直達

項目結構

加入依賴

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.1.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.ybchen</groupId>
    <artifactId>ybchen-mq</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>ybchen-mq</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <!--注意： 這裏的版本,要和部署在服務器上的版本號一致-->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.7.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-common</artifactId>
            <version>4.7.1</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

PayProducer.java

package com.ybchen.ybchenmq.jms;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.springframework.stereotype.Component;

/**
 * 消息生產者
 */
@Component
public class PayProducer {
    /**
     * 生產者所屬的組
     */
    private String producerGroup = "pay_group";
    /**
     * MQ的地址，注意需開放端口號或者關閉防火牆
     */
    private String nameServerAddr = "192.168.199.100:9876";
    private DefaultMQProducer producer;

    public PayProducer() {
        producer = new DefaultMQProducer(producerGroup);
        //指定NameServer地址，多個地址以;隔開
        //如 producer.setNamesrvAddr("192.168.199.100:9876;192.168.199.101:9876;192.168.199.102:9876")
        producer.setNamesrvAddr(nameServerAddr);
        start();
    }

    /**
     * 獲取生產者
     * @return
     */
    public DefaultMQProducer getProducer() {
        return this.producer;
    }

    /**
     * 開啓，對象在使用以前必需要調用一次，只能初始化一次
     */
    public void start() {
        try {
            this.producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    /**
     * 關閉，通常在應用上下文，使用上下文監聽器，進行關閉
     */
    public void shutdown() {
        this.producer.shutdown();
    }
}

PayController.java

package com.ybchen.ybchenmq.controller;

import com.ybchen.ybchenmq.jms.PayProducer;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @ClassName：PayController
 * @Description：支付
 * @Author：chenyb
 * @Date：2020/10/18 2:47 下午
 * @Versiion：1.0
 */
@RestController
@RequestMapping("/api/v1")
public class PayController {
    @Autowired
    private PayProducer payProducer;

    private static final String TOPIC = "ybchen_pay_topic";

    /**
     * 支付回調
     *
     * @param text
     * @return
     */
    @RequestMapping("pay_cb")
    public Object callback(String text) {
        /**
         * String topic：話題
         * String tags：二級分類
         * byte[] body：body消息字節數組
         */
        Message message = new Message(TOPIC,"tag_a",("hello ybchen ==>"+text).getBytes());
        try {
            SendResult send = payProducer.getProducer().send(message);
            System.out.println("send------>"+send);
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "ok";
    }
}

測試

常見錯誤

錯誤一

org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException:
sendDefaultImpl call timeout

緣由：阿里雲存在多網卡，rocketmq會根據當前網卡選擇一個IP使用，當你的機器有多塊網卡時，極可能會有問題，好比，機器上有兩個ip，一個公網ip，一個私網ip，所以須要配置broker.conf指定當前公網的ip，而後重啓broker


修改配置：/opt/soft/rocketmq-all-4.7.1-source-release/distribution/target/rocketmq-4.7.1/rocketmq-4.7.1/conf/broker.conf
新增這個配置：brokerIP1=xxx.xxx.xxx.xxx

啓動命令：nohup sh bin/mqbroker -n localhost:9876 -c ./conf/broker.conf &

錯誤2

MQClientException: No route info of this topic, TopicTest1

緣由：Broker 緊追自動建立Topic，且用戶沒有經過手工方式建立此Topic，或者broker和Nameserver網絡不通

解決：
    經過sh bin/mqbroker -m 查看配置
    autoCreateTopicEnable=true 則自動建立Topic

Centos 7 關閉防火牆：systemctl stop firewalld

錯誤3

控制檯查看不了數據，提示鏈接10909錯誤

緣由：Rocket默認開啓了VIP通道，VPI通道端口號爲10911-2=10909

解決：阿里雲安全組添加一個端口：10909

錯誤4

　　沒法自動建立topic：客戶端版本要和服務端版本保持一致

服務器上裝的是4.7.1

引入依賴項時
        <!--注意： 這裏的版本,要和部署在服務器上的版本號一致-->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.7.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-common</artifactId>
            <version>4.7.1</version>
        </dependency>

檢索消息發送

SpringBoot整合RocketMQ(消費者)

建立SpringBoot項目

項目結構

加入依賴

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.1.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.ybchen</groupId>
    <artifactId>ybchen-mq</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>ybchen-mq</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <!--注意： 這裏的版本,要和部署在服務器上的版本號一致-->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.7.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-common</artifactId>
            <version>4.7.1</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

PayConsumer.java

package com.ybchen.ybchenmqconsumer.jms;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.List;

/**
 * @ClassName：PayConsumer
 * @Description：消費者
 * @Author：chenyb
 * @Date：2020/10/18 4:13 下午
 * @Versiion：1.0
 */
@Component
public class PayConsumer {
    /**
     * 生產者所屬的組
     */
    private String producerGroup = "pay_consumer_group";
    /**
     * MQ的地址，注意需開放端口號或者關閉防火牆
     */
    private String nameServerAddr = "192.168.199.100:9876";
    /**
     * 訂閱主題
     */
    private String topic = "ybchen_pay_topic";
    private DefaultMQPushConsumer consumer;

    public PayConsumer() throws MQClientException {
        consumer = new DefaultMQPushConsumer(producerGroup);
        //指定NameServer地址，多個地址以;隔開
        //如 producer.setNamesrvAddr("192.168.199.100:9876;192.168.199.101:9876;192.168.199.102:9876")
        consumer.setNamesrvAddr(nameServerAddr);
        //設置消費地點，從最後一個開始消費
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        //訂閱主題，監聽主題下的那些標籤
        consumer.subscribe(topic, "*");
        //註解一個監聽器
        //lambda方式
//        consumer.registerMessageListener((MessageListenerConcurrently) (msg, context) -> {
//            try {
//                Message message = msg.get(0);
//                System.out.printf("%s Receive New Messages: %s %n",
//                        Thread.currentThread().getName(), new String(msg.get(0).getBody()));
//                //主題
//                String topic = message.getTopic();
//                //消息內容
//                String body = null;
//                body = new String(message.getBody(), "utf-8");
//                //二級分類
//                String tags = message.getTags();
//                //鍵
//                String keys = message.getKeys();
//                System.out.println("topic=" + topic + ", tags=" + tags + ", keys=" + keys + ", msg=" + body);
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//            } catch (UnsupportedEncodingException e) {
//                e.printStackTrace();
//                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
//            }
//        });

        //通常方式
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                try {
                    Message message = list.get(0);
                    System.out.printf("%s Receive New Messages: %s %n",
                            Thread.currentThread().getName(), new String(list.get(0).getBody(),"utf-8"));
                    //主題
                    String topic = message.getTopic();
                    //消息內容
                    String body = null;
                    body = new String(message.getBody(), "utf-8");
                    //二級分類
                    String tags = message.getTags();
                    //鍵
                    String keys = message.getKeys();
                    System.out.println("topic=" + topic + ", tags=" + tags + ", keys=" + keys + ", msg=" + body);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        });
        consumer.start();
        System.out.println("consumer start ..........");
    }
}

application.properties

server.port=8081

測試生產者消費者

MQ集羣架構模式分析

單節點

優勢

　　本地開發測試，配置簡單，同步刷盤消息一條都不會丟

缺點

　　不可靠，若是宕機，會致使服務不可用

主從(異步、同步雙寫)

優勢

　　同步雙寫消息不丟失，異步複製存在少許丟失你，主節點宕機，從節點能夠對外提供消息的消費，可是不支持寫入

缺點

　　主備有短暫消息延遲，毫秒級，目前不支持自動切換，須要腳本或者其餘程序進行檢測而後中止broker，重啓讓從節點成爲主節點

雙主

優勢

　　配置簡單，能夠靠配置RAID磁盤陣列保證消息可靠，異步刷盤丟失少許消息

缺點

　　master宕機期間，未被消費的消息在機器恢復以前不可消息，實時性會受到影響

雙主雙從，多主多從模式(異步複製)

優勢

　　磁盤損壞，消息丟失的很是小，消息實時性不會受影響，Master宕機後，消費者仍然能夠從Slave消費

缺點

　　主備有短暫消息延遲，毫秒級，若是Master宕機，磁盤損壞狀況，會丟失你少許消息

雙主雙從，多主多從模式(同步雙寫)

優勢

　　同步雙寫方式，主備都寫成功，才嚮應用返回成功，服務可用性與數據可用性很是高

缺點

　　性能比異步複製模式略低，主宕機後，備機不能自動切換爲主機

主從集羣搭建

準備工做

　　準備2臺機器，ip地址分別爲：192.168.199.100；192.168.199.101；

　　環境：RocketMQ4.7.1+jdk8+Maven+Centos 7

啓動兩臺nameserver

　　啓動兩個機器的nameserver

路徑：/opt/soft/rocketmq-all-4.7.1-source-release/distribution/target/rocketmq-4.7.1/rocketmq-4.7.1

啓動：nohup sh bin/mqnamesrc &

編輯並啓動roccketmq

主節點

進入：/opt/soft/rocketmq-all-4.7.1-source-release/distribution/target/rocketmq-4.7.1/rocketmq-4.7.1/conf/2m-2s-async


編輯並修改以下：vim broker-a.properties 
namesrvAddr=192.168.199.100:9876;192.168.199.101:9876
brokerClusterName=YbChenCluster
brokerName=broker-a
brokerId=0
deleteWhen=04
fileReservedTime=48
brokerRole=ASYNC_MASTER
flushDiskType=ASYNC_FLUSH


啓動：nohup sh bin/mqbroker -c conf/2m-2s-async/broker/broker-a.properties &

從節點

進入：/opt/soft/rocketmq-all-4.7.1-source-release/distribution/target/rocketmq-4.7.1/rocketmq-4.7.1/conf/2m-2s-async


編輯並修改以下：vim broker-a-s.properties 
namesrvAddr=192.168.199.100:9876;192.168.199.101:9876
brokerClusterName=YbChenCluster
brokerName=broker-a
brokerId=1
deleteWhen=04
fileReservedTime=48
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH


啓動：nohup sh bin/mqbroker -c conf/2m-2s-async/broker/broker-a-s.properties &

注意事項

namesrvAddr：相同
brokerClusterName：相同
brokerName：相同
brokerId：不一樣，0是主節點
deleteWhen：相同
fileReservedTime：相同
brokerRole：不一樣，分ASYNC_MASTER、SLAVE
flushDiskType：相同

啓動broker

使用管控臺

　　使用192.168.199.100這臺服務器，修改配置

192.168.199.100這臺服務器

進入：/opt/soft/rocketmq-externals-master/rocketmq-console/src/main/resources


修改配置文件：vim application.properties

rocketmq.config.namesrvAddr=192.168.199.100:9876;192.168.199.101:9876


編譯

切換到：/opt/soft/rocketmq-externals-master/rocketmq-console
打包：
mvn clean
mvn install -Dmaven.test.skip=true


啓動

進入：/opt/soft/rocketmq-externals-master/rocketmq-console/target
守護進程方式啓動：nohup java -jar rocketmq-console-ng-2.0.0.jar &

集羣測試

故障演練

　　模擬主掛了，可是從還能夠被消費，此時不能寫入，等主重啓後，能夠繼續寫入(數據不會被重複消費)，如下內容是連續的

總結

　　好了，到目前爲止，主從已經搭建完成了。

　　Broker分爲Master和Slave，一個Master能夠對應多個Slave，但一個Slave只能對應一個Master，Master與Slave經過相同的Broker Name來匹配，不一樣的Broker id來定義時Master仍是Slave

　　　　Broker向全部的NameServer節點創建長鏈接，定時註冊Topic和發送元數據信息

　　　　NameServer定時掃描(默認2分鐘)全部存活Broker的鏈接，若是超過期間沒響應，則斷開鏈接(心跳檢測)，可是Consumer客戶端不能感知，Consumer定時(30秒)從NameServer獲取topic的最新信息，因此broker不可用時，Consumer最多須要30秒才能發現

　　只有Master才能進行寫入操做，Slave不容許寫入只能同步，同步策略取決於Master配置

　　客戶端消費能夠從Master和Slave消費，默認消費者都從Master消費，若是在Master掛了以後，客戶端從NameServer中感知Broker宕機，就會從Slave消費，感知非實時，存在必定的滯後性，Slave不能保證Master的100%都同步過來，會有少許的消息丟失。一旦Master恢復，未同步過去的消息會被最終消費掉。

　　若是Consumer實例的數量比Message Queue的總數量還多的話，多出來的Consumer實例將沒法分到Queue，也就沒法消費到消息，也就沒法起到分攤負載的做用，因此須要控制讓Queue的總數量大於Consumer的數量。

場景模擬

生產和消費重試及處理

生產者重試

消息重試(保證數據的高可靠性)，自己內部支持重試，默認次數是2
若是網絡狀況較差，或者跨集羣則建議多改幾回

生產者設置重試次數，並設置惟一的key(通常惟一標識符)

消費者重試

緣由：消息處理異常，broker端到consumer端各類問題，如網絡緣由閃斷，消費處理失敗，ACK返回失敗等
注意
- 重試間隔時間配置，默認每條消息最多重試16次
- 超太重試次數人工補償
- 消費端去重
- 一條消息不管重試多少次，這些重試消息的Message ID，key不會改變
- 消費重試只針對集羣消費方式生效；廣播方式不提供失敗重試特性，即消費失敗後，失敗消息再也不重試，繼續消費新的消息

設置廣播方式

模擬消息重發

異步發送消息和回調實戰

應用場景

　　好比12306付完錢💰後，異步出票，對性能要求高，能夠支持更高的併發，回調成功後觸發相應的業務(onSuccess)

官方例子

點我直達

改造生產者

演示

onSuccess：由於是異步方式，這裏能夠記錄日誌啥的
onException：補償機制，根據實際狀況使用，看是否進行重試

OneWay(無需等待)

應用場景

　　主要作日誌收集，適用於對性能要求高，但可靠性並不高的場景。

延遲消息實戰

什麼是延遲消息

Producer將消息發送到消息隊列RocketMQ服務端，但並不指望這條消息立馬投遞，而是推遲在當前時間點以後的某一個時間投遞到Consumer進行消費，該消息即定時消息，目前支持固定精度的消息
延遲消息級別，1....18

1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

應用場景

經過消息觸發一些定時任務，好比在某一固定時間點向用戶發送提醒消息
消息生產和消費有時間窗口要求：好比在天貓電商交易中超時未支付關閉訂單的場景，在訂單建立時會發送一條延遲消息。這條消息將會在30分鐘之後投遞給消費者，消費者收到此消息後須要判斷對應的訂單是否已完成支付。如支付未完成，則關閉訂單。如已完成支付則忽略。

改生產者

生產者MessageQueueSelector實戰

簡介

　　生產消息使用MessageQueueSelector投遞到Topic下指定的Queue

應用場景

順序消息
分攤負載

默認topic下的queue數量是4，能夠配置

支持同步，異步發送指定的MessageQueue

選擇的queue數量必須小於配置的，不然會出錯

好處

　　若是隊列中某個產品，流量暴增，隨機分配的話，會致使整個Topic都不能使用，指定到隊列的話，若是這個隊列壞了，其餘隊列不影響使用。

改造生產者

同步發送

發送結果=SEND_OK，msg=SendResult [sendStatus=SEND_OK, msgId=AC1068013E3F18B4AAC276723EAC0000, offsetMsgId=C0A8C76400002A9F000000000009B536, messageQueue=MessageQueue [topic=ybchen_pay_topic, brokerName=broker-a, queueId=0], queueOffset=1]
發送結果=SEND_OK，msg=SendResult [sendStatus=SEND_OK, msgId=AC1068013E3F18B4AAC27672BCD50001, offsetMsgId=C0A8C76400002A9F000000000009B602, messageQueue=MessageQueue [topic=ybchen_pay_topic, brokerName=broker-a, queueId=0], queueOffset=2]
發送結果=SEND_OK，msg=SendResult [sendStatus=SEND_OK, msgId=AC1068013E3F18B4AAC27672CAA20002, offsetMsgId=C0A8C76400002A9F000000000009B6CF, messageQueue=MessageQueue [topic=ybchen_pay_topic, brokerName=broker-a, queueId=0], queueOffset=3]

　　能夠看到打印出來的，queueId=0

異步發送

生產者端代碼修改

    @Autowired
    private PayProducer payProducer;

    private static final String TOPIC = "ybchen_pay_topic";

    /**
     * 支付回調
     *
     * @param text
     * @return
     */
    @RequestMapping("pay_cb")
    public Object callback(String text) {
        /**
         * String topic：話題
         * String tags：二級分類
         * byte[] body：body消息字節數組
         */
        Message message = new Message(TOPIC, "tag_a", text.getBytes());
        //生產者使用MessageQueueSelector投遞到Topic下指定的Queue，arg只能小於等於4
//        try {
//            SendResult sendResult = payProducer.getProducer().send(message, new MessageQueueSelector() {
//                @Override
//                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
//                    int queueNum=Integer.parseInt(o.toString());
//                    return list.get(queueNum);
//                }
//            }, 0);
//            System.out.printf("發送結果=%s，msg=%s",sendResult.getSendStatus(),sendResult);
//        } catch (MQClientException e) {
//            e.printStackTrace();
//        } catch (RemotingException e) {
//            e.printStackTrace();
//        } catch (MQBrokerException e) {
//            e.printStackTrace();
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
        //異步發送到指定的queue
        try {
            payProducer.getProducer().send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    int queueNum = Integer.parseInt(o.toString());
                    return list.get(queueNum);
                }
            }, 3, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.printf("發送結果=%s，msg=%s", sendResult.getSendStatus(), sendResult);
                }

                @Override
                public void onException(Throwable e) {
                    e.printStackTrace();
                }
            });
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
        //message.setDelayTimeLevel(2);
//        try {
//            SendResult send = payProducer.getProducer().send(message);
//            System.out.println("send------>"+send);
//        } catch (MQClientException e) {
//            e.printStackTrace();
//        } catch (RemotingException e) {
//            e.printStackTrace();
//        } catch (MQBrokerException e) {
//            e.printStackTrace();
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
        //異步發送
//        try {
//            payProducer.getProducer().send(message, new SendCallback() {
//                @Override
//                public void onSuccess(SendResult sendResult) {
//                    System.out.printf("發送結果=%s，msg=%s",sendResult.getSendStatus(),sendResult);
//                }
//
//                @Override
//                public void onException(Throwable e) {
//                    e.printStackTrace();
//                    //補償機制，根據實際狀況使用，看是否進行重試
//                }
//            });
//        } catch (MQClientException e) {
//            e.printStackTrace();
//        } catch (RemotingException e) {
//            e.printStackTrace();
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
        return "ok";
    }

順序消息的應用場景

簡介

　　順序消息能夠應用到電商和證券系統，訂單系統。

什麼是順序系統？

　　消息的生產和消費順序一致

全局順序

　　topic下面所有消息都要有序(不多用)

性能要求不高，全部的消息嚴格按照FIFO(先進先出)原則進行消息發佈和消費的場景，並行度成爲消息系統的瓶頸，吞吐量不夠
在證券處理中，以人民幣兌換美圓爲例，在價格相同的狀況下，先出價者優先處理，則能夠經過全局順序的方式進行發佈和消費

局部順序

　　只要保證一組消息被順序消費便可(RocketMQ中使用)

性能要求高
電商的訂單建立，同一訂單相關的建立訂單消息、訂單支付消息、訂單退款消息、訂單物流消息、訂單交易成功消息都會按照前後順序來發布和消費

順序發佈

　　對於指定的一個Topic，客戶端按照必定的前後順序發送消息

順序消費

　　對於指定的一個Topic，按照必定的前後順序接收消息，即先發送的消息必定先會被客戶端接收到

注意事項

順序消息不支持異步發送，不然將沒法保證順序消費
順序消息暫不支持廣播模式

官方例子

點我直達

改造生產者代碼

建立ProductOrder.java

package com.ybchen.ybchenmq.entity;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

/**
 * @ClassName：ProductOrder
 * @Description：訂單
 * @Author：chenyb
 * @Date：2020/10/25 12:56 下午
 * @Versiion：1.0
 */
public class ProductOrder implements Serializable {
    /**
     * 訂單id
     */
    private long orderIdl;
    /**
     * 訂單操做類型
     */
    private String type;

    public long getOrderIdl() {
        return orderIdl;
    }

    public void setOrderIdl(long orderIdl) {
        this.orderIdl = orderIdl;
    }

    public String getType() {
        return type;
    }

    public void setType(String type) {
        this.type = type;
    }

    public ProductOrder() {

    }

    public ProductOrder(long orderIdl, String type) {
        this.orderIdl = orderIdl;
        this.type = type;
    }

    @Override
    public String toString() {
        return "ProductOrder{" +
                "orderIdl=" + orderIdl +
                ", type='" + type + '\'' +
                '}';
    }

    /**
     * 模擬批量建立實體類
     * @return
     */
    public static List<ProductOrder> getOrderList(){
        List<ProductOrder> list=new ArrayList<>();
        list.add(new ProductOrder(111L,"建立訂單"));
        list.add(new ProductOrder(222L,"建立訂單"));
        list.add(new ProductOrder(333L,"建立訂單"));
        list.add(new ProductOrder(111L,"支付訂單"));
        list.add(new ProductOrder(222L,"支付訂單"));
        list.add(new ProductOrder(111L,"完成訂單"));
        list.add(new ProductOrder(222L,"完成訂單"));
        list.add(new ProductOrder(333L,"支付訂單"));
        list.add(new ProductOrder(333L,"完成訂單"));
        return list;
    }
}

控制層：PayController.java

    @Autowired
    private PayProducer payProducer;

    private static final String TOPIC = "ybchen_pay_topic";
    private static final String TOPIC_ORDER = "ybchen_pay_order_topic";

    @RequestMapping("pay_order")
    public Object payOrder() throws Exception{
        //獲取訂單號
        List<ProductOrder> list=ProductOrder.getOrderList();
        for (int i = 0; i < list.size(); i++) {
            ProductOrder order=list.get(i);
            Message message=new Message(TOPIC_ORDER,
                    "",
                    order.getOrderIdl()+"",
                    order.toString().getBytes());
            //發送，同一個訂單id進入同一個隊列中
           SendResult sendResult =payProducer.getProducer().send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message message, Object arg) {
                    Long id=(Long)arg;
                    long index=id%mqs.size();
                    return mqs.get((int) index);
                }
            },order.getOrderIdl());
           //打印輸出結果
            System.out.printf("發送結果=%s，sendResult=%s，orderId=%s，type=%s\n",
                    sendResult.getSendStatus(),
                    sendResult.toString(),
                    order.getOrderIdl(),
                    order.getType());

        }
        return "ok";
    }

改造消費者

package com.ybchen.ybchenmqconsumer.jms;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @ClassName：PayOrderConsumer
 * @Description：消費者-訂單
 * @Author：chenyb
 * @Date：2020/10/18 4:13 下午
 * @Versiion：1.0
 */
@Component
public class PayOrderConsumer {
    /**
     * 生產者所屬的組
     */
    private String producerGroup = "pay_order_consumer_group";
    /**
     * MQ的地址，注意需開放端口號或者關閉防火牆
     */
    private String nameServerAddr = "192.168.199.100:9876;192.168.199.101:9876";
    /**
     * 訂閱主題，訂單
     */
    private static final String TOPIC_ORDER = "ybchen_pay_order_topic";
    private DefaultMQPushConsumer consumer;

    public PayOrderConsumer() throws MQClientException {
        consumer = new DefaultMQPushConsumer(producerGroup);
        //指定NameServer地址，多個地址以;隔開
        //如 producer.setNamesrvAddr("192.168.199.100:9876;192.168.199.101:9876;192.168.199.102:9876")
        consumer.setNamesrvAddr(nameServerAddr);
        //設置消費地點，從最後一個開始消費
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        //訂閱主題，監聽主題下的那些標籤
        consumer.subscribe(TOPIC_ORDER, "*");
        //默認是集羣方式，廣播方式不支持重試
        consumer.setMessageModel(MessageModel.CLUSTERING);
        //註解一個監聽器
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list,
                                                       ConsumeOrderlyContext consumeOrderlyContext) {
                MessageExt msg=list.get(0);
                System.out.printf("%s Receive New Messages: %s %n",Thread.currentThread().getName(),
                        new String(msg.getBody()));
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
        System.out.println("consumer order start ..........");
    }
}

測試順序消息

一個生產者一個消費者

　　能夠看到消費的時候，有點慢，由於我本地安裝了2個虛擬機作一主一從，消費的順序是正確的，都是按照：建立訂單、支付訂單、完成訂單

2020-10-25 13:52:31.822  INFO 1473 --- [nio-8080-exec-1] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring DispatcherServlet 'dispatcherServlet'
2020-10-25 13:52:31.822  INFO 1473 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet        : Initializing Servlet 'dispatcherServlet'
2020-10-25 13:52:31.825  INFO 1473 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet        : Completed initialization in 3 ms
發送結果=SEND_OK，sendResult=SendResult [sendStatus=SEND_OK, msgId=AC10680105C118B4AAC27E92D46F0000, offsetMsgId=C0A8C76400002A9F000000000009C8B2, messageQueue=MessageQueue [topic=ybchen_pay_order_topic, brokerName=broker-a, queueId=3], queueOffset=6]，orderId=111，type=建立訂單
發送結果=SEND_OK，sendResult=SendResult [sendStatus=SEND_OK, msgId=AC10680105C118B4AAC27E92D4930001, offsetMsgId=C0A8C76400002A9F000000000009C9A5, messageQueue=MessageQueue [topic=ybchen_pay_order_topic, brokerName=broker-a, queueId=2], queueOffset=6]，orderId=222，type=建立訂單
發送結果=SEND_OK，sendResult=SendResult [sendStatus=SEND_OK, msgId=AC10680105C118B4AAC27E92D4A90002, offsetMsgId=C0A8C76400002A9F000000000009CA98, messageQueue=MessageQueue [topic=ybchen_pay_order_topic, brokerName=broker-a, queueId=1], queueOffset=6]，orderId=333，type=建立訂單
發送結果=SEND_OK，sendResult=SendResult [sendStatus=SEND_OK, msgId=AC10680105C118B4AAC27E92D4C00003, offsetMsgId=C0A8C76400002A9F000000000009CB8B, messageQueue=MessageQueue [topic=ybchen_pay_order_topic, brokerName=broker-a, queueId=3], queueOffset=7]，orderId=111，type=支付訂單
發送結果=SEND_OK，sendResult=SendResult [sendStatus=SEND_OK, msgId=AC10680105C118B4AAC27E92D4CC0004, offsetMsgId=C0A8C76400002A9F000000000009CC7E, messageQueue=MessageQueue [topic=ybchen_pay_order_topic, brokerName=broker-a, queueId=2], queueOffset=7]，orderId=222，type=支付訂單
發送結果=SEND_OK，sendResult=SendResult [sendStatus=SEND_OK, msgId=AC10680105C118B4AAC27E92D4D00005, offsetMsgId=C0A8C76400002A9F000000000009CD71, messageQueue=MessageQueue [topic=ybchen_pay_order_topic, brokerName=broker-a, queueId=3], queueOffset=8]，orderId=111，type=完成訂單
發送結果=SEND_OK，sendResult=SendResult [sendStatus=SEND_OK, msgId=AC10680105C118B4AAC27E92D4D30006, offsetMsgId=C0A8C76400002A9F000000000009CE64, messageQueue=MessageQueue [topic=ybchen_pay_order_topic, brokerName=broker-a, queueId=2], queueOffset=8]，orderId=222，type=完成訂單
發送結果=SEND_OK，sendResult=SendResult [sendStatus=SEND_OK, msgId=AC10680105C118B4AAC27E92D4DE0007, offsetMsgId=C0A8C76400002A9F000000000009CF57, messageQueue=MessageQueue [topic=ybchen_pay_order_topic, brokerName=broker-a, queueId=1], queueOffset=7]，orderId=333，type=支付訂單
發送結果=SEND_OK，sendResult=SendResult [sendStatus=SEND_OK, msgId=AC10680105C118B4AAC27E92D4F80008, offsetMsgId=C0A8C76400002A9F000000000009D04A, messageQueue=MessageQueue [topic=ybchen_pay_order_topic, brokerName=broker-a, queueId=1], queueOffset=8]，orderId=333，type=完成訂單

ConsumeMessageThread_1 Receive New Messages: ProductOrder{orderIdl=333, type='建立訂單'} 
ConsumeMessageThread_1 Receive New Messages: ProductOrder{orderIdl=333, type='支付訂單'} 
ConsumeMessageThread_1 Receive New Messages: ProductOrder{orderIdl=333, type='完成訂單'} 
ConsumeMessageThread_2 Receive New Messages: ProductOrder{orderIdl=222, type='建立訂單'} 
ConsumeMessageThread_2 Receive New Messages: ProductOrder{orderIdl=222, type='支付訂單'} 
ConsumeMessageThread_2 Receive New Messages: ProductOrder{orderIdl=222, type='完成訂單'} 
ConsumeMessageThread_3 Receive New Messages: ProductOrder{orderIdl=111, type='建立訂單'} 
ConsumeMessageThread_3 Receive New Messages: ProductOrder{orderIdl=111, type='支付訂單'} 
ConsumeMessageThread_3 Receive New Messages: ProductOrder{orderIdl=111, type='完成訂單'}

一個生產者3個消費者

　　消費者會平均分配queue的數量，消費者數量小於等於4！！！

　　本地在線模擬，一個生產者、3個消費者場景，看看消費的順序，內容較長，被分割3塊

消費者核心配置

setConsumeFromWhere

CONSUME_FORM_FIRST_OFFSET：初次從消息隊列頭部開始消費，即歷史消息(還存儲在broker的)所有消費一遍，後續在啓動接着上次消費的進度開始消費
CONSUME_FROM_LAST_OFFSET：默認策略，初次從該隊列尾開始消費，即跳過歷史當心，後續在啓動接着上次消費的進度開始消費
CCONSUME_FROM_TIMESTAMP：從某個時間點開始消費，默認是半小時之前，後續在啓動接着上次消費的進度開始消費

setAllocateMessageQueueStrategy

負載均衡策略算法，即消費者分配到queue的算法，默認值AllocateMessageQueueAveragely即取模平均分配

setOffsetStore

消息消費進度存儲器，2個策略
- LocalFileOffsetStore(廣播模式默認使用)
- RemoteBrokerOffsetStore(集羣模式默認使用)

setConsumeThreadMin

最小消費線程池數量

setConsumeThreadMax

最大消費線程池數量

setPullBatchSize

消費者去broker拉取消息時，一次次拉取多少條

setConsumeMessageBatchMaxSize

單次消費時一次性消費多少條消息

setMessageModel

消費者消費模式
- CLUSTERING：默認是集羣模式
- BROADCASTING：廣播模式

Topic下隊列的奇偶數會影響Customer個數裏面的消費數量

若是是4個隊列(默認隊列爲4)，8個消息，4個節點則各會消費2條，若是不對等，則負載均衡會分配不均勻
若是consumer實例數量比message queue的總數量還多的話，多出來的consumer實例將沒法分到queue，也就沒法消費達到消息，也就沒法起到分攤負載的做用，因此須要控制讓queue的總數量大於consumer的數量

集羣模式(默認)

Consumer實例平均分攤消費生產者發送的消息
例如：訂單消息，只能被消費一次

廣播模式

廣播模式下消費消息，投遞到Broker的消息會被每一個Consumer進行消費，一條消息被多個Consumer消費，廣播消費中ConsumerGroup暫時無用
例如：QQ羣，羣主發一條消息，全部人均可以看到

消息存儲

ConsumeQueue

　　邏輯隊列，默認存儲位置：/root/store/consumequeue

CommitLog

　　真正存儲消息文件的，默認存儲位置：/root/store/commitlog

常見面試題

爲何消息隊列？

優勢

異步：例如秒殺，可使用，點我直達
解耦
削峯：秒殺狀況下，一個個入隊，一個個出隊，有序進行

缺點

系統可用性越低：外部依賴越多，依賴越多，出問題風險越大
系統複雜性提升：須要考慮多種場景，好比消息重複消費、消息丟失
須要更多的機器和人力：消息隊列通常集羣部署，須要運維和監控

如何避免重複消費？

　　RocketMQ不保證消息不重複，若是業務保證嚴格的不能重複消費，須要本身去業務端去重

數據庫表去重

　　指定某個字段惟一值

setNX

　　利用Redis的特性分佈式鎖，下面是我以前的代碼，待改造

package com.cyb.redis.utils;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

public class jedisUtils {
    private static String ip = "192.168.31.200";
    private static int port = 6379;
    private static JedisPool pool;
    static {
        pool = new JedisPool(ip, port);
    }
    public static Jedis getJedis() {
        return pool.getResource();
    }
    public static boolean getLock(String lockKey, String requestId, int timeout) {
        //獲取jedis對象，負責和遠程redis服務器進行鏈接
        Jedis je=getJedis();
        //參數3：NX和XX
        //參數4：EX和PX
        String result = je.set(lockKey, requestId, "NX", "EX", timeout);
        if (result=="ok") {
            return true;
        }
        return false;
    }

    public static synchronized boolean getLock2(String lockKey, String requestId, int timeout) {
        //獲取jedis對象，負責和遠程redis服務器進行鏈接
        Jedis je=getJedis();
        //參數3：NX和XX
        //參數4：EX和PX
        Long result = je.setnx(lockKey, requestId);
        if (result==1) {
            je.expire(lockKey, timeout); //設置有效期
            return true;
        }
        return false;
    }
}

Redis原子遞增

　　利用Redis的incr特性，若是大於0說明消費過了(須要設置過時時間)

如何保證消息的可靠性傳輸？

producer端

不採用oneway發送，使用同步或者一部方式發送，作好重試，可是重試的Message key必須惟一
投遞的日誌須要保存，關鍵字段、投遞時間、投遞狀態、重試次數、請求體、響應體等

broker端

雙主雙從架構，NameServer須要多節點
同步雙寫，異步刷盤

consumer端

消息消費保存日誌文件中

大量堆積到broker裏面，如何處理？

臨時topic隊列擴容，提升消費者能力
編寫臨時處理分發程序，從舊topic快速讀取到臨時新topic中，新topic的queue數量擴容多倍，而後再啓動更多consumer進行臨時新的topic消費

RocketMQ高性能的緣由？

MQ架構配置

順序寫
隨機讀
零拷貝

發送端高可用

雙主雙從架構：建立Topic的時候，MessageQueue建立在多個Broker上，即相同的Broker名稱，不一樣brokerid；當一個Master不可用時，組內其餘的Master仍然可用

消費高可用

主從架構：Broker角色，Master提供讀寫，Slave只支持讀
Consumer不用配置，當Master不可用或者繁忙的時候，Consumer會自動切換到Slave節點進行讀取

提高消息的消費能力

增長多個消費者
修改消費者的線程池最小/大數量

項目源碼

案例源碼

連接: https://pan.baidu.com/s/1Q8iL0lH-bdFEycYGq61hQg  密碼: rww2

Linux下RocketMQ安裝包

連接: https://pan.baidu.com/s/1dkE7sAs9E4TjwDQ38Pv4_A  密碼: mkjm

雙主雙從集羣搭建

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。

RocketMQ 4.7.1 環境搭建、集羣、SpringBoot整合MQ

導讀

部署

官網

前置條件

下載

安裝依賴項

mq上傳至linux

解壓

maven編譯

啓動NameServer

後臺啓動方式

NameServer啓動時內存不足(問題解決)

啓動Broker

Broker內存不足（問題解決）

服務都啓動成功

模擬消費

注意

可視化控制檯

官網地址

解壓

安裝編譯

啓動

SpringBoot整合RocketMQ(生產者)

建立SpringBoot項目

項目結構

加入依賴

PayProducer.java

PayController.java

測試

常見錯誤

錯誤一

錯誤2

錯誤3

錯誤4

檢索消息發送

SpringBoot整合RocketMQ(消費者)

建立SpringBoot項目

項目結構

加入依賴

PayConsumer.java

application.properties

測試生產者消費者

MQ集羣架構模式分析

單節點

優勢

缺點

主從(異步、同步雙寫)

優勢

缺點

雙主

優勢

缺點

雙主雙從，多主多從模式(異步複製)

優勢

缺點

雙主雙從，多主多從模式(同步雙寫)

優勢

缺點

推薦

主從集羣搭建

準備工做

啓動兩臺nameserver

編輯並啓動roccketmq

注意事項

啓動broker

使用管控臺

集羣測試

故障演練

總結

場景模擬

生產和消費重試及處理

生產者重試

消費者重試

模擬消息重發

異步發送消息和回調實戰

應用場景

官方例子

改造生產者

演示