RocketMQ安装部署

Stella981
• 阅读 289

一、简介RocketMQ

RocektMQ是阿里巴巴在2012年开源的一个纯java、分布式、队列模型的第三代消息中间件,不仅在传统高频交易链路有着低延迟的出色表现,在实时计算等大数据领域也有着不错的吞吐。

2016年11月11号,双十一大促见证了RocketMQ低延迟存储架构的成功试水,99.996%的延迟落在了10ms以内,极个别由于GC引发的停顿在50ms以内,其高性能、低延时和高可靠的特性承载了近年来双十一17万笔/秒的交易峰值,在整个生产链路上都有着稳定和出色的表现。其在同年捐赠给Apache后正式进入孵化期。并于2017年9月RocketMQ正式从Apache社区正式毕业,成为Apache顶级项目。

二、相关术语

1. Producer

消息生产者,负责产生消息,一般由业务系统负责产生消息。

Producer Group:一类Producer的集合名称,这类Producer通常发送一类消息,且发送逻辑一致。

2. Consumer

消息消费者,负责消费消息,一般是后台系统负责异步消费。

Push Consumer:Consumer的一种,应用通常向Consumer对象注册一个Listener接口,一旦收到消息,Consumer对象立刻回调Listener接口方法。

Pull Consumer:Consumer的一种,应用通常主动调用Consumer的拉消息方法从Broker拉消息,主动权由应用控制。

Consumer Group:一类Consumer的集合名称,这类Consumer通常消费一类消息,且消费逻辑一致。

3. Broker

消息中转角色, 负责存储消息,转发消息,一般也称为Server。

Master:Broker中的主节点。

Slave:Broker中的副节点。

4. Nameserver

专为RocketMQ设计的 轻量级名称服务。集群中Nameserver互相独立,彼此没有通信关系,单台Nameserver挂掉,不影响其他Nameserver,即使全部挂掉,也不影响业务系统使用。而且Nameserver不会有频繁的读写,所以性能开销非常小,稳定性很高。

5. 广播消费

一条消息被多个Consumer消费,即使这些Consumer属于同一个Consumer Group,消息也会被Consumer Group中的每个Consumer都消费一次,广播消费中的Consumer Group概念可以认为在消息划分方面无意义。

6. 集群消费

一个Consumer Group中的Consumer实例平均分摊消费消息。例如某个Topic有9条消息,其中一个Consumer Group有3个实例(可能是3个进程,或者3台机器),那么每个实例只消费其中的3条消息。

7. 顺序消息

消费消息的顺序要同发送消息的顺序一致,在RocketMQ中,主要指的是局部顺序,即一类消息为满足顺序性,必须Producer单线程顺序发送,且发送到同一个队列,这样Consumer就可以按照Producer发送的顺序去消费消息。

8. 普通顺序消息

顺序消息的一种,正常情况下可以保证完全的顺序消息,但是一旦发生通信异常,Broker重启,由于队列总数发生变化,哈希取模后定位的队列会变化,产生短暂的消息顺序不一致。如果业务能容忍在集群异常情况(如某个Broker宕机或者重启)下,消息短暂的乱序,使用普通顺序方式比较合适。

9. 严格顺序消息

顺序消息的一种,无论正常异常情况都能保证顺序,但是牺牲了分布式Failover特性,即Broker集群中只要有一台机器不可用,则整个集群都不可用,服务可用性大大降低。如果服务器部署为同步双写模式,此缺陷可通过备机自动切换为主避免,不过仍然会存在几分钟的服务不可用。

10. Message Queue

在RocketMQ中,所有消息队列都是持久化,长度无限的数据结构,所谓长度无限是指队列中的每个存储单元都是定长,访问其中的存储单元使用Offset来访问,offset为java long类型,64位,理论上在100年内不会溢出,所以认为是长度无限,另外队列中只保存最近几天的数据,之前的数据会按照过期时间来删除。也可以认为Message Queue是一个长度无限的数组,offset就是下标。

11. 异步复制

消息写入master节点,再由master节点异步复制到slave节点,类似mysql中的master-slave机制。

12. 同步双写

消息同时写入master节点和slave节点。

13. 异步刷盘

Broker的一种持久化策略,消息写入pagecache后,直接返回。由异步线程负责将pagecache写入硬盘。

14. 同步刷盘

Broker的一种持久化策略,消息写入pagecache后,由同步线程将pagecache写入硬盘后,再返回。

15.TPS--每秒发送消息个数

三、RocketMQ集群部署模式

RocketMQ作为消息中间件,其主要功能为消息的Publish/Subscribe。而Broker担任的消息转发和存储功能,其部署方式有很多种:

1. 单Master

优点:除了配置简单没什么优点。

缺点:不可靠,该机器重启或宕机,将导致整个服务不可用。

2. 多Master

优点:配置简单,性能最高。

缺点:可能会有少量消息丢失,单台机器重启或宕机期间,该机器下未被消费的消息在机器恢复前不可订阅,影响消息实时性。

3. 异步多Master多Slave

每个Master配一个Slave,有多对Master-Slave,集群采用 异步复制方式,主备有短暂消息延迟,毫秒级。

优点:性能同多Master几乎一样,实时性高,主备间切换对应用透明,不需人工干预。

缺点:Master宕机或磁盘损坏时会有少量消息丢失。

4. 同步多Master多Slave

每个Master配一个Slave,有多对Master-Slave,集群采用 同步双写方式,主备都写成功,向应用返回成功。

优点:服务可用性与数据可用性非常高。

缺点:性能比异步集群略低,当前版本主宕备不能自动切换为主。

四、RocketMQ集群特性分析

RocketMQ部署架构图如下所示:

RocketMQ安装部署

Nameserver

Nameserver的开发旨在轻量级,多台Nameserver互相独立,彼此间互不通信,这样的设计,保证了单台Nameserver宕机,不影响Nameserver。nameserver不会有频繁的读写,所以性能开销非常小,稳定性很高。

Broker与Nameserver关系

连接:

每个Broker与系统中所有的Nameserver保持长连接。

心跳间隔:

每隔30秒(此时间无法更改)向所有Nameserver发送心跳,心跳包含了自身的topic配置信息。

心跳超时:

Nameserver每隔10秒钟(此时间无法更改),扫描所有还存活的Broker连接,若某个连接2分钟内(当前时间与最后更新时间差值超过2分钟,此时间无法更改)没有发送心跳数据,则断开连接。

断开:

一旦连接断开,Nameserver会立即更新topic与队列的对应关系,但不会通知生产者和消费者。

负载均衡:

一个topic分布在多个Broker上,一个broker可以配置多个topic,它们是多对多的关系。如果某个topic消息量很大,应该给它多配置几个队列,并且尽量多分布在不同broker上,减轻某个broker的压力。topic消息量都比较均匀的情况下,如果某个broker上的队列越多,则该broker压力越大。

可用性:

由于消息分布在各个broker上,一旦某个broker宕机,则该broker上的消息读写都会受到影响。所以rocketmq提供了master/slave的结构,salve定时从master同步数据,如果master宕机,则slave提供消费服务,但是不能写入消息,此过程对应用透明,由rocketmq内部解决。

两个关键点:

1. 一旦某个broker master宕机,生产者和消费者多久才能发现?受限于rocketmq的网络连接机制,默认情况下,最多需要30秒,但这个时间可由应用设定参数来缩短时间。这个时间段内,发往该broker的消息都是失败的,而且该broker的消息无法消费,因为此时消费者不知道该broker已经挂掉。

2. 消费者得到master宕机通知后,转向slave消费,但是slave不能保证master的消息100%都同步过来了,因此会有少量的消息丢失。但是消息最终不会丢的,一旦master恢复,未同步过去的消息会被消费掉。

可靠性:

所有发往broker的消息,有同步刷盘和异步刷盘机制,总的来说,可靠性非常高。

同步刷盘时,消息写入物理文件才会返回成功,因此非常可靠。

异步刷盘时,只有机器宕机,才会产生消息丢失,broker挂掉可能会发生,但是机器宕机崩溃是很少发生的,除非突然断电。

Consumer与Nameserver关系

连接:

单个消费者和一台nameserver保持长连接。定时查询topic配置信息,如果该nameserver挂掉,消费者会自动连接下一个nameserver,直到有可用连接为止,并能自动重连。

心跳:

与nameserver没有心跳。

轮询时间:

默认情况下,消费者每隔30秒从nameserver获取所有topic的最新队列情况,这意味着某个broker如果宕机,客户端最多要30秒才能感知。该时间由DefaultMQPushConsumer的pollNameServerInteval参数决定,可手动配置。

Consumer与Broker关系

连接:

单个消费者和该消费者关联的所有broker保持长连接。

心跳:

默认情况下,消费者每隔30秒向所有broker发送心跳,该时间由DefaultMQPushConsumer的heartbeatBrokerInterval参数决定,可手动配置。broker每隔10秒钟(此时间无法更改),扫描所有还存活的连接,若某个连接2分钟内(当前时间与最后更新时间差值超过2分钟,此时间无法更改)没有发送心跳数据,则关闭连接,并向该消费者分组的所有消费者发出通知,分组内消费者重新分配队列继续消费。

断开:

消费者挂掉;心跳超时导致broker主动关闭连接。

动作:

一旦连接断开,broker会立即感知到,并向该消费者分组的所有消费者发出通知,分组内消费者重新分配队列继续消费。

负载均衡:

集群消费模式下,一个消费者集群多台机器共同消费一个topic的多个队列,一个队列只会被一个消费者消费。如果某个消费者挂掉,分组内其它消费者会接替挂掉的消费者继续消费。

消费机制

本地队列:

消费者不间断的从broker拉取消息,消息拉取到本地队列,然后本地消费线程消费本地消息队列,只是一个异步过程,拉取线程不会等待本地消费线程,这种模式实时性非常高。对消费者对本地队列有一个保护,因此本地消息队列不能无限大,否则可能会占用大量内存,本地队列大小由DefaultMQPushConsumer的pullThresholdForQueue属性控制,默认1000,可手动设置。

轮询间隔:

消息拉取线程拉取间隔时间由DefaultMQPushConsumer的pullInterval属性控制,默认为0,可手动设置。

消息消费数量:

监听器每次接受本地队列的消息数量是由参数DefaultMQPushConsumer的consumeMessageBatchMaxSize属性控制,默认为1,可手动设置。

消费进度存储:

每隔一段时间将各个队列的消费进度存储到对应的broker上,该时间由DefaultMQPushConsumer的persistConsumerOffsetInterval属性控制,默认为5秒,可手动设置。

Producer与Nameserver关系

连接:

单个生产者者和一台nameserver保持长连接,定时查询topic配置信息,如果该nameserver挂掉,生产者会自动连接下一个nameserver,直到有可用连接为止,并能自动重连。

心跳:

与nameserver没有心跳。

Producer与与Broker关系

连接:

单个生产者和该生产者关联的所有broker保持长连接。

心跳:

默认情况下,生产者每隔30秒向所有broker发送心跳,该时间由DefaultMQProducer的heartbeatBrokerInterval参数决定,可手动配置。broker每隔10秒钟(此时间无法更改),扫描所有还存活的连接,若某个连接2分钟内(当前时间与最后更新时间差值超过2分钟,此时间无法更改)没有发送心跳数据,则关闭连接。

断开:

移除Broker上的生产者信息。

负载均衡:

生产者时间没有关系,每个生产者向队列轮流发送消息。

以上论述了RocketMQ各组件特性和关系,除此之外,在允许可自动创建Topic的配置下,Producer负责创建Topic消息和发送消息,发送消息支持三种方式,异步、同步和onway方式。就总体的消息传输层面来说,RocketMQ有集群模式和广播模式,默认是集群模式,集群模式以其原生的Consumer Group(消费组)实现了负载均衡,广播模式下,Consumer Group下的每个Consumer实例则要消费全部的消息数据。就业务适用层面来说,rocketMQ可以根据业务需求,实现Order Message(订单消息)、Broadcasting(广播消息)、Scheduled Message(延迟消息)、Batch(同Topic批量消息发送)、Filter(基本消息过滤和高级消息过滤)、OpenMessaging(流式消息发送)等。

五、安装RocketMQ

1、下载源码或二进制安装文件 http://rocketmq.apache.org/dowloading/releases/

> unzip rocketmq-all-4.4.0-source-release.zip
  > cd rocketmq-all-4.4.0/
  > mvn -Prelease-all -DskipTests clean install -U
  > cd distribution/target/rocketmq-4.4.0

2、启动NameServer,启动之前先根据实际资源情况,合理分配JVM大小 bin/runserver.sh 和 bin/runbroker.sh

RocketMQ的各日志文件默认存放在 ${user.home}/logs/rocketmqlogs,各数据文件默认存放在 ${user.home}/store

> nohup sh bin/mqnamesrv >/dev/null 2>&1 &
 > tail -f ~/logs/rocketmqlogs/namesrv.log
  The Name Server boot success...

3、启动Broker

> nohup sh bin/mqbroker -n localhost:9876 autoCreateTopicEnable=true >/dev/null 2>&1 &
 > tail -f ~/logs/rocketmqlogs/broker.log 
  The broker[%s, 172.30.30.233:10911] boot success...

4、测试发送和接受消息

在发送/接收消息之前,我们需要告诉客户名称服务器的位置。RocketMQ 提供了多种方法来实现这一点。为了简单起见,我们使用环境变量NAMESRV_ADDR

> export NAMESRV_ADDR=localhost:9876
 > sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
 SendResult [sendStatus=SEND_OK, msgId= ...

 > sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
 ConsumeMessageThread_%d Receive New Messages: [MessageExt...

5、关闭服务(先关闭broker再关闭namesrv)

> sh bin/mqshutdown broker
The mqbroker(36695) is running...
Send shutdown request to mqbroker(36695) OK

> sh bin/mqshutdown namesrv
The mqnamesrv(36664) is running...
Send shutdown request to mqnamesrv(36664) OK

六、RocketMQ监控

Apache版的RocketMQ管理界面部署工具可以从github上下载源码,地址:https://github.com/apache/rocketmq-externals 部署流程如下:

1、修改下载的配置文件,关联rocketMQ集群到管理界面。或者在启动服务的时候 手动指定启动参数

# git clone https://github.com/apache/rocketmq-externals# cd rocketmq-externals/rocketmq-console/src/main/resources

  # vim application.properties
      server.port=8888 #指定开启的端口号
      rocketmq.config.nameservAddr=172.16.1.1:9876,172.16.1.2:9876
      rocketmq.config.dataPath=/usr/local/rocketmq-4.5.1/store  #默认存放到 /tmp/rocketmq-console
      rocketmq.config.loginRequired=true #开启登录权限

  # vim user.properties #设置登录账号及密码
      wjoyxt=wjoyxt

  # vim ../../../pom.xml
      <rocketmq.version>4.5.1</rocketmq.version> #根据安装版本的不同修改版本号

2、编译rocketmq-console

# cd rocketmq-externals/rocketmq-console# mvn clean package -Dmaven.test.skip=true

成功后会在rocketmq-externals/rocketmq-console/target目录下产生一个rocketmq-console-ng-1.0.1.jar文件

3、启动rocketmq-console,其默认的访问端口是8080,多个nameserver时需要用 ; 号隔开并用引号引起来

# nohup java -jar /usr/local/rocketmq-4.5.1/prod-mq-console.jar --server.port=8888 --rocketmq.config.namesrvAddr='172.17.213.71:9876;172.17.213.72' >/dev/null 2>&1 &

七、常用命令

查看集群情况 ./mqadmin clusterList -n 127.0.0.1:9876
查看 broker 状态 ./mqadmin brokerStatus -n 127.0.0.1:9876 -b 172.20.1.138:10911 (注意换成你的 broker 地址)
查看 topic 列表 ./mqadmin topicList -n 127.0.0.1:9876
查看 topic 状态 ./mqadmin topicStatus -n 127.0.0.1:9876 -t MyTopic (换成你想查询的 topic)
查看 topic 路由 ./mqadmin topicRoute -n 127.0.0.1:9876 -t MyTopic

八、集群模式

 1、创建store存储目录

如果不采用默认的存储路径的话,需要在配置文件中指定,并手动提前创建好

mkdir -p /usr/local/rocketmq-4.5.1/store/index
mkdir /usr/local/rocketmq-4.5.1/store/commitlog
mkdir /usr/local/rocketmq-4.5.1/store/consumequeue

 2、根据实际情况编辑配置文件

# 所属集群名字
brokerClusterName=my-rocket-mq
# Broker 名字与IP
brokerName=broker-71
brokerIP1=172.17.213.71
# 0 代表 Master 大于0是slave
brokerId=0
# nameServer 分号分割
namesrvAddr=172.17.213.71:9876;172.17.213.72:9876
# 在发送消息时,自动创建服务器不存在的topic,默认创建的队列数
defaultTopicQueueNums=4
# 是否允许 Broker 自动创建Topic,建议线下开启,线上关闭
autoCreateTopicEnable=true
# 是否允许 Broker 自动创建订阅组,建议线下开启,线上关闭
autoCreateSubscriptionGroup=true
# Broker 对外服务的监听端口
listenPort=10911
# 删除文件时间点,默认凌晨 4点
deleteWhen=04
# 文件保留时间,默认 48 小时
fileReservedTime=120
# CommitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
# ConsumeQueue每个文件默认存30W条,根据业务情况调整
mapedFileSizeConsumeQueue=300000
#destroyMapedFileIntervalForcibly=120000
#redeleteHangedFileInterval=120000
# 检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
# store存储路径,master与slave目录要不同
storePathRootDir=/usr/local/rocketmq-4.5.1/store
# CommitLog 存储路径
storePathCommitLog=/usr/local/rocketmq-4.5.1/store/commitlog
# 消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq-4.5.1/store/consumequeue
# 消息索引存储路径
storePathIndex=/usr/local/rocketmq-4.5.1/store/index
# checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq-4.5.1/store/checkpoint
# abort 文件存储路径
abortFile=/usr/local/rocketmq-4.5.1/store/abort
# 限制的消息大小
maxMessageSize=65536
#flushCommitLogLeastPages=4
#flushConsumeQueueLeastPages=2
#flushCommitLogThoroughInterval=10000
#flushConsumeQueueThoroughInterval=60000
#checkTransactionMessageEnable=false
# 发消息线程池数量
#sendMessageThreadPoolNums=128
# 拉消息线程池数量
#pullMessageThreadPoolNums=128
# Broker 的角色
# - ASYNC_MASTER 异步复制Master
# - SYNC_MASTER 同步双写Master
# - SLAVE 从
brokerRole=ASYNC_MASTER
# 刷盘方式
# - ASYNC_FLUSH 异步刷盘
# - SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH

3、双主模式的启动 

nohup sh /usr/local/rocketmq-4.5.1/bin/mqnamesrv >/dev/null 2>&1 &
nohup sh /usr/local/rocketmq-4.5.1/bin/mqbroker -c /usr/local/rocketmq-4.5.1/conf/2m-noslave/broker-a.properties >/dev/null 2>&1 & 

nohup sh /usr/local/rocketmq-4.5.1/bin/mqnamesrv >/dev/null 2>&1 &
nohup sh /usr/local/rocketmq-4.5.1/bin/mqbroker -c /usr/local/rocketmq-4.5.1/conf/2m-noslave/broker-b.properties >/dev/null 2>&1 &
点赞
收藏
评论区
推荐文章
刚刚好 刚刚好
2个月前
css问题
1、 在IOS中图片不显示(给图片加了圆角或者img没有父级) <div<img src""/</div div {width: 20px; height: 20px; borderradius: 20px; overflow: h
blmius blmius
1年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录 问题 用navicat导入数据时,报错: 原因这是因为当前的MySQL不支持datetime为0的情况。 解决修改sql\mode: sql\mode:SQL Mode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。 全局s
晴空闲云 晴空闲云
2个月前
css中box-sizing解放盒子实际宽高计算
我们知道传统的盒子模型,如果增加内边距padding和边框border,那么会撑大整个盒子,造成盒子的宽度不好计算,在实务中特别不方便。boxsizing可以设置盒模型的方式,可以很好的设置固定宽高的盒模型。 盒子宽高计算假如我们设置如下盒子:宽度和高度均为200px,那么这会这个盒子实际的宽高就都是200px。但是当我们设置这个盒子的边框和内间距的时候,那
艾木酱 艾木酱
1个月前
快速入门|使用MemFire Cloud构建React Native应用程序
> MemFire Cloud是一款提供云数据库,用户可以创建云数据库,并对数据库进行管理,还可以对数据库进行备份操作。它还提供后端即服务,用户可以在1分钟内新建一个应用,使用自动生成的API和SDK,访问云数据库、对象存储、用户认证与授权等功能,可专
Easter79 Easter79
1年前
swap空间的增减方法
(1)增大swap空间 去激活swap交换区: #swapoff -v /dev/vg00/lvswap 扩展交换lv: #lvextend -L 10G /dev/vg00/lvswap 重新生成swap交换区: #mkswap /dev/vg00/lvswap 激活新生成的交换区: #swapon -v /dev/vg00/lvswap
Wesley13 Wesley13
1年前
MySQL查询按照指定规则排序
1.按照指定(单个)字段排序 select * from table_name order id desc; 2.按照指定(多个)字段排序 select * from table_name order id desc,status desc; 3.按照指定字段和规则排序 selec
Stella981 Stella981
1年前
Angular material mat
Icon Icon Name mat-icon code _add\_comment_ add comment icon <mat-icon> add\_comment</mat-icon> _attach\_file_ attach file icon <mat-icon> attach\_file</mat-icon> _attach\
Wesley13 Wesley13
1年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
#### 背景描述 # Time: 2019-01-24T00:08:14.705724+08:00 # User@Host: **[**] @ [**] Id: ** # Schema: sentrymeta Last_errno: 0 Killed: 0 # Query_time: 0.315758 Lock_
helloworld_28799839 helloworld_28799839
2个月前
常用知识整理
# Javascript ## 判断对象是否为空 ```js Object.keys(myObject).length === 0 ``` ## 经常使用的三元运算 > 我们经常遇到处理表格列状态字段如 `status` 的时候可以用到 ``` vue
helloworld_34035044 helloworld_34035044
4个月前
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。 uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid() 或 uuid(sep)参数说明:sep 布尔值,生成的uuid中是否包含分隔符'',缺省为