rocketmq入门学习

Posted 2022-03-11 幸福丶如此

官方地址：https://rocketmq.apache.org/
安装包下载地址：https://archive.apache.org/dist/rocketmq/

一、rocketmq简介

RcoketMQ 是一款低延迟、高可靠、可伸缩、易于使用的消息中间件。具有以下特性：

支持发布/订阅（Pub/Sub）和点对点（P2P）消息模型
在一个队列中可靠的先进先出（FIFO）和严格的顺序传递
支持拉（pull）和推（push）两种消息模式
单一队列百万消息的堆积能力
支持多种消息协议，如 JMS、MQTT 等
分布式高可用的部署架构,满足至少一次消息传递语义
提供 docker 镜像用于隔离测试和云集群部署
提供配置、指标和监控等功能丰富的 Dashboard(rocketmq-console-ng)

二、rocketmq架构

由这张图可以看到有四个集群，分别是 NameServer 集群、Broker 集群、Producer 集群和 Consumer 集群：
1、NameServer: 提供轻量级的服务发现和路由。 每个 NameServer 记录完整的路由信息，提供等效的读写服务，并支持快速存储扩展。
2、Broker: 通过提供轻量级的 Topic 和 Queue 机制来处理消息存储,同时支持推（push）和拉（pull）模式以及主从结构的容错机制。
3、Producer：生产者，产生消息的实例，拥有相同 Producer Group 的 Producer 组成一个集群。
4、Consumer：消费者，接收消息进行消费的实例，拥有相同 Consumer Group 的Consumer 组成一个集群。

三、rocketmq部署模式

1、单 master 模式
只有一个 master 节点，一旦这个 master 节点宕机，那么整个服务就不可用，适合个人学习使用。
2、多 master 模式
多个 master 节点组成集群，单个 master 节点宕机或者重启对应用没有影响。
优点： 所有模式中性能最高
缺点： 单个 master 节点宕机期间，未被消费的消息在节点恢复之前不可用，消息的实时性就受到影响。
注意： 使用同步刷盘可以保证消息不丢失，同时 Topic 相对应的 queue 应该分布在集群中各个节点，而不是只在某各节点上，否则，该节点宕机会对订阅该 topic 的应用造成影响。
3、多 master 多 slave 异步复制模式
在多 master 模式的基础上，每个 master 节点都有至少一个对应的 slave。master
节点可读可写，但是 slave 只能读不能写，类似于 mysql 的主备模式。
优点： 在 master 宕机时，消费者可以从 slave 读取消息，消息的实时性不会受影响，性能几乎和多 master 一样。
缺点： 使用异步复制的同步方式有可能会有消息丢失的问题。
4、多 master 多 slave 同步双写模式
同多 master 多 slave 异步复制模式类似，区别在于 master 和 slave 之间的数据同步方式。
优点： 同步双写的同步模式能保证数据不丢失。
缺点： 发送单个消息 RT 会略长，性能相比异步复制低10%左右。
刷盘策略： 同步刷盘和异步刷盘（指的是节点自身数据是同步还是异步存储）
同步方式： 同步双写和异步复制（指的一组 master 和 slave 之间数据的同步）
注意： 要保证数据可靠，需采用同步刷盘和同步双写的方式，但性能会较其他方式低。

四、单 master 模式部署

1、下载二进制安装包

> wget https://archive.apache.org/dist/rocketmq/4.9.3/rocketmq-all-4.9.3-bin-release.zip
> unzip rocketmq-all-4.9.3-bin-release.zip

扩展：
如果下载 https://archive.apache.org/dist/rocketmq/4.9.3/rocketmq-all-4.9.3-source-release.zip
解压后需要使用 mvn -Prelease-all -DskipTests clean install -U 命令编译，编译好以后的目录 distribution/target/rocketmq-4.9.3/rocketmq-4.9.3 就可以使用了

2、修改 NameServer 的启动内存

修改这两个文件配置
> bin/runserver.sh
将 JAVA_OPT="$JAVA_OPT -server -Xms4g -Xmx4g -Xmn2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
改成 JAVA_OPT="$JAVA_OPT -server -Xms512m -Xmx512m -Xmn256m"

> bin/runbroker.sh
将 JAVA_OPT="$JAVA_OPT -server -Xms8g -Xmx8g"
改成 JAVA_OPT="$JAVA_OPT -server -Xms512m -Xmx512m"

3、启动和关闭sever

###启动server
> nohup sh bin/mqnamesrv &
> tail -f ~/logs/rocketmqlogs/namesrv.log
The Name Server boot success...

###关闭server
> sh bin/mqshutdown namesrv

4、启动和关闭broker

###启动broker
> nohup sh bin/mqbroker -n localhost:9876 &
> tail -f ~/logs/rocketmqlogs/broker.log 
The broker[%s, 172.30.30.233:10911] boot success...

###关闭broker
> sh bin/mqshutdown broker

5、多网卡情况下指定IP启动broker

###设置broker的IP地址
> echo "brokerIP1=192.168.16.239" > conf/broker.properties

### -c 参数指定配置文件
> nohup sh bin/mqbroker -n 192.168.16.239:9876 autoCreateTopicEnable=true -c conf/broker.properties & 

6、创建topic

> sh bin/mqadmin updateTopic -n localhost:9876 -b localhost:10911 -t demo

-b 指定Broker

7、发送和接收消息

###发送消息
> 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...

五、可视化监控平台 rocketmq-console-ng

项目地址：https://github.com/apache/rocketmq-externals

### 下载编译
> git clone -b release-rocketmq-console-1.0.0 https://github.com/apache/rocketmq-externals.git 
或者
> git clone -b mqtt-mesh https://github.com/apache/rocketmq-externals.git
> cd rocketmq-externals/rocketmq-console/
> mvn clean install -Dmaven.test.skip
> ls target/rocketmq-console-ng-*.jar
target/rocketmq-console-ng-2.0.0.jar


### 启动ng，这里指定了启动端口
nohup java -jar -Xms512m -Xmx512m rocketmq-console-ng-2.0.0.jar --server.port=8090 & 


### 访问
http://192.168.16.239:8090