RocketMQ最佳实践（一）4.0版本/概念介绍/安装调试/客户端demo

为什么选择RocketMQ

我们来看看官方回答：

“我们研究发现，对于ActiveMQ而言，随着越来越多的使用queues和topics，其IO成为了瓶颈。某些情况下，消费者缓慢（消费能力不足）还会拖慢生产者（造成消息阻塞）。虽然我们做了最大努力进行优化：节流、断路器或者回退，但是并不能进行优雅的扩展。因此我们开始专注于使用时下非常流行的kafka，但是仍然不能满足我们的要求，如低延迟和高可靠性，详情见这里。在这样的背景下，我们决定开发一个新的消息中间件来处理一系列广泛的使用场景，包括从传统的发布/订阅场景到高容量的实时交易系统中不允许消息丢失的场景。”

各位看官也可以撮这里去看看RocketMQ与ActiveMQ以及Kafka的比较。

 

核心概念

• 生产者（Producer）：消息发送方，将业务系统中产生的消息发送到brokers（brokers可以理解为消息代理，生产者和消费者之间是通过brokers进行消息的通信），rocketmq提供了以下消息发送方式：同步、异步、单向。
• 生产者组（Producer Group）：相同角色的生产者被归为同一组，比如通常情况下一个服务会部署多个实例，这多个实例就是一个组，生产者分组的作用只体现在消息回查的时候，即如果一个生产者组中的一个生产者实例发送一个事务消息到broker后挂掉了，那么broker会回查此实例所在组的其他实例，从而进行消息的提交或回滚操作。
• 消费者（Consumer）：消息消费方，从brokers拉取消息。站在用户的角度，有以下两种消费者。
• 主动消费者（PullConsumer）：从brokers拉取消息并消费。
• 被动消费者（PushConsumer）：内部也是通过pull方式获取消息，只是进行了扩展和封装，并给用户预留了一个回调接口去实现，当消息到底的时候会执行用户自定义的回调接口。
• 消费者组（Consumer Group）：和生产者组类似。其作用体现在实现消费者的负载均衡和容错，有了消费者组变得异常容易。需要注意的是：同一个消费者组的每个消费者实例订阅的主题必须相同。
• 主题（Topic）：主题就是消息传递的类型。一个生产者实例可以发送消息到多个主题，多个生产者实例也可以发送消息到同一个主题。同样的，对于消费者端来说，一个消费者组可以订阅多个主题的消息，一个主题的消息也可以被多个消费者组订阅。
• 消息（Message）：消息就像是你传递信息的信封。每个消息必须指定一个主题，就好比每个信封上都必须写明收件人。
• 消息队列（Message Queues）：在主题内部，逻辑划分了多个子主题，每个子主题被称为消息队列。这个概念在实现最大并发数、故障切换等功能上有巨大的作用。
• 标签（Tag）：标签，可以被认为是子主题。通常用于区分同一个主题下的不同作用或者说不同业务的消息。同时也是避免主题定义过多引起性能问题，通常情况下一个生产者组只向一个主题发送消息，其中不同业务的消息通过标签或者说子主题来区分。
• 消息代理（Broker）：消息代理是RockerMQ中很重要的角色。它接收生产者发送的消息，进行消息存储，为消费者拉取消息服务。它还存储消息消耗相关的元数据，包括消费群体，消费进度偏移和主题/队列信息。
• 命名服务（Name Server）：命名服务作为路由信息提供程序。生产者/消费者进行主题查找、消息代理查找、读取/写入消息都需要通过命名服务获取路由信息。
• 消息顺序（Message Order）：当我们使用DefaultMQPushConsumer时，我们可以选择使用“orderly”还是“concurrently”。
  • orderly：消费消息的有序化意味着消息被生产者按照每个消息队列发送的顺序消费。如果您正在处理全局顺序为强制的场景，请确保您使用的主题只有一个消息队列。注意：如果指定了消费顺序，则消息消费的最大并发性是消费组订阅的消息队列数。
  • concurrently：当同时消费时，消息消费的最大并发仅限于为每个消费客户端指定的线程池。注意：此模式不再保证消息顺序。

安装与调试

官方要求的环境：

• 64bit OS, Linux/Unix/Mac is recommended;
• 64bit JDK 1.7+;
• Maven 3.2.x
• Git

我的环境：（我喜欢使用较新的版本）

• CentOS Linux release 7.3.1611;
• 64bit JDK 1.8.0_91;
• apache-maven-3.5.0;
• Git 1.8.3.1

安装jdk

麻烦各位看官自行搜索，资料多的吓人。。。

安装maven

先去官网下载maven

然后上传到安装目录，解压：

[plain] view plain copy

 

1. sudo tar zxvf apache-maven-3.5.0-bin.tar.gz  

解压完成设置环境变量：

[plain] view plain copy

 

1. sudo vi /etc/profile  

然后使环境变量生效：

[plain] view plain copy

 

1. source /etc/profile  

最后验证是否安装成功：

[plain] view plain copy

 

1. mvn -v  

安装Git（so easy）

先检查看看是否已经安装过了：

[plain] view plain copy

 

1. git --version  

如果没有就开始安装：

[plain] view plain copy

 

1. sudo yum install git  

安装完毕再看看：

[plain] view plain copy

 

1. git --version  

下面进行RocketMq安装

编译：

[plain] view plain copy

 

1. > git clone https://github.com/apache/incubator-rocketmq.git  

[plain] view plain copy

 

1. > cd incubator-rocketmq  

[plain] view plain copy

 

1. > mvn clean package install -Prelease-all assembly:assembly -U  

[plain] view plain copy

 

1. > cd target/apache-rocketmq-all  

在执行mvn编译的时候，你可能会遇到如下的问题：

这是由于没有权限创建目录造成的。所以，要么你切换到root用户，要么使用sudo：

[plain] view plain copy

 

1. sudo mvn clean package install -Prelease-all assembly:assembly -U  

提示：sudo: mvn: command not found。好吧，也是醉了。我们还需要在你当前用户的Home目录下的一个隐藏文件（.bashrc）中添加点东西：

[plain] view plain copy

 

1. > cd ~  

[plain] view plain copy

 

1. > sudo vi .bashrc  

添加完成后，执行：source .bashrc  使修改生效。然后再重新执行看看：

[plain] view plain copy

 

1. sudo mvn clean package install -Prelease-all assembly:assembly -U  

时间稍微有点长，我的环境用了16分钟，请看官耐心等待，完成后如下图：

启动RocketMQ

修改默认配置

由于RocketMQ默认配置要求很高，比如内存至少就要4个G，开发调试环境根本吃不消，所以我们开始启动前需要先修改这些参数。否则的话，我们很有会遇到内存分配或者不够的问题。

修改target/apache-rocketmq-all/bin/runserver.sh

[plain] view plain copy

 

1. JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:PermSize=128m -XX:MaxPermSize=320m"  

修改target/apache-rocketmq-all/bin/runbroker.sh

[plain] view plain copy

 

1. JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m  

修改target/apache-rocketmq-all/bin/tools.sh

[plain] view plain copy

 

1. JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:PermSize=128m -XX:MaxPermSize=128m"  

启动NameServer

进入target/apache-rocketmq-all目录下

[plain] view plain copy

 

1. > nohup sh bin/mqnamesrv &  

[plain] view plain copy

 

1. > tail -f ~/logs/rocketmqlogs/namesrv.log  

[plain] view plain copy

 

1. The Name Server boot success...  

启动Broker

[plain] view plain copy

 

1. > nohup sh bin/mqbroker -n localhost:9876 &  

[plain] view plain copy

 

1. > tail -f ~/logs/rocketmqlogs/broker.log   

[plain] view plain copy

 

1. The broker[%s, 172.17.0.1:10911] boot success...  

开放端口

[plain] view plain copy

 

1. sudo vi /etc/sysconfig/iptables  

然后重启生效：

[plain] view plain copy

 

1. sudo systemctl restart iptables  

添加ROCKETMQ_HOME环境变量

[plain] view plain copy

 

1. sudo vi /etc/profile  

[plain] view plain copy

 

1. source /etc/profile  

java客户端

pom.xml

[html] view plain copy

 

1. <rocketmq.version>4.0.0-incubating</rocketmq.version>  
2.   
3. <dependency>  
4.     <groupId>org.apache.rocketmq</groupId>  
5.     <artifactId>rocketmq-client</artifactId>  
6.     <version>${rocketmq.version}</version>  
7. </dependency>  
8. <dependency>  
9.     <groupId>org.apache.rocketmq</groupId>  
10.     <artifactId>rocketmq-common</artifactId>  
11.     <version>${rocketmq.version}</version>  
12. </dependency>  

生产者

[java] view plain copy

 

1. import org.apache.rocketmq.client.exception.MQClientException;  
2. import org.apache.rocketmq.client.producer.DefaultMQProducer;  
3. import org.apache.rocketmq.client.producer.SendResult;  
4. import org.apache.rocketmq.common.message.Message;  
5.   
6. import java.util.concurrent.TimeUnit;  
7.   
8. public class Producer {  
9.     public static void main(String[] args) throws MQClientException,  
10.             InterruptedException {  
11.         /** 
12.          * 一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例<br> 
13.          * 注意：ProducerGroupName需要由应用来保证唯一<br> 
14.          * ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键， 
15.          * 因为服务器会回查这个Group下的任意一个Producer 
16.          */  
17.         DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");  
18.         producer.setNamesrvAddr("192.168.56.101:9876");  
19.         producer.setInstanceName("Producer");  
20.         producer.setVipChannelEnabled(false);  
21.   
22.         /** 
23.          * Producer对象在使用之前必须要调用start初始化，初始化一次即可<br> 
24.          * 注意：切记不可以在每次发送消息时，都调用start方法 
25.          */  
26.         producer.start();  
27.   
28.         /** 
29.          * 下面这段代码表明一个Producer对象可以发送多个topic，多个tag的消息。 
30.          * 注意：send方法是同步调用，只要不抛异常就标识成功。但是发送成功也可会有多种状态，<br> 
31.          * 例如消息写入Master成功，但是Slave不成功，这种情况消息属于成功，但是对于个别应用如果对消息可靠性要求极高，<br> 
32.          * 需要对这种情况做处理。另外，消息可能会存在发送失败的情况，失败重试由应用来处理。 
33.          */  
34.         for (int i = 0; i < 1; i++) {  
35.             try {  
36.                 {  
37.                     Message msg = new Message("TopicTest1",// topic  
38.                             "TagA",// tag  
39.                             "OrderID001",// key  
40.                             ("Hello MetaQ").getBytes());// body  
41.                     SendResult sendResult = producer.send(msg);  
42.                     System.out.println(sendResult);  
43.                 }  
44.   
45.                 {  
46.                     Message msg = new Message("TopicTest2",// topic  
47.                             "TagB",// tag  
48.                             "OrderID0034",// key  
49.                             ("Hello MetaQ").getBytes());// body  
50.                     SendResult sendResult = producer.send(msg);  
51.                     System.out.println(sendResult);  
52.                 }  
53.   
54.                 {  
55.                     Message msg = new Message("TopicTest3",// topic  
56.                             "TagC",// tag  
57.                             "OrderID061",// key  
58.                             ("Hello MetaQ").getBytes());// body  
59.                     SendResult sendResult = producer.send(msg);  
60.                     System.out.println(sendResult);  
61.                 }  
62.             } catch (Exception e) {  
63.                 e.printStackTrace();  
64.             }  
65.             TimeUnit.MILLISECONDS.sleep(1000);  
66.         }  
67.   
68.         /** 
69.          * 应用退出时，要调用shutdown来清理资源，关闭网络连接，从MetaQ服务器上注销自己 
70.          * 注意：我们建议应用在JBOSS、Tomcat等容器的退出钩子里调用shutdown方法 
71.          */  
72.         producer.shutdown();  
73.     }  
74. }  

消费者

 

[java] view plain copy

 

1. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;  
2. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;  
3. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;  
4. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;  
5. import org.apache.rocketmq.client.exception.MQClientException;  
6. import org.apache.rocketmq.common.message.MessageExt;  
7.   
8. import java.util.List;  
9.   
10. public class PushConsumer {  
11.   
12.     /** 
13.      * 当前例子是PushConsumer用法，使用方式给用户感觉是消息从RocketMQ服务器推到了应用客户端。<br> 
14.      * 但是实际PushConsumer内部是使用长轮询Pull方式从MetaQ服务器拉消息，然后再回调用户Listener方法<br> 
15.      */  
16.     public static void main(String[] args) throws InterruptedException,  
17.             MQClientException {  
18.         /** 
19.          * 一个应用创建一个Consumer，由应用来维护此对象，可以设置为全局对象或者单例<br> 
20.          * 注意：ConsumerGroupName需要由应用来保证唯一 
21.          */  
22.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(  
23.                 "ConsumerGroupName");  
24.         consumer.setNamesrvAddr("192.168.56.101:9876");  
25.         consumer.setInstanceName("Consumber");  
26.   
27.         /** 
28.          * 订阅指定topic下tags分别等于TagA或TagC或TagD 
29.          */  
30.         consumer.subscribe("TopicTest1", "TagA || TagC || TagD");  
31.         /** 
32.          * 订阅指定topic下所有消息<br> 
33.          * 注意：一个consumer对象可以订阅多个topic 
34.          */  
35.         consumer.subscribe("TopicTest2", "*");  
36.   
37.         consumer.registerMessageListener(new MessageListenerConcurrently() {  
38.   
39.             /** 
40.              * 默认msgs里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息 
41.              */  
42.             @Override  
43.             public ConsumeConcurrentlyStatus consumeMessage(  
44.                     List<MessageExt> msgs, ConsumeConcurrentlyContext context) {  
45.                 System.out.println(Thread.currentThread().getName()  
46.                         + " Receive New Messages: " + msgs.size());  
47.   
48.                 MessageExt msg = msgs.get(0);  
49.                 if (msg.getTopic().equals("TopicTest1")) {  
50.                     // 执行TopicTest1的消费逻辑  
51.                     if (msg.getTags() != null && msg.getTags().equals("TagA")) {  
52.                         // 执行TagA的消费  
53.                         System.out.println(new String(msg.getBody()));  
54.                     } else if (msg.getTags() != null  
55.                             && msg.getTags().equals("TagC")) {  
56.                         // 执行TagC的消费  
57.                     } else if (msg.getTags() != null  
58.                             && msg.getTags().equals("TagD")) {  
59.                         // 执行TagD的消费  
60.                     }  
61.                 } else if (msg.getTopic().equals("TopicTest2")) {  
62.                     System.out.println(new String(msg.getBody()));  
63.                 }  
64.   
65.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;  
66.             }  
67.         });  
68.   
69.         /** 
70.          * Consumer对象在使用之前必须要调用start初始化，初始化一次即可<br> 
71.          */  
72.         consumer.start();  
73.   
74.         System.out.println("Consumer Started.");  
75.     }  
76. }  

 

 

http://blog.csdn.net/jayjjb/article/details/69948357