Linux系統 Centos7 環境基於Docker部署Rocketmq服務

阿新 • • 發佈：2020-07-07

訊息佇列

基本概述

MQ,Message Queue,基於TCP協議構建的簡單協議，區別於具體的通訊協議。

基於通訊協議定義和抽象的更高層次的通訊模型,一般都是生產者和消費者模型，又或者說服務端和客戶端模型。
生產者/消費者模型：一般通過定義生產者和消費者實現訊息通訊從而遮蔽複雜的底層通訊協議。應用於分散式應用系統，而且為之提供非同步解耦和削峰填谷的能力，同時也具備網際網路應用所需的海量堆積，高吞吐和可靠性重試機制的特性。

核心概念

訊息主題：Message Topic,一級訊息型別，生產者向其傳送訊息
訊息生產者：Message Producer，訊息釋出者，或者訊息服務終端，負責生產訊息和傳送訊息到訊息主題

訊息消費者：Message Consumer，訊息訂閱者，或者訊息客戶終端，負責從訊息主題接收並處理消費訊息
訊息實體：Message Object,訊息物件，生產者向訊息主題傳送並最終傳送給訊息者的資料和屬性的符號以及組合
訊息屬性：Message Attributes,訊息內容，生產者對訊息進行抽象和定義的相關屬性，包含Message Key 和Message Target
訊息組：Message Group，訊息分類組別，通稱一類生產者和消費者，通常生產和消費同一類訊息，且訊息釋出和訂閱的邏輯基本一致

程式設計思想

資料結構：訊息佇列的資料結構採用FIFO方式來定義與實現

設計模式：採用觀察者模式

觀察者模式：定義物件間一對多的依賴關係，當一個物件的狀態發生改變時，所有依賴於它的物件都得到通知自動更新

訊息流派：

有Broker：基於訊息中介軟體，所有訊息通過中介軟體中轉，基於Broker把訊息推送給消費者[消費者主動輪詢]

1.重Topic：Kafka和RocketMq,Activemq，生產者把訊息傳送Key和資料到Broker，然後由Broker比較Key之後決定推送給具體的消費者[推送訊息給消費者]

2.輕Topic：Rabbitmq[AMQP],生產者把生產訊息且傳送Key和資料，消費者定義訂閱的佇列，Broker收到請求之後，按照一定的邏輯和規則計算出相應的Key對應的佇列，然後把訊息推送給具體的消費者[消費者向生產者訂閱訊息，消費者主動拉取訊息]

無Broker：基於Socket網路[傳輸層]直連通訊，代表作ZeroMQ

基本概述

NameServer：名稱伺服器[MQ名稱空間伺服器],大致相當於 jndi技術，更新和發現 broker服務。用於儲存Broker相關元資訊，並給生產者和消費者查詢Broker訊息。每個Broker在啟動都會在名稱伺服器[NameServer]註冊,生產者在傳送訊息前會根據訊息主題到名稱伺服器查詢獲取Broker路由訊息，消費者也會定時獲取主題的路由訊息。

ps：屬於無狀態服務設計，可橫向擴充套件，節點之間無通訊，可以部署多臺機器來標識偽叢集。

Broker:訊息儲存中心[訊息中轉角色]，負責儲存和轉發訊息。接收來自生產者的訊息並進行儲存，消費者從這拉取訊息。儲存與訊息相關的元資料，主要包括使用者組,訊息進度偏移量，佇列訊息等。其中Broker分為Master和Slave節點：

Master節點：可讀可寫
Slave節點：只可讀不可寫

其部署方式：

單機Master:Broker重啟和宕機之後服務不可用,不建議生產使用
多機Master:所有訊息伺服器均是Master,沒有Slave。

優點：配置簡單，單個Master重啟和宕機維護對應用無影響

缺點：單機重啟和宕機期間，機器上未被消費的訊息在機器恢復期間不可訂閱，訊息實時性會受影響
多機Master多機Slave[同步雙寫]:

優點：資料和服務無單點故障，Master重啟和宕機訊息無延遲,資料和服務可用高

缺點：相對非同步複製效能較低，傳送訊息的延時較高
多機Master多機Slave[非同步複製]:

優點：訊息丟失較少，實時性高，Master重啟和宕機可繼續從Slave消費

缺點：Master重啟和宕機後在磁碟損壞會導致訊息丟失，但是情況比較少

ps：一個Master可以對應多個Slave，但是一個Slave只能對應一個Master。Master和Slave的對應關係通過指定相同的BrokerName，不同的BrokerId來定義。BrokerId為0表示Master，BrokerId非0表示Slave。然後所有的Broker和Name Server上的節點建立長連線,定時註冊Topic資訊到所有Name Server。

Producer：訊息生產者->負責生產訊息,生產者向訊息伺服器傳送業務應用程式生成的訊息。主要有同步傳送和非同步傳送方式兩種，其中：

同步傳送：訊息傳送方發出資料後，在訊息接收方傳送響應之後再傳送下一個資料。一般適用於重要訊息通知場景[重要通知郵件，營銷簡訊推送]
非同步傳送：訊息傳送方發出資料後，不用等待接收方發回響應，接著傳送下一個資料。一般適用於鏈路耗時較長，而對響應時間敏感的業務場景[視訊上傳通知啟動自動轉碼服務處理通知]
單向傳送：負責傳送訊息而不等待伺服器迴應且沒有回撥函式觸發。一般適用於對可靠性要求不高的業務場景[日誌收集]

ps：Producer與Name Server其中一個節點建立連線。定期從Name Server取Topic資訊。並與提供該Topic資訊的Master建立長連線。Producer也可以叢集部署。

Consumer：訊息消費者 負責消費訊息，從訊息伺服器拉取訊息並將其輸入使用者應用程式中。主要分為拉取型消費者和推送型消費者：

拉取型消費者：Pull Consumer->主動從訊息伺服器拉取訊息，只要批量拉取訊息,使用者就會啟動消費過程

推送型消費者：Push Consumer->封裝訊息的拉取,消費進度和其它內部維護工作，訊息到達之後便執行回撥介面留給使用者應用程式來實現。屬於被動消費型別，Push拉取時需要註冊訊息費者監聽器，當監聽器被觸發之後開始消費訊息。

ps：Consumer 與Name Server 叢集中的其中一個節點（隨機選擇）建立長連線，定期從Name Server 取Topic 路由資訊，並向提供Topic服務的Master、Slave建立長連線，且定時向Master、Slave傳送心跳。Consumer既可以從Master訂閱訊息，也可以從Slave訂閱訊息，訂閱規則由Broker配置決定。

部署Rocketmq

拉取rocketmq映象：docker pull foxiswho/rocketmq

1.查詢映象images：docker search rocketmq

[[email protected] ~]# docker search rocketmq

NAME                               DESCRIPTION                                     STARS               OFFICIAL            AUTOMATED

styletang/rocketmq-console-ng      rocketmq-console-ng                             20

rocketmqinc/rocketmq               Image repository for Apache RocketMQ            17

foxiswho/rocketmq                  rocketmq                                        14

laoyumi/rocketmq                                                                   10                                      [OK]

xlxwhy/rocketmq                    alibaba's rocketmq                              4

huanwei/rocketmq-broker                                                            2

2019liurui/rocketmq-broker         RocketMQ broker image for RocketMQ-Operator     1

2019liurui/rocketmq-namesrv        RocketMQ name service image for RocketMQ-Ope…   1

apacherocketmq/rocketmq            Docker Image for Apache RocketMQ                1

rocketmqinc/rocketmq-operator      The Kubernetes operator for RocketMQ            0

2019liurui/rocketmq-operator       Kubernetes Operator for RocketMQ !              0

apacherocketmq/rocketmq-operator   RocketMQ Operator is to manage RocketMQ serv…   0

coder4/rocketmq                    rocketmq                                        0                                       [OK]

rocketmqinc/rocketmq-namesrv       Customized RocketMQ Name Server Image for Ro…   0

rocketmqinc/rocketmq-broker        Customized RocketMQ Broker Image for RocketM…   0

slpcat/rocketmq-console-ng                                                         0

huanwei/rocketmq                                                                   0

huanwei/rocketmq-broker-k8s                                                        0

king019/rocketmq                   rocketmq                                        0

pengzu/rocketmq-console-ng         web console for rocketmq ,this code is from …   0

fengzt/rocketmq-broker             apache rocketmq 4.2.0 broker server(官方檔案…       0

huanwei/rocketmq-operator                                                          0

slpcat/rocketmq                                                                    0

fengzt/rocketmq-nameserver         apache rocketmq 4.2.0 nameserver                0

icyblazek/rocketmq                 RocketMQ                                        0

[[email protected] ~]#

2.執行：docker pull foxiswho/rocketmq

3.建立docker儲存根目錄並且授權:

mkdir docker && chmod -R 777 docker/

[[email protected] /]# cd docker/

[[email protected] docker]# pwd

/docker

[[email protected] docker]#

4.部署名稱伺服器rocketmq-namesrv-server[9876]:

rocketmq-namesrv-server

docker run -itd --restart=always --privileged=true -p 9876:9876 --name rocketmq-namesrv-server -v /docker/rocketmq/namesrv/logs:/root/rocketmq/logs -v /docker/rocketmq/namesrv/store:/root/rocketmq/store -e "MAX_POSSIBLE_HEAP=100000000" -e "JAVA_OPTS=-Duser.home=/opt" -e "JAVA_OPT_EXT=-server -Xms512m -Xmx512m -Xmn256m" rocketmqinc/rocketmq sh mqnamesrv

docker run -itd --restart=always --privileged=true -p 9876:9876 --name  rocketmq-namesrv-server -v /docker/rocketmq/namesrv/logs:/root/rocketmq/logs -v /docker/rocketmq/namesrv/store:/root/rocketmq/store -e "MAX_POSSIBLE_HEAP=100000000" -e "JAVA_OPTS=-Duser.home=/opt" -e "JAVA_OPT_EXT=-server -Xms512m -Xmx512m -Xmn256m" rocketmqinc/rocketmq sh mqnamesrv

ps：建立/docker/rocketmq/namesrv/logs和/docker/rocketmq/namesrv/store目錄

5.部署訊息伺服器rocketmq-broker-server[10911]：

rocketmq-broker-server

docker run -itd --restart=always --privileged=true -p 10909:10909 -p 10911:10911 -p 10912:10912 --name rocketmq-broker-server --link rocketmq-namesrv-server:namesrv -v /docker/rocketmq/broker/logs:/root/rocketmq/logs -v /docker/rocketmq/broker/store:/root/rocketmq/store -v /docker/rocketmq/broker/conf/broker.conf:/opt/rocketmq-4.4.0/conf/broker.conf -e "NAMESRV_ADDR=namesrv:9876" -e "MAX_POSSIBLE_HEAP=200000000" -e "JAVA_OPTS=-Duser.home=/opt" -e "JAVA_OPT_EXT=-server -Xms512m -Xmx512m -Xmn256m" rocketmqinc/rocketmq sh mqbroker -c /opt/rocketmq-4.4.0/conf/broker.conf

docker run -itd --restart=always --privileged=true -p 10909:10909 -p 10911:10911 -p 10912:10912 --name  rocketmq-broker-server --link rocketmq-namesrv-server:namesrv -v /docker/rocketmq/broker/logs:/root/rocketmq/logs -v /docker/rocketmq/broker/store:/root/rocketmq/store -v /docker/rocketmq/broker/conf/broker.conf:/opt/rocketmq-4.4.0/conf/broker.conf -e "NAMESRV_ADDR=namesrv:9876" -e "MAX_POSSIBLE_HEAP=200000000" -e "JAVA_OPTS=-Duser.home=/opt" -e "JAVA_OPT_EXT=-server -Xms512m -Xmx512m -Xmn256m" rocketmqinc/rocketmq sh mqbroker -c /opt/rocketmq-4.4.0/conf/broker.conf

ps：

1.建立/docker/rocketmq/broker/logs和/docker/rocketmq/broker/store目錄

2.編寫broker.conf配置檔案：

brokerClusterName = rocketmq-cluster

brokerName = broker-server

brokerId = 0

deleteWhen = 04

fileReservedTime = 48

# Broker 的角色

# - ASYNC_MASTER 非同步複製Master

# - SYNC_MASTER 同步雙寫Master

# - SLAVE

brokerRote=ASYNC_MASTER

brokerRole = ASYNC_MASTER

flushDiskType = ASYNC_FLUSH

# 如果是本地程式呼叫雲主機 mq，這個需要設定成 雲主機 IP

brokerIP1=Server-IP

#限制的訊息大小

maxMessageSize=65536

# 檢測物理檔案磁碟空間

diskMaxUsedSpaceRatio=88

mapedFileSizeCommitLog=1073741824

mapedFileSizeConsumeQueue=50000000

#併發send執行緒數，多執行緒來傳送訊息可能會出現broker busy

sendMessageThreadPoolNums=128

useReentrantLockWhenPutMessage=true

#在傳送訊息時，自動建立伺服器不存在的topic，預設建立的佇列數

defaultTopicQueueNums=8

highSpeedMode=false

commercialBaseCount=1

maxErrorRateOfBloomFilter=20

accessMessageInMemoryMaxRatio=40

#無讀寫客戶端存活時間

clientChannelMaxIdleTimeSeconds=120

flushDelayOffsetInterval=10000

serverSocketRcvBufSize=131072

#單次 Pull 訊息(記憶體)傳輸的 最大位元組數

maxTransferBytesOnMessageInMemory=262144

clientManageThreadPoolNums=32

serverChannelMaxIdleTimeSeconds=120

serverCallbackExecutorThreads=0

enablePropertyFilter=false

transientStorePoolSize=5

enableConsumeQueueExt=false

#rocketmq server config

serverPooledByteBufAllocatorEnable=true

serverSocketRcvBufSize=131072

#rocketmq client config

6.部署控制後臺rocketmq-consloe-server[8082]:

rocketmq-consloe-server

docker run -itd -p 8082:8080 --restart=always --privileged=true --name rocketmq-console-server -v /docker/rocketmq/console/data:/tmp -e "JAVA_OPTS=-Drocketmq.namesrv.addr=47.104.22.10:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false" -e "JAVA_OPT_EXT=-server -Xms512m -Xmx512m -Xmn256m -Duser.home=/opt" styletang/rocketmq-console-ng:latest

docker run -itd -p 8082:8080  --restart=always --privileged=true  --name rocketmq-console-server -v /docker/rocketmq/console/data:/tmp -e "JAVA_OPTS=-Drocketmq.namesrv.addr=Server-IP:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false" -e "JAVA_OPT_EXT=-server -Xms512m -Xmx512m -Xmn256m -Duser.home=/opt" styletang/rocketmq-console-ng:latest

ps：

1.建立/docker/rocketmq/console/data目錄

2.Server-IP為broker.conf配置brokerIP1值

7.最終部署結果：

整合Rocketmq開發實戰

1.配置Rocketmq的Maven依賴：

<!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-client -->

<dependency>

    <groupId>org.apache.rocketmq</groupId>

    <artifactId>rocketmq-client</artifactId>

    <version>4.4.0</version>

</dependency>

ps：

1.目前rocketmq最新版本是4.7.1：

<!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-client -->

<dependency>

    <groupId>org.apache.rocketmq</groupId>

    <artifactId>rocketmq-client</artifactId>

    <version>4.7.1</version>

</dependency>

2.這裡選擇的是rocketmq版本[4.4.0],因為部署是4.4.0版本，推薦部署版本相同的rocketmq

2.建立pivotal-cloud-queue工程:

ps：建立訊息佇列統一模組工程，在業務應用程式服務模組依賴該工程，並封裝訊息生產者[Producer]和訊息消費者[Consumer]處理類。

3.封裝rocketmq屬性配置類:

新增spring-boot-configuration-processor依賴:

  <dependency>

    <groupId>org.projectlombok</groupId>

    <artifactId>lombok</artifactId>

    <version>1.18.12</version>

    <scope>provided</scope>

 </dependency>

 <dependency>

   <groupId>org.springframework.boot</groupId>

   <artifactId>spring-boot-configuration-processor</artifactId>

   <optional>true</optional>

 </dependency>

編寫RocketmqProperties屬性配置類:

package com.pivotal.cloud.queue.properties;

import lombok.Data;

import org.springframework.boot.context.properties.ConfigurationProperties;

import org.springframework.context.annotation.Configuration;

/**

 * @className: com.pivotal.cloud.queue.properties.RocketmqProperties

 * @title: RocketmqProperties

 * @description: 封裝Pivotal專案RocketmqProperties類

 * @content: PivotalCloud專案系統RocketmqProperties自定義屬性配置類

 * @author: marklin

 * @datetime: 2020-07-07 01:32

 * @version: 1.0.0

 * @copyright: Copyright  2018-2020 Pivotal Systems Incorporated. All rights reserved.

 */

@Data

@Configuration

@ConfigurationProperties(prefix = "pivotal.cloud.rocketmq")

public class RocketmqProperties {

    /**

     * rocketmq訊息佇列生產者-Producer

     */

    private final Producer producer = new Producer();

    /**

     * rocketmq訊息佇列消費者-Consumer

     */

    private final Consumer consumer = new Consumer();

    @Data

    public static class Producer {

    }

    @Data

    public static class Consumer {

    }

}

編寫RocketmqConfiguration配置類:

package com.pivotal.cloud.queue.configuration;

import com.pivotal.cloud.queue.properties.RocketmqProperties;

import lombok.AllArgsConstructor;

import org.springframework.boot.context.properties.EnableConfigurationProperties;

import org.springframework.context.annotation.Configuration;

/**

 * @className: com.pivotal.cloud.queue.configuration.RocketmqConfiguration

 * @title: RocketmqConfiguration

 * @description: 封裝Pivotal專案RocketmqConfiguration類

 * @content: //TODO

 * @author: marklin

 * @datetime: 2020-07-07 02:25

 * @version: 1.0.0

 * @copyright: Copyright  2018-2020 Pivotal Systems Incorporated. All rights reserved.

 */

@AllArgsConstructor

@Configuration(proxyBeanMethods = false)

@EnableConfigurationProperties({RocketmqProperties.class})

public class RocketmqConfiguration {

}

編寫RocketmqTemplate生產者模板類:

package com.pivotal.cloud.queue.template;

/**

 * @className: com.pivotal.cloud.queue.template.RocketmqTemplate

 * @title: RocketmqTemplate

 * @description: 封裝Pivotal專案RocketmqTemplate類

 * @content: PivotalCloud專案系統RocketmqTemplate生產者模板類

 * @author: marklin

 * @datetime: 2020-07-07 02:48

 * @version: 1.0.0

 * @copyright: Copyright  2018-2020 Pivotal Systems Incorporated. All rights reserved.

 */

public class RocketmqTemplate {

}

編寫RocketmqListener消費者監聽器類:

package com.pivotal.cloud.queue.listener;

/**

 * @className: com.pivotal.cloud.queue.listener.RocketmqListener

 * @title: RocketmqListener

 * @description: 封裝Pivotal專案RocketmqListener類

 * @content: PivotalCloud專案系統RocketmqListener消費者監聽器類

 * @author: marklin

 * @datetime: 2020-07-07 02:52

 * @version: 1.0.0

 * @copyright: Copyright  2018-2020 Pivotal Systems Incorporated. All rights reserved.

 */

public interface RocketmqListener {

}

編寫META-INF/spring.factories工廠類:

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\

  com.pivotal.cloud.queue.configuration.RocketmqConfiguration,\

  com.pivotal.cloud.queue.configuration.RabbitmqConfiguration,\

  com.pivotal.cloud.queue.configuration.ActivemqConfiguration,\

  com.pivotal.cloud.queue.configuration.KafkaConfiguration,

版權宣告：本文為博主原創文章，遵循相關版權協議，如若轉載或者分享請附上原文出處連結和連結來源。

Linux系統 Centos7 環境基於Docker部署Rocketmq服務

訊息佇列

基本概述

核心概念

程式設計思想

訊息流派：

基本概述

部署Rocketmq

rocketmq-namesrv-server

rocketmq-broker-server

rocketmq-consloe-server

整合Rocketmq開發實戰

Linux系統 Centos7 環境基於Docker部署Rocketmq服務

Linux系統環境基於Docker搭建Mysql資料庫服務實戰

Linux系統Centos7 基於Docker搭建ELK分散式日誌系統

基於Docker部署GitLab環境搭建

基於Kubernetes可擴充套件的Selenium 並行自動化測試部署及搭建(1)——Win10環境下Docker部署

centos7.9環境基於docker和pipeline構建jenkins的ci平臺

Linux系統中Java Web應用部署環境的安裝與配置（3）--Tomcat的安裝與配置

基於Docker部署LNMP架構

Linux系統CentOS7版yum安裝mongoBD(3步完成)

基於docker部署prometheus監控平臺

Amazon免費CE2基於docker部署nginx,並實現訪問

基於docker部署skywalking實現全鏈路監控

[redis] linux下叢集篇(4) docker部署

Linux系統Centos7安裝Mysql

VMware虛擬機器Centos7環境下docker安裝

jenkins 基於docker部署springboot專案詳細步驟

Docker部署RocketMQ

VMware安裝linux系統CentOs7.4 mini版過程

基於docker部署的discuz備份

linux系統安裝（基於Mac）

Linux系統 Centos7 環境基於Docker部署Rocketmq服務

訊息佇列

基本概述

核心概念

程式設計思想

訊息流派：

基本概述

部署Rocketmq

rocketmq-namesrv-server

rocketmq-broker-server

rocketmq-consloe-server

整合Rocketmq開發實戰

相關推薦