RocketMq部署(三)

环境准备

一台win7

一台centos7

基础环境备注能够参看前面的文章

部署2m-noslave

修改配置文件

win7 上

C:\Windows\System32\drivers\etc 修改host文件，添加

10.10.10.190 rocketmq-master-2

10.10.10.190 rocketmq-nameserver-2

10.10.10.108 rocketmq-master-1

10.10.10.108 rocketmq-nameserver-1

centos7

vi /etc/hosts

添加如win7上同样的内容

启动环境

win7上

cmd启动nameserver

F:\rocketmq\bin>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. The Name Server boot success. serializeType=JSON

启动broker-b

F:\rocketmq\bin>mqbroker -c F:\rocketmq\conf\2m-noslave\broker-b.properties

启动成功

2018-03-07 11:33:33 INFO BrokerControllerScheduledThread1 - dispatch behind commit log 0 bytes

2018-03-07 11:33:33 INFO BrokerControllerScheduledThread1 - Slave fall behind master: 1255662 bytes

2018-03-07 11:33:37 INFO BrokerControllerScheduledThread1 - register broker to name server rocketmq-nameserver-1:9876 OK

2018-03-07 11:33:37 INFO BrokerControllerScheduledThread1 - register broker to name server rocketmq-nameserver-2:9876 OK

2018-03-07 11:34:07 INFO BrokerControllerScheduledThread1 - register broker to name server rocketmq-nameserver-1:9876 OK

2018-03-07 11:34:07 INFO BrokerControllerScheduledThread1 - register broker to name server rocketmq-nameserver-2:9876 OK

centos7

启动nameserver

[root@Dralen apache-rocketmq]# nohup sh bin/mqnamesrv &

启动broker-a

[root@Dralen apache-rocketmq]# sh bin/mqbroker -c /soft/rocketmq-all-4.2.0/distribution/target/apache-rocketmq/conf/2m-noslave/broker-a.properties

启动日志

[root@Dralen 2m-noslave]# tail -f ~/logs/rocketmqlogs/broker.log

2018-03-07 11:34:52 INFO BrokerControllerScheduledThread1 - register broker to name server rocketmq-nameserver-1:9876 OK

2018-03-07 11:34:52 INFO BrokerControllerScheduledThread1 - register broker to name server rocketmq-nameserver-2:9876 OK

2018-03-07 11:35:22 INFO BrokerControllerScheduledThread1 - register broker to name server rocketmq-nameserver-1:9876 OK

2018-03-07 11:35:22 INFO BrokerControllerScheduledThread1 - register broker to name server rocketmq-nameserver-2:9876 OK

2018-03-07 11:35:49 INFO BrokerControllerScheduledThread1 - dispatch behind commit log 0 bytes

2018-03-07 11:35:49 INFO BrokerControllerScheduledThread1 - Slave fall behind master: 521133 bytes

2018-03-07 11:35:52 INFO BrokerControllerScheduledThread1 - register broker to name server rocketmq-nameserver-1:9876 OK

2018-03-07 11:35:52 INFO BrokerControllerScheduledThread1 - register broker to name server rocketmq-nameserver-2:9876 OK

2018-03-07 11:36:22 INFO BrokerControllerScheduledThread1 - register broker to name server rocketmq-nameserver-1:9876 OK

2018-03-07 11:36:22 INFO BrokerControllerScheduledThread1 - register broker to name server rocketmq-nameserver-2:9876 OK

在多Master模式中，若是某个Master进程挂了，显然这台broker将不可用，上面的消息也将没法消费，要知道开源版本的RocketMQ是没有提供切换程序，来自动恢复故障的，所以在实际开发中，咱们通常提供一个监听程序，用于监控Master的状态。

** 上面的消费者采用的是Push Consumer的方式，那么监听的Listener中的消息List究竟是多少条呢？虽然提供了API，如consumer.setConsumeMessageBatchMaxSize(10)，实际上即便设置了批量的条数，可是注意了，是最大是10，并不意味着每次batch的都是10，只有在消息有挤压的状况下才有可能。并且Push Consumer的最佳实践方式就是一条条的消费，若是须要batch，可使用Pull Consumer。 **

pull Consumer模式消费

务必保证先启动消费者进行Topic订阅，而后在启动生产者进行生产（不然极有可能致使消息的重复消费，重复消费，重复消费！重要的事情说三遍！关于消息的重复问题后续给你们介绍~）。并且在实际开发中，有时候不会批量的处理消息，而是原子性的，单线程的去一条一条的处理消息，这样就是实时的在处理消息。（批量的处理海量的消息，能够考虑Kafka）

初步了解消息失败重试机制

消息失败，无非涉及到2端：从生产者端发往MQ的失败；消费者端从MQ消费消息的失败；

生产消息 发送失败重试

消费者消费消息重试

在启动broker的过程当中，能够观察下日志，你会发现RECONSUME_LATER的策略。

若是消费失败，那么1S后再次消费，若是失败，那么5S后，再次消费，......直至2H后若是消费还失败，那么该条消息就会终止发送给消费者了！

RocketMQ为咱们提供了这么屡次数的失败重试，可是在实际中也许咱们并不须要这么多重试，好比重试3次，尚未成功，咱们但愿把这条消息存储起来并采用另外一种方式处理，并且但愿RocketMQ不要在重试呢，由于重试解决不了问题了！

注意到reconsumeTimes属性，这个属性就表明消息重试的次数！

注意了，对于消费消息而言，存在2种指定的状态（成功 OR 失败重试），若是一条消息在消费端处理没有返回这2个状态，那么至关于这条消息没有达到消费者，势必会再次发送给消费者！也便是消息的处理必须有返回值，不然就进行重发。

web管理控制台

集群消费 AND 广播消费

RocketMQ的消费方式有2种，在默认状况下，就是集群消费。另外一种消费模式，是广播消费。广播消费，相似于ActiveMQ中的发布订阅模式，消息会发给Consume Group中的每个消费者进行消费。

//消息消费方式 默认集群消费 还有一种是广播消费
    consumer.setMessageModel(MessageModel.BROADCASTING);//广播消费

这种模式就是会重复，消费消息

同一台机器，一个消费者群组，2个消费者

一个生产者