伸手摘星丶
今でもあなたは私の光丶
伸手摘星丶
rocketMQ 消息中间件

RocketMQ(1)RocketMQ架构与实战

来自 |已发表

RocketMQ是阿里巴巴中间件团队自研的一款高性能、高吞吐量、低延迟、高可用、高可靠（具备 金融级稳定性）的分布式消息中间件，开源后并于2016年捐赠给Apache社区孵化，目前已经成为了 Apache顶级项目。当前在国内被广泛的使用，包括互联网、电商、金融、企业服务等领域，包括：字 节跳动、滴滴、微众银行等知名的互联网公司。

RocketMQ的前世今生

RocketMQ在阿里内部叫做Metaq（最早名为Metamorphosis，中文意思“变形记”，是作家卡夫卡 的中篇小说代表作，可见是为了致敬Kafka）。

RocketMQ是Metaq 3.0之后开源的版本。

Metaq在阿里巴巴集团内部、蚂蚁金服、菜鸟等各业务中被广泛使用，接入了上万个应用系统中。 并平稳支撑了历年的双十一大促（万亿级的消息），在性能、稳定性、可靠性等方面表现出色，在整个 阿里技术体系和大中台战略中发挥着举足轻重的作用。

Metaq最早源于Kafka，早期借鉴了Kafka很多优秀的设计。但是由于Kafka是Scale语言编写而阿里 系主要使用Java，且无法满足阿里的电商、金融业务场景，所以誓嘉（花名）团队用Java重新造轮子， 并做了大量的改造和优化。

在此之前，淘宝有一款消息中间件名为Notify，目前已经逐步被Metaq所取代。

第一代的Notify主要使用了推模型，解决了事务消息；第二代的MetaQ主要使用了拉模型，解决了 顺序消息和海量堆积的问题。相比起Kafka使用的Scale语言编写，RabbitMQ 使用Erlang语言编写，基 于Java的RocketMQ开源后更容易被广泛的研究，以及其他大厂定制开发。

RocketMQ的使用场景

  • 应用解耦
    系统的耦合性越高，容错性就越低。以电商应用为例，用户创建订单后，如果耦合调用库存系统、 物流系统、支付系统，任何一个子系统出了故障或者因为升级等原因暂时不可用，都会造成下单操作异 常，影响用户使用体验。
  • 流量削峰
    应用系统如果遇到系统请求流量的瞬间猛增，有可能会将系统压垮。有了消息队列可以将大量请求 缓存起来，分散到很长一段时间处理，这样可以大大提到系统的稳定性和用户体验。
    举例:业务系统正常时段的QPS如果是1000，流量最高峰是10000，为了应对流量高峰配置高性能 的服务器显然不划算，这时可以使用消息队列对峰值流量削峰
  • 数据分发
    通过消息队列可以让数据在多个系统之间进行流通。数据的产生方不需要关心谁来使用数据，只需 要将数据发送到消息队列，数据使用方直接在消息队列中直接获取数据即可

RocketMQ 部署架构

RocketMQ的角色介绍

  • Producer：消息的发送者；举例：发信者
  • Consumer：消息接收者；举例：收信者
  • Broker：暂存和传输消息；举例：邮局
  • NameServer：管理Broker；举例：各个邮局的管理机构
  • Topic：区分消息的种类；一个发送者可以发送消息给一个或者多个Topic；一个消息的接收者 可以订阅一个或者多个Topic消息
  • Message Queue：相当于是Topic的分区；用于并行发送和接收消息
  • NameServer是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。
  • Broker部署相对复杂，Broker分为Master与Slave，一个Master可以对应多个Slave，但是一 个Slave只能对应一个Master，Master与Slave 的对应关系通过指定相同的BrokerName，不 同的BrokerId来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。 每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有 NameServer。 注意：当前RocketMQ版本在部署架构上支持一Master多Slave，但只有 BrokerId=1的从服务器才会参与消息的读负载。
  • Producer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从 NameServer获取Topic路由信息，并向提供Topic 服务的Master建立长连接，且定时向 Master发送心跳。Producer完全无状态，可集群部署。
  • Consumer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从 NameServer获取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时 向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消 息，消费者在向Master拉取消息时，Master服务器会根据拉取偏移量与最大偏移量的距离 （判断是否读老消息，产生读I/O），以及从服务器是否可读等因素建议下一次是从Master还 是Slave拉取。

执行流程:

  1. 启动NameServer，NameServer起来后监听端口，等待Broker、Producer、Consumer连上 来，相当于一个路由控制中心。
  2. Broker启动，跟所有的NameServer保持长连接，定时发送心跳包。心跳包中包含当前 Broker信息(IP+端口等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic 跟Broker的映射关系。
  3. 收发消息前，先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，也可以在 发送消息时自动创建Topic。
  4. Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从 NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列， 然后与队列所在的Broker建立长连接从而向Broker发消息。
  5. Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在 哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息

RocketMQ特性

订阅与发布

消息的发布是指某个生产者向某个topic发送消息；消息的订阅是指某个消费者关注了某个topic中 带有某些tag的消息。

消息顺序

消息有序指的是一类消息消费时，能按照发送的顺序来消费。例如：一个订单产生了三条消息分别 是订单创建、订单付款、订单完成。消费时要按照这个顺序消费才能有意义，但是同时订单之间是可以 并行消费的。RocketMQ可以严格的保证消息有序。

消息过滤

RocketMQ的消费者可以根据Tag进行消息过滤，也支持自定义属性过滤。消息过滤目前是在 Broker端实现的，优点是减少了对于Consumer无用消息的网络传输，缺点是增加了Broker的负担、而 且实现相对复杂。

消息可靠性

RocketMQ支持消息的高可靠，影响消息可靠性的几种情况：

  1. Broker非正常关闭
  2. Broker异常 Crash
  3. OS Crash
  4. 机器掉电，但是能立即恢复供电情况
  5. 机器无法开机（可能是cpu、主板、内存等 关键设备损坏）
  6. 磁盘设备损坏

1)、2)、3)、4) 四种情况都属于硬件资源可立即恢复情况，RocketMQ在这四种情况下能保证消息 不丢，或者丢失少量数据（依赖刷盘方式是同步还是异步）。

5)、6)属于单点故障，且无法恢复，一旦发生，在此单点上的消息全部丢失。

RocketMQ在这两种情况下，通过异步复制，可保证99%的消息不丢，但是仍然会有极少量的消息 可能丢失。
通过同步双写技术可以完全避免单点，同步双写势必会影响性能，适合对消息可靠性要求极高的场 合，例如与Money相关的应用。注：RocketMQ从3.0版本开始支持同步双写。

至少一次

至少一次(At least Once)指每个消息必须投递一次。Consumer先Pull消息到本地，消费完成后，才 向服务器返回ack，如果没有消费一定不会ack消息，所以RocketMQ可以很好的支持此特性。

回溯消费

回溯消费是指Consumer已经消费成功的消息，由于业务上需求需要重新消费，要支持此功能， Broker在向Consumer投递成功消息后，消息仍然需要保留。并且重新消费一般是按照时间维度，例如 由于Consumer系统故障，恢复后需要重新消费1小时前的数据，那么Broker要提供一种机制，可以按 照时间维度来回退消费进度。RocketMQ支持按照时间回溯消费，时间维度精确到毫秒。

事务消息

RocketMQ事务消息（Transactional Message）是指应用本地事务和发送消息操作可以被定义到 全局事务中，要么同时成功，要么同时失败。

RocketMQ的事务消息提供类似 X/Open XA 的分布事务功能，通过事务消息能达到分布式事务的 最终一致性。

定时消息

定时消息（延迟队列）是指消息发送到broker后，不会立即被消费，等待特定时间投递给真正的 topic。
broker有配置项messageDelayLevel，默认值为“1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h”，18个level。
messageDelayLevel是broker的属性，不属于某个topic。发消息时，设置delayLevel等级即可： msg.setDelayLevel(level)。level有以下三种情况：

  • level == 0，消息为非延迟消息
  • 1<=level<=maxLevel，消息延迟特定时间，例如level==1，延迟1s
  • level > maxLevel，则level== maxLevel，例如level==20，延迟2h

定时消息会暂存在名为SCHEDULE_TOPIC_XXXX的topic中，并根据delayTimeLevel存入特定的 queue，queueId = delayTimeLevel – 1，即一个queue只存相同延迟的消息，保证具有相同发送延迟 的消息能够顺序消费。broker会调度地消费SCHEDULE_TOPIC_XXXX，将消息写入真实的topic。
需要注意的是，定时消息会在第一次写入和调度写入真实topic时都会计数，因此发送数量、tps都 会变高。

消息重试

Consumer消费消息失败后，要提供一种重试机制，令消息再消费一次。Consumer消费消息失败 通常可以认为有以下几种情况：

  1. 由于消息本身的原因，例如反序列化失败，消息数据本身无法处理（例如话费充值，当前消息的 手机号被注销，无法充值）等。这种错误通常需要跳过这条消息，再消费其它消息，而这条失败的消息 即使立刻重试消费，99%也不成功，所以最好提供一种定时重试机制，即过10秒后再重试。
  2. 由于依赖的下游应用服务不可用，例如db连接不可用，外系统网络不可达等。遇到这种错误，即 使跳过当前失败的消息，消费其他消息同样也会报错。这种情况建议应用sleep 30s，再消费下一条消 息，这样可以减轻Broker重试消息的压力。

消息重投

生产者在发送消息时：

  • 同步消息失败会重投
  • 异步消息有重试
  • oneway没有任何保证。

消息重投保证消息尽可能发送成功、不丢失，但可能会造成消息重复，消息重复在RocketMQ中是 无法避免的问题。消息重复在一般情况下不会发生，当出现消息量大、网络抖动，消息重复就会是大概 率事件。另外，生产者主动重发、consumer负载变化也会导致重复消息。

如下方法可以设置消息重试策略：

  • retryTimesWhenSendFailed：同步发送失败重投次数，默认为2，因此生产者会最多尝试发送 retryTimesWhenSendFailed + 1次。不会选择上次失败的broker，尝试向其他broker发送，最大程度 保证消息不丢失。超过重投次数，抛异常，由客户端保证消息不丢失。当出现RemotingException、 MQClientException和部分MQBrokerException时会重投。
  • retryTimesWhenSendAsyncFailed：异步发送失败重试次数，异步重试不会选择其他broker， 仅在同一个broker上做重试，不保证消息不丢。
  • retryAnotherBrokerWhenNotStoreOK：消息刷盘（主或备）超时或slave不可用（返回状态非 SEND_OK），是否尝试发送到其他broker，默认false。十分重要消息可以开启。

流量控制

生产者流控，因为broker处理能力达到瓶颈；消费者流控，因为消费能力达到瓶颈。

生产者流控：

  • commitLog文件被锁时间超过osPageCacheBusyTimeOutMills时，参数默认为1000ms，发 生流控。
  • 如果开启transientStorePoolEnable = true，且broker为异步刷盘的主机，且 transientStorePool中资源不足，拒绝当前send请求，发生流控。
  • broker每隔10ms检查send请求队列头部请求的等待时间，如果超过 waitTimeMillsInSendQueue，默认200ms，拒绝当前send请求，发生流控。
  • broker通过拒绝send 请求方式实现流量控制。

注意，生产者流控，不会尝试消息重投。

2) 消费者流控：

  • 消费者本地缓存消息数超过pullThresholdForQueue时，默认1000。
  • 消费者本地缓存消息大小超过pullThresholdSizeForQueue时，默认100MB。
  • 消费者本地缓存消息跨度超过consumeConcurrentlyMaxSpan时，默认2000。
  • 消费者流控的结果是降低拉取频率。

死信队列

死信队列用于处理无法被正常消费的消息。
当一条消息初次消费失败，消息队列会自动进行消息重试；
达到最大重试次数后，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息，
此时，消息队列 不会立刻将消息丢弃，而是将其发送到该消费者对应的特殊队列中。

RocketMQ将这种正常情况下无法被消费的消息称为死信消息（Dead-Letter Message），
将存储死信消息的特殊队列称为死信队列（Dead-Letter Queue）。
在RocketMQ中，可以通过使用console控制台对死信队列中的消息进行重发来使得消费者实例再 次进行消费。

消费模式Push or Pull

RocketMQ消息订阅有两种模式，一种是Push模式（MQPushConsumer），即MQServer主动向 消费端推送；另外一种是Pull模式（MQPullConsumer），即消费端在需要时，主动到MQ Server拉 取。但在具体实现时，Push和Pull模式本质都是采用消费端主动拉取的方式，即consumer轮询从 broker拉取消息。

  • Push模式特点
    好处就是实时性高。不好处在于消费端的处理能力有限，当瞬间推送很多消息给消费端时，容易造 成消费端的消息积压，严重时会压垮客户端。
  • Pull模式
    好处就是主动权掌握在消费端自己手中，根据自己的处理能力量力而行。缺点就是如何控制Pull的 频率。定时间隔太久担心影响时效性，间隔太短担心做太多“无用功”浪费资源。比较折中的办法就是长 轮询。
  • Push模式与Pull模式的区别:
    Push方式里，consumer把长轮询的动作封装了，并注册MessageListener监听器，取到消息后， 唤醒MessageListener的consumeMessage()来消费，对用户而言，感觉消息是被推送过来的。
    Pull方式里，取消息的过程需要用户自己主动调用，首先通过打算消费的Topic拿到 MessageQueue的集合，遍历MessageQueue集合，然后针对每个MessageQueue批量取消息，一次 取完后，记录该队列下一次要取的开始offset，直到取完了，再换另一个MessageQueue。
    RocketMQ使用长轮询机制来模拟Push效果，算是兼顾了二者的优点。

RocketMQ中的角色及相关术语

1)消息模型（Message Model）

RocketMQ主要由 Producer、Broker、Consumer 三部分组成，其中Producer 负责生产消息， Consumer 负责消费消息，Broker 负责存储消息。Broker 在实际部署过程中对应一台服务器，每个 Broker 可以存储多个Topic的消息，每个Topic的消息也可以分片存储于不同的 Broker。Message Queue 用于存储消息的物理地址，每个Topic中的消息地址存储于多个 Message Queue 中。 ConsumerGroup 由多个Consumer 实例构成。

2)Producer

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

3)Consumer

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

4)PushConsumer

Consumer消费的一种类型，该模式下Broker收到数据后会主动推送给消费端。应用通常向 Consumer对象注册一个Listener接口，一旦收到消息，Consumer对象立刻回调Listener接口方法。该 消费模式一般实时性较高。

5)PullConsumer

Consumer消费的一种类型，应用通常主动调用Consumer的拉消息方法从Broker服务器拉消息、 主动权由应用控制。一旦获取了批量消息，应用就会启动消费过程。

6)ProducerGroup

同一类Producer的集合，这类Producer发送同一类消息且发送逻辑一致。如果发送的是事务消息 且原始生产者在发送之后崩溃，则Broker服务器会联系同一生产者组的其他生产者实例以提交或回溯消 费。

7)ConsumerGroup

同一类Consumer的集合，这类Consumer通常消费同一类消息且消费逻辑一致。消费者组使得在 消息消费方面，实现负载均衡和容错的目标变得非常容易。要注意的是，消费者组的消费者实例必须订 阅完全相同的Topic。RocketMQ 支持两种消息模式：集群消费（Clustering）和广播消费 （Broadcasting）。

8)Broker

消息中转角色，负责存储消息，转发消息，一般也称为 Server。在 JMS 规范中称为Provider。

9)广播消费

一条消息被多个 Consumer 消费，即使这些 Consumer 属于同一个 Consumer Group，消息也会 被 Consumer Group 中的每个 Consumer 都消费一次，广播消费中的 Consumer Group 概念可以认 为在消息划分方面无意义。
在 CORBA Notification 规范中，消费方式都属于广播消费。
在 JMS 规范中，相当于 JMS Topic（ publish/subscribe ）模型

10)集群消费

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

11)顺序消息

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

12)普通顺序消息

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

13)严格顺序消息

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

目前已知的应用只有数据库 binlog 同步强依赖严格顺序消息，其他应用绝大部分都可以容忍短暂 乱序，推荐使用普通的顺序消息。

14)Message Queue

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

15)标签（Tag）

为消息设置的标志，用于同一主题下区分不同类型的消息。来自同一业务单元的消息，可以根据不 同业务目的在同一主题下设置不同标签。标签能够有效地保持代码的清晰度和连贯性，并优化 RocketMQ提供的查询系统。消费者可以根据Tag实现对不同子主题的不同消费逻辑，实现更好的扩展 性。

RocketMQ环境搭建

软件准备

RocketMQ最新版本：4.5.1

环境要求

  • JDK 11.0.5
  • Linux64位系统(CentOS Linux release 7.7.1908)
  • 源码安装需要安装Maven 3.2.x
  • 4G+ free

安装及启动

1、 下载rocketmq 

#下载
wget https://archive.apache.org/dist/rocketmq/4.5.1/rocketmq-all4.5.1-bin-release.zip

2、 修改脚本 

bin/runserver.sh
bin/runbroker.sh
bin/tools.sh
nameserver:
#!/bin/sh
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#==========================================================================
=================
# Java Environment Setting
#==========================================================================
=================
error_exit ()
{
echo "ERROR: $1 !!"
exit 1
}
[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=$HOME/jdk/java
[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=/usr/java
[ ! -e "$JAVA_HOME/bin/java" ] && error_exit "Please set the JAVA_HOME
variable in your environment, We need java(x64)!"
export JAVA_HOME
export JAVA="$JAVA_HOME/bin/java"
export BASE_DIR=$(dirname $0)/..
export
CLASSPATH=.:${BASE_DIR}/conf:${JAVA_HOME}/jre/lib/ext:${BASE_DIR}/lib/*
#==========================================================================
=================
# JVM Configuration
#==========================================================================
=================
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -
XX:MetaspaceSize=64m -XX:MaxMetaspaceSize=160m"
JAVA_OPT="${JAVA_OPT} -XX:CMSInitiatingOccupancyFraction=70 -
XX:+CMSParallelRemarkEnabled -XX:SoftRefLRUPolicyMSPerMB=0 -
XX:+CMSClassUnloadingEnabled -XX:SurvivorRatio=8"
JAVA_OPT="${JAVA_OPT} -verbose:gc -Xlog:gc:/dev/shm/rmq_srv_gc.log -
XX:+PrintGCDetails"
JAVA_OPT="${JAVA_OPT} -XX:-OmitStackTraceInFastThrow"
JAVA_OPT="${JAVA_OPT} -XX:-UseLargePages"
# JAVA_OPT="${JAVA_OPT} -
Djava.ext.dirs=${JAVA_HOME}/jre/lib/ext:${BASE_DIR}/lib"
#JAVA_OPT="${JAVA_OPT} -Xdebug -
Xrunjdwp:transport=dt_socket,address=9555,server=y,suspend=n"
JAVA_OPT="${JAVA_OPT} ${JAVA_OPT_EXT}"
JAVA_OPT="${JAVA_OPT} -cp ${CLASSPATH}"
$JAVA ${JAVA_OPT} $@
vim bin/runserver.sh
删除
UseCMSCompactAtFullCollection
UseParNewGC
UseConcMarkSweepGC
修改内存：
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -
XX:MetaspaceSize=64mm -XX:MaxMetaspaceSize=160mm"
-Xloggc修改为-Xlog:gc

broker: 

#!/bin/sh
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#==========================================================================
=================
# Java Environment Setting
#==========================================================================
=================
error_exit ()
{
echo "ERROR: $1 !!"
exit 1
}
[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=$HOME/jdk/java
[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=/usr/java
[ ! -e "$JAVA_HOME/bin/java" ] && error_exit "Please set the JAVA_HOME
variable in your environment, We need java(x64)!"
export JAVA_HOME
export JAVA="$JAVA_HOME/bin/java"
export BASE_DIR=$(dirname $0)/..
export
CLASSPATH=.${JAVA_HOME}/jre/lib/ext:${BASE_DIR}/lib/*:${BASE_DIR}/conf:${CL
ASSPATH}
#==========================================================================
=================
# JVM Configuration
#==========================================================================
=================
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m"
JAVA_OPT="${JAVA_OPT} -XX:+UseG1GC -XX:G1HeapRegionSize=16m -
XX:G1ReservePercent=25 -XX:InitiatingHeapOccupancyPercent=30 -
XX:SoftRefLRUPolicyMSPerMB=0"
JAVA_OPT="${JAVA_OPT} -verbose:gc -Xloggc:/dev/shm/mq_gc_%p.log -
XX:+PrintGCDetails"
JAVA_OPT="${JAVA_OPT} -XX:-OmitStackTraceInFastThrow"
JAVA_OPT="${JAVA_OPT} -XX:+AlwaysPreTouch"
JAVA_OPT="${JAVA_OPT} -XX:MaxDirectMemorySize=15g"
JAVA_OPT="${JAVA_OPT} -XX:-UseLargePages -XX:-UseBiasedLocking"
#JAVA_OPT="${JAVA_OPT} -Xdebug -
Xrunjdwp:transport=dt_socket,address=9555,server=y,suspend=n"
JAVA_OPT="${JAVA_OPT} ${JAVA_OPT_EXT}"
JAVA_OPT="${JAVA_OPT} -cp ${CLASSPATH}"
numactl --interleave=all pwd > /dev/null 2>&1
if [ $? -eq 0 ]
then
if [ -z "$RMQ_NUMA_NODE" ] ; then
numactl --interleave=all $JAVA ${JAVA_OPT} $@
else
numactl --cpunodebind=$RMQ_NUMA_NODE --membind=$RMQ_NUMA_NODE $JAVA
${JAVA_OPT} $@
fi
else
$JAVA ${JAVA_OPT} --add-exports=java.base/jdk.internal.ref=ALL-UNNAMED
$@
fi
vim bin/runbroker.sh
删除：
PrintGCDateStamps
PrintGCApplicationStoppedTime
PrintAdaptiveSizePolicy
UseGCLogFileRotation
NumberOfGCLogFiles=5
GCLogFileSize=30m

tools: 

#!/bin/sh
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#==========================================================================
=================
# Java Environment Setting
#==========================================================================
=================
error_exit ()
{
echo "ERROR: $1 !!"
exit 1
}
[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=$HOME/jdk/java
[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=/usr/java
[ ! -e "$JAVA_HOME/bin/java" ] && error_exit "Please set the JAVA_HOME
variable in your environment, We need java(x64)!"
export JAVA_HOME
export JAVA="$JAVA_HOME/bin/java"
export BASE_DIR=$(dirname $0)/..
# export CLASSPATH=.:${BASE_DIR}/conf:${CLASSPATH}
export
CLASSPATH=.${JAVA_HOME}/jre/lib/ext:${BASE_DIR}/lib/*:${BASE_DIR}/conf:${CL
ASSPATH}
#==========================================================================
=================
# JVM Configuration
#==========================================================================
=================
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn256m -XX:PermSize=128m
-XX:MaxPermSize=128m"
# JAVA_OPT="${JAVA_OPT} -
Djava.ext.dirs=${BASE_DIR}/lib:${JAVA_HOME}/jre/lib/ext"
JAVA_OPT="${JAVA_OPT} -cp ${CLASSPATH}"
$JAVA ${JAVA_OPT} $@
vim bin/tools.sh
# 删除 JAVA_OPT="${JAVA_OPT} -
Djava.ext.dirs=${BASE_DIR}/lib:${JAVA_HOME}/jre/lib/ext"

3、 启动NameServer 

# 1.启动NameServer
mqnamesrv
# 2.查看启动日志
tail -f ~/logs/rocketmqlogs/namesrv.log

4、 启动Broker 

# 1.启动Broker
mqbroker -n localhost:9876
# 2.查看启动日志
tail -f ~/logs/rocketmqlogs/broker.log

RocketMQ环境测试

发送消息 

# 1.设置环境变量
export NAMESRV_ADDR=localhost:9876
# 2.使用安装包的Demo发送消息
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer

接收消息 

# 1.设置环境变量
export NAMESRV_ADDR=localhost:9876
# 2.接收消息
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer

关闭RocketMQ 

# 1.关闭NameServer
mqshutdown namesrv
# 2.关闭Broker
mqshutdown broker

RocketMQ相关API使用

具体参考代码，学员需熟悉各种用法，API等
DefaultMQProducer 生产者的默认实现
生产消息分同步发送和异步发送
DefaultMQConsumer：消费者的默认实现
消息的拉取和消息的推送

MyProducer.java 

package com.lagou.rocket.demo.produer;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.io.UnsupportedEncodingException;
public class MyProducer {
public static void main(String[] args) throws
UnsupportedEncodingException, InterruptedException, RemotingException,
MQClientException, MQBrokerException {
// 在实例化生产者的同时，指定了生产组名称
DefaultMQProducer producer = new
DefaultMQProducer("myproducer_grp_01");
// 指定NameServer的地址
producer.setNamesrvAddr("node1:9876");
// 对生产者进行初始化，然后就可以使用了
producer.start();
// 创建消息，第一个参数是主题名称，第二个参数是消息内容
Message message = new Message(
"tp_demo_01",
"hello lagou 01".getBytes(RemotingHelper.DEFAULT_CHARSET)
);
// 发送消息
final SendResult result = producer.send(message);
System.out.println(result);
// 关闭生产者
producer.shutdown();
}
}

MyAsyncProducer.java 

package com.lagou.rocket.demo.produer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.io.UnsupportedEncodingException;
public class MyAsyncProducer {
public static void main(String[] args) throws MQClientException,
UnsupportedEncodingException, RemotingException, InterruptedException {
// 实例化生产者，并指定生产组名称
DefaultMQProducer producer = new
DefaultMQProducer("producer_grp_01");
// 指定nameserver的地址
producer.setNamesrvAddr("node1:9876");
// 初始化生产者
producer.start();
for (int i = 0; i < 100; i++) {
Message message = new Message(
"tp_demo_02",
("hello lagou " + i).getBytes("utf-8")
);
// 消息的异步发送
producer.send(message, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.println("发送成功:" + sendResult);
}
@Override
public void onException(Throwable throwable) {
System.out.println("发送失败：" +
throwable.getMessage());
}
});
}
// 由于是异步发送消息，上面循环结束之后，消息可能还没收到broker的响应
// 如果不sleep一会儿，就报错
Thread.sleep(10_000);
// 关闭生产者
producer.shutdown();
}
}

MyPullConsumer.java 

package com.lagou.rocket.demo.consumer;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.Set;
/**
* 拉取消息的消费者
*/
public class MyPullConsumer {
public static void main(String[] args) throws MQClientException,
RemotingException, InterruptedException, MQBrokerException,
UnsupportedEncodingException {
// 拉取消息的消费者实例化，同时指定消费组名称
DefaultMQPullConsumer consumer = new
DefaultMQPullConsumer("consumer_grp_01");
// 设置nameserver的地址
consumer.setNamesrvAddr("node1:9876");
// 对消费者进行初始化，然后就可以使用了
consumer.start();
// 获取指定主题的消息队列集合
final Set<MessageQueue> messageQueues =
consumer.fetchSubscribeMessageQueues("tp_demo_01");
// 遍历该主题的各个消息队列，进行消费
for (MessageQueue messageQueue : messageQueues) {
// 第一个参数是MessageQueue对象，代表了当前主题的一个消息队列
// 第二个参数是一个表达式，对接收的消息按照tag进行过滤
// 支持"tag1 || tag2 || tag3"或者 "*"类型的写法；null或者"*"表示不对
消息进行tag过滤
// 第三个参数是消息的偏移量，从这里开始消费
// 第四个参数表示每次最多拉取多少条消息
final PullResult result = consumer.pull(messageQueue, "*", 0,
10);
// 打印消息队列的信息
System.out.println("message******queue******" + messageQueue);
// 获取从指定消息队列中拉取到的消息
final List<MessageExt> msgFoundList = result.getMsgFoundList();
if (msgFoundList == null) continue;
for (MessageExt messageExt : msgFoundList) {
System.out.println(messageExt);
System.out.println(new String(messageExt.getBody(), "utf8"));
}
}
// 关闭消费者
consumer.shutdown();
}
}

MyPushConsumer.java 

package com.lagou.rocket.demo.consumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import
org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import
org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import
org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import java.io.UnsupportedEncodingException;
import java.util.List;
/**
* 推送消息的消费
*/
public class MyPushConsumer {
public static void main(String[] args) throws MQClientException,
InterruptedException {
// 实例化推送消息消费者的对象，同时指定消费组名称
DefaultMQPushConsumer consumer = new
DefaultMQPushConsumer("consumer_grp_02");
// 指定nameserver的地址
consumer.setNamesrvAddr("node1:9876");
// 订阅主题
consumer.subscribe("tp_demo_02", "*");
// 添加消息监听器，一旦有消息推送过来，就进行消费
consumer.setMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus
consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
final MessageQueue messageQueue =
context.getMessageQueue();
System.out.println(messageQueue);
for (MessageExt msg : msgs) {
try {
System.out.println(new String(msg.getBody(), "utf8"));
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
// 消息消费成功
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
// 消息消费失败
// return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
});
// 初始化消费者，之后开始消费消息
consumer.start();
// 此处只是示例，生产中除非运维关掉，否则不应停掉，长服务
// Thread.sleep(30_000);
// // 关闭消费者
// consumer.shutdown();
}
}

RocketMQ和Spring的整合

下载rocketmq-spring项目 

mvn install -Dmaven.skip.test=true

消息生产者

1）添加依赖 

<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.0.1.RELEASE</version>
</parent>
<properties>
<rocketmq-spring-boot-starter-version>2.0.3</rocketmq-spring-bootstarter-version>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>${rocketmq-spring-boot-starter-version}</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.6</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>

2）配置文件 

# application.properties
rocketmq.name-server=192.168.80.121:9876;192.168.80.122:9876
rocketmq.producer.group=my-group

3）启动类 

@SpringBootApplication
public class MQProducerApplication {
public static void main(String[] args) {
SpringApplication.run(MQSpringBootApplication.class);
}
}

4）测试类 

@RunWith(SpringRunner.class)
@SpringBootTest(classes = {MQSpringBootApplication.class})
public class ProducerTest {
@Autowired
private RocketMQTemplate rocketMQTemplate;
@Test
public void test1(){
rocketMQTemplate.convertAndSend("springboot-mq","hello springboot
rocketmq");
}
}

消息消费者

1）添加依赖

同消息生产者

2）配置文件

同消息生产者

3）启动类 

@SpringBootApplication
public class MQConsumerApplication {
public static void main(String[] args) {
SpringApplication.run(MQSpringBootApplication.class);
}
}

4）消息监听器 

@Slf4j
@Component
@RocketMQMessageListener(topic = "springboot-mq",consumerGroup =
"springboot-mq-consumer-1")
public class Consumer implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
log.info("Receive message："+message);
}
}

你可能也喜欢