[前言:本教程基于RocketMQ4.2，项目基于maven构建，使用操作系统为*nix]
今天我们要介绍的就是RocketMQ，它是阿里巴巴在2012年开源的分布式消息中间件，目前已经捐赠给Apache基金会，在2016年11月成为Apache孵化项目。
在最开始，来介绍一下什么是消息队列，比如一个用户注册后，我们可能需要更新很多统计信息，比如用户总数、当天注册用户数、当月注册用户数，有些网站还会给这个用户发邮件，会给用户注册im账号，这些操作可能比较耗时，那怎么办呢?
我们可以通过消息队列来进行解耦，在用户注册后，我们把用户的信息发送到消息队列中，然后其他功能只需要不断的从消息队列中获取数据，然后进行相关的操作即可。
这里可以看到，消息队列至少有两个好处:
1.解耦了各个模块，让各个模块可以独立开发和部署。
2.减少主流程的步骤，非核心流程操作不会阻塞主流程。
那么都有哪些消息队列可用呢?最简单的消息中间件可能要数Redis，但是它不是一个传统意义上的消息队列，它缺少可靠性保证，它更注重的是性能，但是传统意义上的消息队列不仅要注重性能，还要关注其他几个点。
消息队列需要关注的点有:
第一，消息的可靠性。如果纯内存的消息队列，那么就可能会因为程序突然中止而导致数据丢失，所以一般消息队列都会记录到磁盘来保证其可靠性。
第二，消息的顺序。比如我们发送两个消息，比如消息a是修改用户名为aa，消息b是修改用户名为bb，正确的顺序是先发送消息a，然后发送消息b，如果我们的发送顺序是正确的，但是我们的消息队列因为网络等原因让消息b先到达，那么就会产生未知的bug。
首先来科普几个基础概念:
第一个概念即topic，也就是"主题"，它是用来区分消息的类型，比如我们在用户注册时发的消息和用户退出时发的消息，应该是两种类型的消息，这两种消息是不相关的，我们如何区分它们呢?我们可以通过"主题"来区分不同的消息。
第二个概念即producer，也就是"生产者"，它用来生产消息。比如我们要在用户注册时发一个消息，那么我们就可以写一个SignupProducer来作为消息的生产者。
第三个概念即consumer，也就是"消费者"，它用来消费消息。比如我们要在用户注册的时候修改统计数据，我们就可以把这个代码写到SignupConsumer来作为消息的消费者。
第四个概念即Broker，它相当于一个中转站，它从生产者获取消息，然后把消息分发到消费者，需要说明的是，broker需要有一些确认机制，它需要确保我们的消费者正确的收到消息。
对于RocketMQ来说，还有一个概念，即NameServer，它是命名服务器，即RocketMQ的寻址服务，它用于把Broker的路由信息做聚合，客户端通过命名服务器来获取指定topic的路由信息，从而决定对哪些broker做连接。
理论说了那么多，我们首先来下载一下RocketMQ吧，我们可以从http://rocketmq.apache.org/ 这个官网下载，然后我们解压缩之后，我们来看一下bin目录的内容吧:
xinguimeng ~/work/rocketmq-all-4.2.0-bin-release/bin $ ls
README.md mqbroker mqfiltersrv mqshutdown.cmd runserver.cmd
cachedog.sh mqbroker.cmd mqfiltersrv.cmd os.sh runserver.sh
cleancache.sh mqbroker.numanode0 mqfiltersrv.xml play.cmd setcache.sh
cleancache.v1.sh mqbroker.numanode1 mqnamesrv play.sh startfsrv.sh
mqadmin mqbroker.numanode2 mqnamesrv.cmd runbroker tools.cmd
mqadmin.cmd mqbroker.numanode3 mqnamesrv.xml runbroker.cmd tools.sh
mqadmin.xml mqbroker.xml mqshutdown runbroker.sh
xinguimeng ~/work/rocketmq-all-4.2.0-bin-release/bin $
我们首先启动命名服务器，我们操作如下:
xinguimeng ~/work/rocketmq-all-4.2.0-bin-release/bin $ sh mqnamesrv
Java HotSpot(TM) 64-Bit Server VM warning: Using the DefNew young collector with the CMS collector is deprecated and will likely be removed in a future release
Java HotSpot(TM) 64-Bit Server VM warning: UseCMSCompactAtFullCollection is deprecated and will likely be removed in a future release.
Java HotSpot(TM) 64-Bit Server VM warning: Cannot open file /dev/shm/rmq_srv_gc.log due to No such file or directory
The Name Server boot success. serializeType=JSON
这里会有报警，我们可以先忽略，然后我们启动broker，如下所示:
xinguimeng ~/work/rocketmq-all-4.2.0-bin-release/bin $ sh mqbroker -n "127.0.0.1:9876"
Java HotSpot(TM) 64-Bit Server VM warning: Cannot open file /dev/shm/mq_gc_pid61639.log due to No such file or directory
在我们启动完毕之后，我们可以新建一个maven项目，我们再pom.xm中写入如下内容:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.mengzhidu</groupId>
<artifactId>rocketmq-demo</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>com.alibaba.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>3.2.6</version>
</dependency>
</dependencies>
</project>
然后我们新建一个生产者范例，我们写入如下代码:
package com.mengzhidiu.rocketmq.demo;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
/**
* 生产者范例
*/
public class ProducerDemo {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("mengzhidu-user");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();
for (int i = 0; i < 5; i ++) {
Message message = new Message("user", "push", String.valueOf(i), new String("辛星-" + i).getBytes());
SendResult result = producer.send(message);
System.out.println("消息id为: " + result.getMsgId() + " 发送状态为:" + result.getSendStatus());
}
}
}
在生产者范例中，我们制定了消费组为"mengzhidu-user"，并且我们制定了命名服务器的地址，然后我们就开始了生产者，然后我们在一个for循环中发送了五次消息，然后我们在生产者中获取了消息的状态。
然后我们新建一个消费者范例，我们写入如下代码:
package com.mengzhidiu.rocketmq.demo;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageExt;
import java.util.List;
/**
* 消费者范例
*/
public class ConsumerDemo {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("mengzhidu-user");
consumer.setNamesrvAddr("127.0.0.1:9876");
consumer.subscribe("user", "push");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.registerMessageListener(new MessageListenerConcurrently() {
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
ConsumeConcurrentlyContext consumeConcurrentlyContext) {
Message message = list.get(0);
System.out.println("消费者收到消息的内容:" + new String(message.getBody()));
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
}
}
我们的消费者也制定了主题为"mengzhidu-user"，而且它也指定了命名服务器的地址，然后设置了在收到消息后的处理方式，然后就启动了消费者。
然后我们首先启动消费者，然后我们启动生产者，在一小段时间后，我们会看到消费者端打印如下:
消费者收到消息的内容:辛星-1
消费者收到消息的内容:辛星-2
消费者收到消息的内容:辛星-0
消费者收到消息的内容:辛星-3
消费者收到消息的内容:辛星-4
而生产者端打印如下:
消息id为: C0A81F5200002A9F0000000000000CB2 发送状态为:SEND_OK
消息id为: C0A81F5200002A9F0000000000000D34 发送状态为:SEND_OK
消息id为: C0A81F5200002A9F0000000000000DB6 发送状态为:SEND_OK
消息id为: C0A81F5200002A9F0000000000000E38 发送状态为:SEND_OK
消息id为: C0A81F5200002A9F0000000000000EBA 发送状态为:SEND_OK
至此，我们的第一个基于RocketMQ的例子就完成了，不过我们的操作略显粗糙，在之后，我们会做得更加细腻一些。
