分布式消息中间件RocketMQ学习教程

一时失言乱红尘 2022-05-05 14:00 340阅读 0赞

*  消息中间件简介
 *  常用消息队列对比
    
     *  RocketMQ
     *  Kafka
     *  ActiveMQ
     *  RabbitMQ
     *  ZeroMQ
 *  RocketMQ简介
    
     *  RocketMQ主要组成
        
         *  NameServer
         *  Producer
         *  Consumer
         *  Broker
 *  RocketMQ存储特点
 *  RocketMQ集群配置
 *  附录安装教程
    
     *  JDK安装
        
         *  在usr目录下创建java目录
         *  进入java目录
         *  下载jdk到usrjava目录然后tar解压可以使用xftp在本地下载jdk或者使用命令curl -O url下载
         *  设置环境变量
         *  vi编辑
         *  profile文件加入如下配置如何按esc退出wq保持
         *  让修改生效
         *  验证jdk有效性
         *  设置环境变量其它操作同上
     *  RocketMQ安装
        
         *  解压并复制到usrlocal目录
         *  进入usrlocal目录
         *  建立一个同步的链接使用ln命令
         *  进入同步的文件夹
         *  vim编辑配置文件
         *  编辑JAVA内存块
         *  创建数据目录
         *  启动RocketMQ
         *  查看启动日志
         *  查看启动端口
         *  关闭RocketMQ
         *  再次启动

# 消息中间件简介 #

消息队列(Message Queue，简称MQ)。消息中间件作为实现分布式消息系统可拓展、可伸缩性的关键组件，具有高吞吐量、高可用等等优点。

# 常用消息队列对比 #

常用的消息队列有RocketMQ、kafka、ActiveMQ、RabbitMQ、ZeroMQ等等  
activemq rabbitmq 与 kafka、rocketmq有很大的区别就是前2个只支持主从模式，后2个是分布式消息系统，支持分布式。  
持久化消息比较： zeroMq不支持，activeMq和rabbitMq都支持。

## RocketMQ ##

RocketMQ 是一款开源的分布式消息系统，基于高可用分布式集群技术，提供低延时的、高可靠的消息发布与订阅服务。

## Kafka ##

Kafka是没有重发机制的消息队列。它以可水平扩展和高吞吐率而被广泛使用。目前越来越多的开源分布式处理系统如Cloudera、Apache Storm、Spark都支持与Kafka集成。kafka不完全符合jms规范，注重吞吐量，类似udp 和 tcp。kafka吞吐量大 ，对数据不是百分之百保证的，会有数据丢失，不是百分百送达。所以kafka适合大数据量流转， 比如日志数据 比如用作统计的数据

## ActiveMQ ##

ActiveMQ类似于ZemoMQ，它可以部署于代理模式和P2P模式。类似于RabbitMQ，它易于实现高级场景，而且只需付出低消耗。

## RabbitMQ ##

RabbitMQ是AMQP协议领先的一个实现，它实现了代理(Broker)架构，意味着消息在发送到客户端之前可以在中央节点上排队。此特性使得RabbitMQ易于使用和部署，适宜于很多场景如路由、负载均衡或消息持久化等，用消息队列只需几行代码即可搞定。但是，这使得它的可扩展性差，速度较慢，因为中央节点增加了延迟，消息封装后也比较大。

## ZeroMQ ##

ZeroMQ是一个非常轻量级的消息系统，专门为高吞吐量/低延迟的场景开发，在金融界的应用中经常可以发现它。

# RocketMQ简介 #

RocketMQ 是一款分布式、队列模型的消息中间件。RocketMQ是由Metaq发展而来，Metaq3.0的时候，改名RocketMQ。  
RocketMQ 是一款开源的分布式消息系统，基于高可用分布式集群技术，提供低延时的、高可靠的消息发布与订阅服务。

## RocketMQ主要组成： ##

RocketMQ主要组成：NameServer、Broker（代理）、Producer（消  
息生产者）、Cosumer（消息消费者）

### NameServer ###

NameServer： rocketmq名称服务器，大致相当于 jndi技术，更新和发现 broker服务。一个几乎无状态节点，可集群部署，节点之间无任何信息同步

### Producer ###

Producer： 消息生产者。

> Producer与Name Server其中一个节点建立连接。定期从Name Server取Topic信息。并与提供该Topic信息的Master建立长连接。Producer也可以集群部署。

### Consumer ###

Consumer： 消息消费者。

> Consumer 与Name Server 集群中的其中一个节点（随机选择）建立长连接，定期从Name Server 取Topic 路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，订阅规则由Broker配置决定。

### Broker ###

Broker：消息中转角色，负责存储和转发消息。Broker分为Master和Slave。

> 一个Master可以对应多个Slave，但是一个Slave只能对应一个Master。Master和Slave的对应关系通过指定相同的BrokerName，不同的BrokerId来定义。BrokerId为0表示Master，BrokerId非0表示Slave。然后所有的Broker和Name Server上的节点建立长连接,定时注册Topic信息到所有Name Server。

Broker-Master：broker 消息主机服务器

Broker-Slave： broker 消息从机服务器

![这里写图片描述][SouthEast]

# RocketMQ存储特点 #

Consumer消费消息过程，使用了零拷贝，零拷贝包含以下两种方式：

1.  使用mmap + write方式

优点：即使频繁调用，使用小块文件传输，效率也很高

缺点：不能很好的利用DMA方式，会比sendfile多消耗CPU，内存安全性控制复杂，需要避免JVM Crash问题。

1.  使用sendfile方式

优点：可以利用DMA方式，消耗CPU较少，大块文件传输效率高，无内存安全新问题。

缺点：小块文件效率低于mmap方式，只能是BIO方式传输，不能使用NIO。

RocketMQ选择了第一种方式，mmap+write方式，因为有小块数据传输的需求，效果会比sendfile更好。

# RocketMQ集群配置 #

RocketMQ集群配置  
![这里写图片描述][SouthEast 1]

# 附录：安装教程 #

[http://www.linuxidc.com/Linux/2016-09/134941.htm][http_www.linuxidc.com_Linux_2016-09_134941.htm]

## 【JDK安装】 ##

因为RocketMQ是基于Java开发的，所以安装RocketMQ之前，我们需要先安装JDK，因为服务器一般采用Linux，所以本博客只介绍基于Linux系统的教程

### 1.在/usr/目录下创建java目录 ###

mkdir /usr/java

### 进入java目录 ###

cd /usr/java

### 2.下载jdk到/usr/java目录，然后tar解压(可以使用xftp在本地下载jdk，或者使用命令curl -O url下载) ###

tar -zxvf jdk-8u102-linux-x64.tar.gz

### 3.设置环境变量 ###

### vi编辑 ###

vi /etc/profile

### profile文件加入如下配置，如何按esc退出，:wq!保持 ###

export JAVA\_HOME=/usr/java/jdk1.8.0\_102  
export PATH=PATH:PATH:JAVA\_HOME/bin  
export CLASSPATH=.:$JAVA\_HOME/lib

### 让修改生效 ###

source /etc/profile

### 验证jdk有效性 ###

java -version  
假如系统有安装yum，就可以直接使用yum install java-1.7.0-openjdk下载jdk

### 设置环境变量，其它操作同上 ###

export JAVA\_HOME=/usr/java/jdk1.8.0\_102  
export PATH=PATH:PATH:JAVA\_HOME/bin  
export CLASSPATH=.:$JAVA\_HOME/lib

## 【RocketMQ安装】 ##

[http://www.linuxidc.com/Linux/2015-10/124112.htm][http_www.linuxidc.com_Linux_2015-10_124112.htm]

### 解压并复制到/usr/local目录 ###

tar zxf alibaba-rocketmq-3.2.2.tar.gz -C /usr/local/

### 进入/usr/local目录 ###

cd /usr/local/

### 建立一个同步的链接,使用ln命令 ###

ln -s /usr/local/alibaba-rocketmq /usr/local/rocketmq

### 进入同步的文件夹 ###

cd rocketmq/

### vim编辑配置文件 ###

vim conf/2m-noslave/broker-a.properties  
参考配置文件  
brokerClusterName=FusionCluster  
brokerName=broker-a  
brokerId=0  
namesrvAddr=dbTest249:9876;webTest251:9876  
deleteWhen=04  
fileReservedTime=120  
brokerRole=ASYNC\_MASTER  
flushDiskType=ASYNC\_FLUSH  
sstorePathRootDir=/data/rocketmq/store  
storePathCommitLog=/data/rocketmq/store/commitlog

### 编辑JAVA内存块 ###

vim bin/runbroker.sh  
JAVA\_OPT\_1=”-server -Xms512m -Xmx1g -XX:PermSize=128m -XX:MaxPermSize=320m”

### 创建数据目录 ###

mkdir -p /data/rocketmq/store/commitlog  
mkdir /data/logs  
cd conf/  
sed -i ‘s\#\{user.home\}\#/data\#g’ \*.xml　　　　　　  
//将conf目录下所有xml文件中的\{user.home\}\#/data\#g’ \*.xml　　　　　　 //将conf目录下所有xml文件中的\{user.home\}替换成/data,进入rocketmq/conf目录下执行该命令

### 启动RocketMQ ###

cd ../bin/  
nohup sh mqnamesrv >/var/log/ns.log 2>&1 &  
nohup sh mqbroker -c ../conf/2m-noslave/broker-a.properties > /var/log/mq.log 2>&1 &

### 查看启动日志 ###

tail -f /var/log/ns.log  
tail -f /var/log/mq.log

### 查看启动端口 ###

netstat -tunpl  
jps

### 关闭RocketMQ ###

sh mqshutdown  
sh mqshutdown broker  
sh mqshutdown namesrv

### 再次启动 ###

nohup sh mqnamesrv >/var/log/ns.log 2>&1 &  
nohup sh mqbroker -c ../conf/2m-noslave/broker-a.properties > /var/log/mq.log 2>&1 &

[SouthEast]: /images/20220505/d6f8f5922fc447fcae16eae59f1f57e9.png
[SouthEast 1]: /images/20220505/28e1ecaa00b945cc867b392ae67030b9.png
[http_www.linuxidc.com_Linux_2016-09_134941.htm]: http://www.linuxidc.com/Linux/2016-09/134941.htm
[http_www.linuxidc.com_Linux_2015-10_124112.htm]: http://www.linuxidc.com/Linux/2015-10/124112.htm