RocketMQ
- 编译安装
- HelloWorld
官方网站
GitHub
https://github.com/apache/rocketmq
Quick Start
Linux下使用Maven编译源码安装
Rocketmq4.6+需要jdk1.8环境编译和运行
各版本要求
Version | Client | Broker | NameServer |
---|---|---|---|
4.0.0-incubating | >=1.7 | >=1.8 | >=1.8 |
4.1.0-incubating | >=1.6 | >=1.8 | >=1.8 |
4.2.0 | >=1.6 | >=1.8 | >=1.8 |
4.3.x | >=1.6 | >=1.8 | >=1.8 |
4.4.x | >=1.6 | >=1.8 | >=1.8 |
4.5.x | >=1.6 | >=1.8 | >=1.8 |
4.6.x | >=1.6 | >=1.8 | >=1.8 |
1.从GitHub上下载源码并上传到服务器
2.在Linux上安装Maven
下载Maven
wget https://mirrors.tuna.tsinghua.edu.cn/apache/maven/maven-3/3.6.3/binaries/apache-maven-3.6.3-bin.tar.gz
添加阿里云镜像
修改maven/conf
目录下的settings.xml
在mirrors
节点下添加
<mirror>
<id>aliyun-maven</id>
<mirrorOf>*</mirrorOf>
<name>aliyun maven</name>
<url>http://maven.aliyun.com/nexus/content/groups/public</url>
</mirror>
配置maven环境变量
修改/etc/profile
export M2_HOME=/usr/local/maven
export PATH=$PATH:$M2_HOME/bin
配置java环境变量
export JAVA_HOME="/usr/java/jdk1.8.0_181-amd64"
export CLASS_PATH="$JAVA_HOME/lib"
export PATH=".$PATH:$JAVA_HOME/bin"
环境变量修完执行source /etc/profile
立即生效
进入rocketmq主目录编译项目
mvn -Prelease-all -DskipTests clean install -U
3.启动nameserver
bin
目录下执行
./mqnamesrv
正常提示
4.启动Broker
./mqbroker -n localhost:9876
正常提示
5.测试消息发送
使用tool.sh
脚本执行测试程序
在bin
目录下执行
./tools.sh org.apache.rocketmq.example.quickstart.Producer
提示如下表示成功
6.接受消息
./tools.sh org.apache.rocketmq.example.quickstart.Consumer
控制台rocketmq-console编译安装
下载
https://github.com/apache/rocketmq-externals
中文指南
https://github.com/apache/rocketmq-externals/blob/master/rocketmq-console/doc/1_0_0/UserGuide_CN.md
上传到服务器并解压缩
编译
进入rocketmq-console
目录
执行编译
mvn clean package -Dmaven.test.skip=true
启动
编译成功后在rocketmq-console/target
目录下执行rocketmq-console-ng-1.0.1.jar
启动时,直接动态添加nameserver
地址或编辑application.properties
添加属性
java -jar rocketmq-console-ng-1.0.1.jar --rocketmq.config.namesrvAddr=127.0.0.1:9876
启动成功后访问服务器8080端口即可
pom.xml依赖
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.6.1</version>
</dependency>
安装启动常见错误
编译时包无法在mirror上找到 提示502错误
原因:网络不好或maven仓库服务器出错
重试即可,或者欢迎镜像仓库
发送失败提示connect to null failed
./tools.sh org.apache.rocketmq.example.quickstart.Producer
22:49:02.470 [main] DEBUG i.n.u.i.l.InternalLoggerFactory - Using SLF4J as the default logging framework
RocketMQLog:WARN No appenders could be found for logger (io.netty.util.internal.PlatformDependent0).
RocketMQLog:WARN Please initialize the logger system properly.
java.lang.IllegalStateException: org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to
null failed
原因:不知道nameserver
在哪儿
在tools
脚本中添加
export NAMESRV_ADDR=localhost:9876
启动broker失败 Cannot allocate memory
原因:jvm启动初始化内存分配大于物理内存
[root@node-113b bin]# ./mqbroker -n localhost:9876
Java HotSpot(TM) 64-Bit Server VM warning: INFO: os::commit_memory(0x00000005c0000000, 8589934592, 0) failed
; error='Cannot allocate memory' (errno=12)#
# There is insufficient memory for the Java Runtime Environment to continue.
# Native memory allocation (mmap) failed to map 8589934592 bytes for committing reserved memory.
# An error report file with more information is saved as:
# /usr/local/rocketmq/bin/hs_err_pid1997.log
修改启动脚本中的jvm参数
runbroker.sh
broker
runserver.sh
nameserver
默认数值给的都很大,改小即可
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=64m -XX:MaxMetaspaceSize=100m"
启动broker成功但提示:Failed to obtain the host name
[root@node-113b bin]# ./mqbroker -n localhost:9876
22:30:42.307 [main] ERROR RocketmqCommon - Failed to obtain the host name
java.net.UnknownHostException: node-113b: node-113b: No address associated with hostname
at java.net.InetAddress.getLocalHost(InetAddress.java:1505) ~[na:1.8.0_181]
at org.apache.rocketmq.common.BrokerConfig.localHostName(BrokerConfig.java:189) [rocketmq-common-4.6
.1.jar:4.6.1] at org.apache.rocketmq.common.BrokerConfig.<init>(BrokerConfig.java:38) [rocketmq-common-4.6.1.jar:4
.6.1] at org.apache.rocketmq.broker.BrokerStartup.createBrokerController(BrokerStartup.java:110) [rocketmq
-broker-4.6.1.jar:4.6.1] at org.apache.rocketmq.broker.BrokerStartup.main(BrokerStartup.java:58) [rocketmq-broker-4.6.1.jar:4
.6.1]Caused by: java.net.UnknownHostException: node-113b: No address associated with hostname
at java.net.Inet6AddressImpl.lookupAllHostAddr(Native Method) ~[na:1.8.0_181]
at java.net.InetAddress$2.lookupAllHostAddr(InetAddress.java:928) ~[na:1.8.0_181]
at java.net.InetAddress.getAddressesFromNameService(InetAddress.java:1323) ~[na:1.8.0_181]
at java.net.InetAddress.getLocalHost(InetAddress.java:1500) ~[na:1.8.0_181]
... 4 common frames omitted
The broker[DEFAULT_BROKER, 192.168.150.213:10911] boot success. serializeType=JSON and name server is localh
ost:9876
原因:无法解析当前的主机名
hosts里添加映射即可
192.168.150.213 node-113b
linux日期校准
安装ntpdate
yum install ntpdate
ntpdate ntp1.aliyun.com
RocketMQ 功能 大纲
主流的MQ有很多,比如ActiveMQ、RabbitMQ、RocketMQ、Kafka、ZeroMQ等。
之前阿里巴巴也是使用ActiveMQ,随着业务发展,ActiveMQ IO 模块出现瓶颈,后来阿里巴巴 通过一系列优化但是还是不能很好的解决,之后阿里巴巴把注意力放到了主流消息中间件kafka上面,但是kafka并不能满足他们的要求,尤其是低延迟和高可靠性。
所以RocketMQ是站在巨人的肩膀上(kafka)MetaQ的内核,又对其进行了优化让其更满足互联网公司的特点。它是纯Java开发,具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。 RocketMQ目前在阿里集团被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景。
RocketMQ介绍
- 消息队列应用场景
- rocketmq中的角色
- nameserver
- 官网解读
消息发送
-
同步发送
-
异步发送
-
单向发送
-
消息批量发送
-
消息结构
-
消息发送流程
消息存储
- 存储方式
- 发送消息时存储流程
- 存储文件与内存映射
- 刷盘机制
- 文件恢复与过期删除机制
- 索引
消息消费
- 消息订阅
- 消息拉取和推送
- 消息处理队列
- 顺序消费
- 定时消息机制
- 消息过滤TAG与sql92
- FilterServer过滤机制
- 并发消息消费
- 消费负载与算法
- 消费者动态添加
- 消息消费过程
- ACK
- 消费进度与offset
Rocketmq集群 HA
- 集群搭建
- 主从同步复制原理
- 读写分离机制
整合
- 使用spring
- SpringCloud整合
监控与运维
- rocketmq-console监控平台
- 命令行运维 MQAdmin
- 自定义日志
消息队列介绍
消息队列是《数据结构》中先进先出的一种数据结构,在当前的架构中,作为中间件提供服务。
消息中间件功能
应用解耦
AB应用不在互相依赖
流量削峰
流量达到高峰的时候,通常使用限流算法来控制流量涌入系统,避免系统被击瘫,但是这种方式损失了一部分请求
此时可以使用消息中间件来缓冲大量的请求,匀速消费,当消息队列中堆积消息过多时,我们可以动态上线增加消费端,来保证不丢失重要请求。
大数据处理
消息中间件可以把各个模块中产生的管理员操作日志、用户行为、系统状态等数据文件作为消息收集到主题中
数据使用方可以订阅自己感兴趣的数据内容互不影响,进行消费
异构系统
跨语言
RocketMQ 角色
broker
- Broker面向producer和consumer接受和发送消息
- 向nameserver提交自己的信息
- 是消息中间件的消息存储、转发服务器。
- 每个Broker节点,在启动时,都会遍历NameServer列表,与每个NameServer建立长连接,注册自己的信息,之后定时上报。
broker集群
- Broker高可用,可以配成Master/Slave结构,Master可写可读,Slave只可以读,Master将写入的数据同步给Slave。
- 一个Master可以对应多个Slave,但是一个Slave只能对应一个Master
- Master与Slave的对应关系通过指定相同的BrokerName,不同的BrokerId来定义BrokerId为0表示Master,非0表示Slave
- Master多机负载,可以部署多个broker
- 每个Broker与nameserver集群中的所有节点建立长连接,定时注册Topic信息到所有nameserver。
producer
- 消息的生产者
- 通过集群中的其中一个节点(随机选择)建立长连接,获得Topic的路由信息,包括Topic下面有哪些Queue,这些Queue分布在哪些Broker上等
- 接下来向提供Topic服务的Master建立长连接,且定时向Master发送心跳
consumer
消息的消费者,通过NameServer集群获得Topic的路由信息,连接到对应的Broker上消费消息。
注意,由于Master和Slave都可以读取消息,因此Consumer会与Master和Slave都建立连接。
nameserver
底层由netty实现,提供了路由管理、服务注册、服务发现的功能,是一个无状态节点
nameserver是服务发现者,集群中各个角色(producer、broker、consumer等)都需要定时想nameserver上报自己的状态,以便互相发现彼此,超时不上报的话,nameserver会把它从列表中剔除
nameserver可以部署多个,当多个nameserver存在的时候,其他角色同时向他们上报信息,以保证高可用,
NameServer集群间互不通信,没有主备的概念
nameserver内存式存储,nameserver中的broker、topic等信息默认不会持久化
为什么不用zookeeper?:rocketmq希望为了提高性能,CAP定理,客户端负载均衡
对比JSM中的Topic和Queue
Topic是一个逻辑上的概念,实际上Message是在每个Broker上以Queue的形式记录。
对应到JMS中的topic实现是由客户端来完成的
consumer.setMessageModel(MessageModel.BROADCASTING);