RocketMQ环境搭建和quickexample

在分布式系统中，各个子系统、节点之间需要进行及时的数据交换，消息中间件应运而生，现在比较常见的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

QQ图片20180403145532.png

上传到安装目录，解压：

tar -zvxf apache-maven-3.5.3-bin.tar.gz

解压完成设置环境变量：

sudo vi /etc/profile

QQ图片20180403143543.png

最后验证是否安装成功：

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安装

QQ图片20180403145639.png

进入目录

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