在分布式系统中，各个子系统、节点之间需要进行及时的数据交换，消息中间件应运而生，现在比较常见的MQ产品主要是ActiveMQ、RabbitMQ、ZeroMQ、Kafka、MetaMQ、RocketMQ。
消息中间件对比
那么这么多的消息中间件之间有什么异同呢？带着这个疑问,阿里中间件测试组对常见的三类消息产品(Kafka、RabbitMQ、RocketMQ)做了性能比较，并得出以下结论
Kafka是LinkedIn开源的分布式发布-订阅消息系统，目前归属于Apache定级项目。Kafka主要特点是基于Pull的模式来处理消息消费，追求高吞吐量，一开始的目的就是用于日志收集和传输。0.8版本开始支持复制，不支持事务，对消息的重复、丢失、错误没有严格要求，适合产生大量数据的互联网服务的数据收集业务。
RabbitMQ是使用Erlang语言开发的开源消息队列系统，基于AMQP协议来实现。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。AMQP协议更多用在企业系统内，对数据一致性、稳定性和可靠性要求很高的场景，对性能和吞吐量的要求还在其次。
RocketMQ是阿里开源的消息中间件，它是纯Java开发，具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。RocketMQ思路起源于Kafka，但并不是Kafka的一个Copy，它对消息的可靠传输及事务性做了优化，目前在阿里集团被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景。
可以看出Kafka追求高吞吐低可靠，RabbitMQ高可靠性能次之，RocketMQ两者兼顾。我们选择中庸的万金油RocketMQ来开启消息中间件之旅。
RocketMQ搭建
1.基础环境
Ubuntu 16.04 LTS (GNU/Linux 4.4.0-116-generic x86_64)
JDK1.8
2.安装Maven
先去官网下载maven
上传到安装目录，解压：
tar -zvxf apache-maven-3.5.3-bin.tar.gz
解压完成设置环境变量：
sudo vi /etc/profile
最后验证是否安装成功：
mvn -v
Apache Maven 3.5.3 (3383c37e1f9e9b3bc3df5050c29c8aff9f295297; 2018-02-25T03:49:05+08:00)
Maven home: /home/zx/apache-maven-3.5.3
Java version: 1.8.0_144, vendor: Oracle Corporation
Java home: /home/gaojz/jdk1.8.0_144/jre
Default locale: en_US, platform encoding: UTF-8
OS name: "linux", version: "4.4.0-116-generic", arch: "amd64", family: "unix"
3.RocketMq安装
下载RocketMq包：https://github.com/apache/rocketmq
进入目录
cd incubator-rocketmq
编译rocketmq项目
mvn clean package install -Prelease-all assembly:assembly -U
编译成功后启动
启动NameServer
nohup sh bin/mqnamesrv &
启动Broker
nohup sh bin/mqbroker -n 192.168.1.207:9876 autoCreateTopicEnable=true &
简单生产消费
依赖文件pom.xml
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.0.0-incubating</version>
</dependency>
生产者
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
producer.setNamesrvAddr("192.168.1.205:9876");
producer.start();
for (int i = 0; i < 10; i++){
try {
//Message(String topic, String tags, String keys, byte[] body)
Message msg = new Message("TopicTest","TagA","OrderID188",("Hello world"+i).getBytes(RemotingHelper.DEFAULT_CHARSET));
SendResult sendResult = producer.send(msg);
System.out.println(sendResult.toString());
} catch (Exception e) {
e.printStackTrace();
}
}
producer.shutdown();
}
消费者
public static void main(String[] args) throws InterruptedException, MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.setNamesrvAddr("192.168.1.205:9876");
consumer.subscribe("TopicTest", "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context) {
System.out.println(Thread.currentThread().getName() + " 收到新消息: ");
for(MessageExt me:msgs){
try {
System.out.println(new String(me.getBody(),"utf-8"));
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.println("Consumer Started.");
}
输出结果
Consumer Started.
ConsumeMessageThread_1 收到新消息:
Hello world0
ConsumeMessageThread_2 收到新消息:
Hello world1
ConsumeMessageThread_3 收到新消息:
Hello world2
ConsumeMessageThread_4 收到新消息:
Hello world3
ConsumeMessageThread_5 收到新消息:
Hello world4
ConsumeMessageThread_6 收到新消息:
Hello world5
ConsumeMessageThread_7 收到新消息:
Hello world6
ConsumeMessageThread_8 收到新消息:
Hello world7
ConsumeMessageThread_9 收到新消息:
Hello world8
ConsumeMessageThread_10 收到新消息:
Hello world9
项目地址：https://github.com/granett/rocketmq/tree/master/src/main/java/rocketmq/rocketmq/quickstart
后续继续探索RocketMQ的其他类型的实例。
