RocketMQ 实战入门

RocketMQ 是什么

Github 上关于 RocketMQ 的介绍：

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

支持发布/订阅（Pub/Sub）和点对点（P2P）消息模型
在一个队列中可靠的先进先出（FIFO）和严格的顺序传递
支持拉（pull）和推（push）两种消息模式
单一队列百万消息的堆积能力
支持多种消息协议，如 JMS、MQTT 等
分布式高可用的部署架构,满足至少一次消息传递语义
提供 docker 镜像用于隔离测试和云集群部署
提供配置、指标和监控等功能丰富的 Dashboard
对于这些特性描述，大家简单过一眼就即可，深入学习之后自然就明白了。

专业术语

Producer

消息生产者，生产者的作用就是将消息发送到 MQ，生产者本身既可以产生消息，如读取文本信息等。也可以对外提供接口，由外部应用来调用接口，再由生产者将收到的消息发送到 MQ。

Producer Group

生产者组，简单来说就是多个发送同一类消息的生产者称之为一个生产者组。在这里可以不用关心，只要知道有这么一个概念即可。

Consumer

消息消费者，简单来说，消费 MQ 上的消息的应用程序就是消费者，至于消息是否进行逻辑处理，还是直接存储到数据库等取决于业务需要。

Consumer Group

消费者组，和生产者类似，消费同一类消息的多个 consumer 实例组成一个消费者组。

Topic

Topic 是一种消息的逻辑分类，比如说你有订单类的消息，也有库存类的消息，那么就需要进行分类，一个是订单 Topic 存放订单相关的消息，一个是库存 Topic 存储库存相关的消息。

Message

Message 是消息的载体。一个 Message 必须指定 topic，相当于寄信的地址。Message 还有一个可选的 tag 设置，以便消费端可以基于 tag 进行过滤消息。也可以添加额外的键值对，例如你需要一个业务 key 来查找 broker 上的消息，方便在开发过程中诊断问题。

Tag

标签可以被认为是对 Topic 进一步细化。一般在相同业务模块中通过引入标签来标记不同用途的消息。

Broker

Broker 是 RocketMQ 系统的主要角色，其实就是前面一直说的 MQ。Broker 接收来自生产者的消息，储存以及为消费者拉取消息的请求做好准备。

Name Server

Name Server 为 producer 和 consumer 提供路由信息。

RocketMQ 架构

由这张图可以看到有四个集群，分别是 NameServer 集群、Broker 集群、Producer 集群和 Consumer 集群：

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

Consumer 组成一个集群。
简单说明一下图中箭头含义，从 Broker 开始，Broker Master1 和 Broker Slave1 是主从结构，它们之间会进行数据同步，即 Date Sync。同时每个 Broker 与

NameServer 集群中的所有节

点建立长连接，定时注册 Topic 信息到所有 NameServer 中。

Producer 与 NameServer 集群中的其中一个节点（随机选择）建立长连接，定期从 NameServer 获取 Topic 路由信息，并向提供 Topic 服务的 Broker Master 建立长连接，且定时向 Broker 发送心跳。Producer 只能将消息发送到 Broker master，但是 Consumer 则不一样，它同时和提供 Topic 服务的 Master 和 Slave

建立长连接，既可以从 Broker Master 订阅消息，也可以从 Broker Slave 订阅消息。

RocketMQ 集群部署模式

单 master 模式

也就是只有一个 master 节点，称不上是集群，一旦这个 master 节点宕机，那么整个服务就不可用，适合个人学习使用。
多 master 模式

多个 master 节点组成集群，单个 master 节点宕机或者重启对应用没有影响。

优点：所有模式中性能最高

缺点：单个 master 节点宕机期间，未被消费的消息在节点恢复之前不可用，消息的实时性就受到影响。
注意：使用同步刷盘可以保证消息不丢失，同时 Topic 相对应的 queue 应该分布在集群中各个节点，而不是只在某各节点上，否则，该节点宕机会对订阅该 topic 的应用造成影响。
多 master 多 slave 异步复制模式

在多 master 模式的基础上，每个 master 节点都有至少一个对应的 slave。master

节点可读可写，但是 slave 只能读不能写，类似于 mysql 的主备模式。

优点： 在 master 宕机时，消费者可以从 slave 读取消息，消息的实时性不会受影响，性能几乎和多 master 一样。

缺点：使用异步复制的同步方式有可能会有消息丢失的问题。
多 master 多 slave 同步双写模式

同多 master 多 slave 异步复制模式类似，区别在于 master 和 slave 之间的数据同步方式。

优点：同步双写的同步模式能保证数据不丢失。

缺点：发送单个消息 RT 会略长，性能相比异步复制低10%左右。

刷盘策略：同步刷盘和异步刷盘（指的是节点自身数据是同步还是异步存储）

同步方式：同步双写和异步复制（指的一组 master 和 slave 之间数据的同步）
注意：要保证数据可靠，需采用同步刷盘和同步双写的方式，但性能会较其他方式低。

RocketMQ 单主部署

鉴于是快速入门，我选择的是第一种单 master 的部署模式。先说明一下我的安装环境：

Centos 7.2
jdk 1.8
Maven 3.2.x
Git
这里 git 可用可不用，主要是用来直接下载 github 上的源码。也可以选择自己到

github 上下载，然后上传到服务器上。以git操作为示例。

clone 源码并用 maven 编译

> git clone https://github.com/alibaba/RocketMQ.git /opt/RocketMQ
> cd /opt/RocketMQ && mvn -Dmaven.test.skip=true clean package install assembly:assembly -U
> cd target/alibaba-rocketmq-broker/alibaba-rocketmq

此处可能遇到的问题

一、执行"git clone https://github.com/alibaba/RocketMQ.git /home/inspkgs/RocketMQ"时出现以下提示：

fatal: unable to access 'https://github.com/alibaba/RocketMQ.git/': Could not resolve host: github.com; Unknown error

解决办法：一般是由于网络原因造成的，执行以下命令

> ping github.com

确定可以 ping 通之后，再重新执行 git clone 命令。

二、执行"mvn -Dmaven.test.skip=true clean package install assembly:assembly -U"编译时，可能出现下载相关jar很慢的情况。

这也是由于默认 maven 中央仓库在国外的原因，可以根据需要在 /home/maven/conf/setting.xml 中的 <mirrors></mirrors> 添加以下内容后重新编译：

<mirror>
<id>aliyun</id>
<mirrorOf>central</mirrorOf>
<name>aliyun maven</name>
<url>http://maven.aliyun.com/nexus/content/groups/public/</url>
</mirror>

启动 Name Server

> nohup sh /opt/RocketMQ/bin/mqnamesrv &
//执行 jps 查看进程
> jps
25913 NamesrvStartup
//查看日志确保服务已正常启动
> tail -f ~/logs/rocketmqlogs/namesrv.log
The Name Server boot success...

启动 broker

> nohup sh /opt/RocketMQ/bin/mqbroker -n localhost:9876 &
//执行 jps 查看进程
> jps
25954 BrokerStartup
//查看日志确保服务已正常启动
> tail -f ~/logs/rocketmqlogs/broker.log
The broker[broker-a, 10.1.54.121:10911] boot success...

发送和接收消息

发送/接收消息之前,我们需要告诉客户端 NameServer 地址。RocketMQ 提供了多种方式来实现这一目标。为简单起见,我们使用环境变量 NAMESRV_ADDR。

> export NAMESRV_ADDR=localhost:9876
> sh /opt/RocketMQ/bin/tools.sh com.alibaba.rocketmq.example.quickstart.Producer
SendResult [sendStatus=SEND_OK, msgId= ...
> sh /opt/RocketMQ/bin/tools.sh com.alibaba.rocketmq.example.quickstart.Consumer
ConsumeMessageThread_%d Receive New Messages: [MessageExt...

关闭服务

> sh /opt/RocketMQ/bin/mqshutdown broker
The mqbroker(36695) is running...
Send shutdown request to mqbroker(36695) OK
> sh /opt/RocketMQ/bin/mqshutdown namesrv
The mqnamesrv(36664) is running...
Send shutdown request to mqnamesrv(36664) OK

生产者、消费者 Demo

生产者

public class Producer {
public static void main(String[] args) throws MQClientException, InterruptedException {

//声明并初始化一个producer
//需要一个producer group名字作为构造方法的参数，这里为producer1
DefaultMQProducer producer = new DefaultMQProducer("producer1");

//设置NameServer地址,此处应改为实际NameServer地址，多个地址之间用；分隔
//NameServer的地址必须有，但是也可以通过环境变量的方式设置，不一定非得写死在代码里
producer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");

//调用start()方法启动一个producer实例
producer.start();

//发送10条消息到Topic为TopicTest，tag为TagA，消息内容为“Hello RocketMQ”拼接上i的值
for (int i = 0; i < 10; i++) {
try {
Message msg = new Message("TopicTest",// topic
"TagA",// tag
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)// body
);

//调用producer的send()方法发送消息
//这里调用的是同步的方式，所以会有返回结果
SendResult sendResult = producer.send(msg);

//打印返回结果，可以看到消息发送的状态以及一些相关信息
System.out.println(sendResult);
} catch (Exception e) {
e.printStackTrace();
Thread.sleep(1000);
}
}

//发送完消息之后，调用shutdown()方法关闭producer
producer.shutdown();
}
}

消费者

public class Consumer {

public static void main(String[] args) throws InterruptedException, MQClientException {

//声明并初始化一个consumer
//需要一个consumer group名字作为构造方法的参数，这里为consumer1
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer1");

//同样也要设置NameServer地址
consumer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");

//这里设置的是一个consumer的消费策略
//CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
//CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
//CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

//设置consumer所订阅的Topic和Tag，*代表全部的Tag
consumer.subscribe("TopicTest", "*");

//设置一个Listener，主要进行消息的逻辑处理
consumer.registerMessageListener(new MessageListenerConcurrently() {

@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {

System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);

//返回消费状态
//CONSUME_SUCCESS 消费成功
//RECONSUME_LATER 消费失败，需要稍后重新消费
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});

//调用start()方法启动consumer
consumer.start();

System.out.println("Consumer Started.");
}
}

必知必会的RocketMQ消息类型

普通消息

普通消息也叫做无序消息，简单来说就是没有顺序的消息，producer 只管发送消息，consumer 只管接收消息，至于消息和消息之间的顺序并没有保证，可能先发送的消息先消费，也可能先发送的消息后消费。

举个简单例子，producer 依次发送 order id 为 1、2、3 的消息到 broker，consumer 接到的消息顺序有可能是 1、2、3，也有可能是 2、1、3 等情况，这就是普通消息。

因为不需要保证消息的顺序，所以消息可以大规模并发地发送和消费，吞吐量很高，适合大部分场景。

代码示例：

生产者

public class Producer {
public static void main(String[] args) throws MQClientException, InterruptedException {

//声明并初始化一个producer
//需要一个producer group名字作为构造方法的参数，这里为concurrent_producer
DefaultMQProducer producer = new DefaultMQProducer("concurrent_producer");

//设置NameServer地址,此处应改为实际NameServer地址，多个地址之间用；分隔
//NameServer的地址必须有，但是也可以通过环境变量的方式设置，不一定非得写死在代码里
producer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");

//调用start()方法启动一个producer实例
producer.start();

//发送10条消息到Topic为TopicTest，tag为TagA，消息内容为“Hello RocketMQ”拼接上i的值
for (int i = 0; i < 10; i++) {
try {
Message msg = new Message("TopicTestConcurrent",// topic
"TagA",// tag
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)// body
);

//调用producer的send()方法发送消息
//这里调用的是同步的方式，所以会有返回结果，同时默认发送的也是普通消息
SendResult sendResult = producer.send(msg);

//打印返回结果，可以看到消息发送的状态以及一些相关信息
System.out.println(sendResult);
} catch (Exception e) {
e.printStackTrace();
Thread.sleep(1000);
}
}

//发送完消息之后，调用shutdown()方法关闭producer
producer.shutdown();
}
}

消费者

public class Consumer {

public static void main(String[] args) throws InterruptedException, MQClientException {

//声明并初始化一个consumer
//需要一个consumer group名字作为构造方法的参数，这里为concurrent_consumer
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("concurrent_consumer");

//同样也要设置NameServer地址
consumer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");

//这里设置的是一个consumer的消费策略
//CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
//CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
//CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

//设置consumer所订阅的Topic和Tag，*代表全部的Tag
consumer.subscribe("TopicTestConcurrent", "*");

//设置一个Listener，主要进行消息的逻辑处理
//注意这里使用的是MessageListenerConcurrently这个接口
consumer.registerMessageListener(new MessageListenerConcurrently() {

@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {

System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);

//返回消费状态
//CONSUME_SUCCESS 消费成功
//RECONSUME_LATER 消费失败，需要稍后重新消费
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});

//调用start()方法启动consumer
consumer.start();

System.out.println("Consumer Started.");
}
}

有序消息

有序消息就是按照一定的先后顺序的消息类型。

举个例子来说，producer 依次发送 order id 为 1、2、3 的消息到 broker，consumer 接到的消息顺序也就是 1、2、3 ，而不会出现普通消息那样的 2、1、3 等情况。

那么有序消息是如何保证的呢？我们都知道消息首先由 producer 到 broker，再从 broker 到 consumer，分这两步走。那么要保证消息的有序，势必这两步都是要保证有序的，即要保证消息是按有序发送到 broker，broker 也是有序将消息投递给 consumer，两个条件必须同时满足，缺一不可。

进一步还可以将有序消息分成

全局有序消息
局部有序消息
之前我们讲过，topic 只是消息的逻辑分类，内部实现其实是由 queue 组成。当 producer 把消息发送到某个 topic 时，默认是会消息发送到具体的 queue 上。



举个例子，producer 发送 order id 为 1、2、3、4 的四条消息到 topicA 上，假设 topicA 的 queue 数为 3 个（queue0、queue1、queue2），那么消息的分布可能就是这种情况，id 为 1 的在 queue0，id 为 2 的在 queue1，id 为 3 的在 queue2，id 为 4 的在 queue0。同样的，consumer 消费时也是按 queue 去消费，这时候就可能出现先消费 1、4，再消费 2、3，和我们的预期不符。那么我们如何实现
1、2、3、4 的消费顺序呢？道理其实很简单，只需要把订单 topic 的 queue 数改为 1，如此一来，只要 producer 按照 1、2、3、4 的顺序去发送消息，那么 consumer 自然也就按照 1、2、3、4 的顺序去消费，这就是全局有序消息。

由于一个 topic 只有一个 queue ，即使我们有多个 producer 实例和 consumer 实例也很难提高消息吞吐量。就好比过独木桥，大家只能一个挨着一个过去，效率低下。

那么有没有吞吐量和有序之间折中的方案呢？其实是有的，就是局部有序消息。



我们知道订单消息可以再细分为订单创建、订单付款、订单完成等消息，这些消息都有相同的 order id。同时，也只有按照订单创建、订单付款、订单完成的顺序去消费才符合业务逻辑。但是不同 order id 的消息是可以并行的，不会影响到业务。这时候就常见做法就是将 order id 进行处理，将 order id 相同的消息发送到 topicB 的同一个 queue，假设我们 topicB 有 2 个 queue，那么我们可以简单的对 id 取余，奇数的发往 queue0，偶数的发往 queue1，消费者按照
queue 去消费时，就能保证 queue0 里面的消息有序消费，queue1 里面的消息有序消费。

由于一个 topic 可以有多个 queue，所以在性能比全局有序高得多。假设 queue 数是 n，理论上性能就是全局有序的 n 倍，当然 consumer 也要跟着增加才行。在实际情况中，这种局部有序消息是会比全局有序消息用的更多。

示例代码：

生产者

public class Producer {
public static void main(String[] args) throws UnsupportedEncodingException {
try {
// 声明并初始化一个producer
// 需要一个producer group名字作为构造方法的参数，这里为ordered_producer
DefaultMQProducer orderedProducer = new DefaultMQProducer("ordered_producer");

// 设置NameServer地址,此处应改为实际NameServer地址，多个地址之间用；分隔
//NameServer的地址必须有，但是也可以通过环境变量的方式设置，不一定非得写死在代码里
orderedProducer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");

// 调用start()方法启动一个producer实例
orderedProducer.start();

// 自定义一个tag数组
String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};

// 发送10条消息到Topic为TopicTestOrdered，tag为tags数组按顺序取值，
// key值为“KEY”拼接上i的值，消息内容为“Hello RocketMQ”拼接上i的值
for (int i = 0; i < 10; i++) {

int orderId = i % 10;
Message msg =
new Message("TopicTestOrdered", tags[i % tags.length], "KEY" + i,
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));

SendResult sendResult = orderedProducer.send(msg, new MessageQueueSelector() {

// 选择发送消息的队列
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {

// arg的值其实就是orderId
Integer id = (Integer) arg;

// mqs是队列集合，也就是topic所对应的所有队列
int index = id % mqs.size();

// 这里根据前面的id对队列集合大小求余来返回所对应的队列
return mqs.get(index);
}
}, orderId);

System.out.println(sendResult);
}

orderedProducer.shutdown();
} catch (MQClientException e) {
e.printStackTrace();
} catch (RemotingException e) {
e.printStackTrace();
} catch (MQBrokerException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

至于是要实现全局有序，还是局部有序，在此示例代码中，就取决于 TopicTestOrdered 这个 Topic 的队列数了。

消费者

public class Consumer {

public static void main(String[] args) throws MQClientException {

//声明并初始化一个consumer
//需要一个consumer group名字作为构造方法的参数，这里为concurrent_consumer
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ordered_consumer");

//同样也要设置NameServer地址
consumer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");

//这里设置的是一个consumer的消费策略
//CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
//CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
//CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

//设置consumer所订阅的Topic和Tag
consumer.subscribe("TopicTestOrdered", "TagA || TagC || TagD");

//设置一个Listener，主要进行消息的逻辑处理
//注意这里使用的是MessageListenerOrderly这个接口
consumer.registerMessageListener(new MessageListenerOrderly() {

@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {

System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);

//返回消费状态
//SUCCESS 消费成功
//SUSPEND_CURRENT_QUEUE_A_MOMENT 消费失败，暂停当前队列的消费
return ConsumeOrderlyStatus.SUCCESS;
}
});

//调用start()方法启动consumer
consumer.start();

System.out.println("Consumer Started.");
}
}

延时消息

延时消息，简单来说就是当 producer 将消息发送到 broker 后，会延时一定时间后才投递给 consumer 进行消费。

RcoketMQ的延时等级为：1s，5s，10s，30s，1m，2m，3m，4m，5m，6m，7m，8m，9m，10m，20m，30m，1h，2h。level=0，表示不延时。level=1，表示 1 级延时，对应延时 1s。level=2 表示 2 级延时，对应5s，以此类推。

这种消息一般适用于消息生产和消费之间有时间窗口要求的场景。比如说我们网购时，下单之后是有一个支付时间，超过这个时间未支付，系统就应该自动关闭该笔订单。那么在订单创建的时候就会就需要发送一条延时消息（延时15分钟）后投递给 consumer，consumer 接收消息后再对订单的支付状态进行判断是否关闭订单。

设置延时非常简单，只需要在Message设置对应的延时级别即可：

Message msg = new Message("TopicTest",// topic
"TagA",// tag
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)// body
);
// 这里设置需要延时的等级即可
msg.setDelayTimeLevel(3);
SendResult sendResult = producer.send(msg);

来源https://www.jianshu.com/p/824066d70da8 https://www.jianshu.com/p/11e875074a8f
其它相关文章：http://blog.csdn.net/a19881029/article/details/34446629 http://blog.csdn.net/u010634288/article/details/56686978 http://blog.csdn.net/damacheng/article/details/42846549 http://blog.csdn.net/oMaverick1/article/details/51331004