RocketMQ安装和小试

àì夳堔傛蜴生んèń 2023-06-16 08:11 4阅读 0赞

由于RocketMQ是java实现，在安装前须有java环境，并且有maven环境。项目地址：[https://github.com/alibaba/RocketMQ/][https_github.com_alibaba_RocketMQ]

1.下载、编译（window）

下载RocketMQ源码,并解压。

maven编译（不了解maven，百度一下，这里就不介绍了）：[http://download.csdn.net/detail/tianwei7518/8137909][http_download.csdn.net_detail_tianwei7518_8137909]

进入到源码解压目录下运行install.bat

或DOS命令行切换到解压目录运行： mvn -Dmaven.test.skip=true clean package install assembly:assembly -U

编译成功后，在target目录下会有alibaba-rocketmq-3.1.1.tar.gz，该压缩包就是安装包。

2.安装，运行（linux）

将alibaba-rocketmq-3.1.1.tar.gz上传到linux服务器，解压：

tar -zxvf alibaba-rocketmq-3.1.1.tar.gz

设置执行权限chmod +x  ./alibaba-rocketmq/bin/\*

由于我的虚拟器内存比较小，所以启动前需要调节一下，启动的虚拟内存参数配置。

vi  ./alibaba-rocketmq/bin/runserver.sh   \#nameserver 内存

vi  ./alibaba-rocketmq/bin/runbroker.sh  \#broke内存

JAVA\_OPT\_1="-server -Xms256m -Xmx256m -Xmn128m -XX:PermSize=64m -XX:MaxPermSize=128m"（参考你自己的机器内存）

启动nameserver：nohup ./bin/mqnamesrv >/dev/null 2>&1 &         \#默认端口9876

关闭nameserver：./bin/mqshutdown namesrv

启动mqbroker ：nohup ./bin/mqbroker -n 192.168.36.189:9876 >/dev/null 2>&1 &     \#默认端口10911（192.168.36.189:9876为nameserver，链接进行注册）

关闭mqbroker ：./bin/mqshutdown broker

下面看一个实例，体验一下：

**一、生产者Producer.java**

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

生产者实现类：

DefaultMQProducer:默认消息生产者，适合使用spring初始化

TransactionMQProducer：继承DefaultMQProducer，支持分布式事务Producer

**二、消费者Consumer.java**

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

消费者实现类：

DefaultMQPullConsumer：消费者，主动拉取方式消费

DefaultMQPushConsumer：类似于Broker Push消息到Consumer方式，但实际仍然是Consumer内部后台从Broker Pull消息，采用长轮询方式拉消息，实时性同push方式一致，且不会无谓的拉消息导致Broker、Consumer压力增大。

运行看一下结果吧。

**三、Consumer最佳实践**  
1.消费过程要做到幂等（即消费端去重）  
RocketMQ无法做到消息重复，所以如果业务对消息重复非常敏感，务必要在业务层面去重，有以下一些方式：  
（1）将消息的唯一键，可以是MsgId，也可以是消息内容中的唯一标识字段，例如订单ID，消费之前判断是否在DB或Tair（全局KV存储）中存在，如果不存在则插入，并消费，否则跳过。（实践过程要考虑原子性问题，判断是否存在可以尝试插入，如果报主键冲突，则插入失败，直接跳过） msgid一定是全局唯一的标识符，但是可能会存在同样的消息有两个不同的msgid的情况（有多种原因），这种情况可能会使业务上重复，建议最好使用消息体中的唯一标识字段去重  
（2）使业务层面的状态机去重  
2.批量方式消费  
如果业务流程支持批量方式消费，则可以很大程度上的提高吞吐量，可以通过设置Consumer的consumerMessageBatchMaxSize参数，默认是1，即一次消费一条参数  
3.跳过非重要的消息  
发生消息堆积时，如果消费速度一直跟不上发送速度，可以选择丢弃不重要的消息  
4.优化每条消息消费过程  
举例如下，某条消息的消费过程如下  
1.  根据消息从 DB 查询数据 1  
2.  根据消息从 DB 查询数据2  
3.  复杂的业务计算  
4.  向 DB 插入数据3  
5.  向 DB 插入数据 4  
这条消息的消费过程与 DB 交互了 4 次，如果按照每次 5ms 计算，那么总共耗时 20ms，假设业务计算耗时 5ms，那么总过耗时 25ms，如果能把 4 次 DB 交互优化为 2 次，那么总耗时就可以优化到 15ms，也就是说总体性能提高了 40%。  
对于 Mysql 等 DB，如果部署在磁盘，那么与 DB 进行交互，如果数据没有命中 cache，每次交互的 RT 会直线上升， 如果采用 SSD，则 RT 上升趋势要明显好于磁盘。  
个别应用可能会遇到这种情况：在线下压测消费过程中，db 表现非常好，每次 RT 都很短，但是上线运行一段时间，RT 就会变长，消费吞吐量直线下降  
主要原因是线下压测时间过短，线上运行一段时间后，cache 命中率下降，那么 RT 就会增加。建议在线下压测时，要测试足够长时间，尽可能模拟线上环境，压测过程中，数据的分布也很重要，数据不同，可能 cache 的命中率也会完全不同

**四、Producer最佳实践**  
1.发送消息注意事项  
（1） 一个应用尽可能用一个 Topic，消息子类型用 tags 来标识，tags 可以由应用自由设置。只有发送消息设置了tags，消费方在订阅消息时，才可以利用 tags 在 broker 做消息过滤。  
（2）每个消息在业务层面的唯一标识码，要设置到 keys 字段，方便将来定位消息丢失问题。服务器会为每个消息创建索引（哈希索引），应用可以通过 topic，key 来查询这条消息内容，以及消息被谁消费。由于是哈希索引，请务必保证 key 尽可能唯一，这样可以避免潜在的哈希冲突。  
（3）消息发送成功或者失败，要打印消息日志，务必要打印 sendresult 和 key 字段  
（4）send 消息方法，只要不抛异常，就代表发送成功。但是发送成功会有多个状态，在 sendResult 里定义  
SEND\_OK：消息发送成功  
FLUSH\_DISK\_TIMEOUT：消息发送成功，但是服务器刷盘超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失  
FLUSH\_SLAVE\_TIMEOUT：消息发送成功，但是服务器同步到 Slave 时超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失  
SLAVE\_NOT\_AVAILABLE：消息发送成功，但是此时 slave 不可用，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失。对于精确发送顺序消息的应用，由于顺序消息的局限性，可能会涉及到主备自动切换问题，所以如果sendresult 中的 status 字段不等于 SEND\_OK，就应该尝试重试。对于其他应用，则没有必要这样  
（5）对于消息不可丢失应用，务必要有消息重发机制  
2.消息发送失败处理  
Producer 的 send 方法本身支持内部重试，重试逻辑如下：  
（1） 至多重试 3 次  
（2）  如果发送失败，则轮转到下一个 Broker  
（3）  这个方法的总耗时时间不超过 sendMsgTimeout 设置的值，默认 10s所以，如果本身向 broker 发送消息产生超时异常，就不会再做重试  
如果调用 send 同步方法发送失败，则尝试将消息存储到 db，由后台线程定时重试，保证消息一定到达 Broker。  
上述 db 重试方式为什么没有集成到 MQ 客户端内部做，而是要求应用自己去完成，基于以下几点考虑：  
（1）MQ 的客户端设计为无状态模式，方便任意的水平扩展，且对机器资源的消耗仅仅是 cpu、内存、网络  
（2）如果 MQ 客户端内部集成一个 KV 存储模块，那么数据只有同步落盘才能较可靠，而同步落盘本身性能开销较大，所以通常会采用异步落盘，又由于应用关闭过程不受 MQ 运维人员控制，可能经常会发生 kill  -9 这样暴力方式关闭，造成数据没有及时落盘而丢失  
（3）Producer 所在机器的可靠性较低，一般为虚拟机，不适合存储重要数据。 综上，建议重试过程交由应用来控制。

[https_github.com_alibaba_RocketMQ]: https://github.com/alibaba/RocketMQ/
[http_download.csdn.net_detail_tianwei7518_8137909]: http://download.csdn.net/detail/tianwei7518/8137909