RocketMQ集群消息测试（二）

最近两天发生了很多事情，李咏（勇）去世，金庸讣告，重庆公交车坠江，印尼狮航JT610航班坠毁。让我感觉生命实在是脆弱，有人寿终正寝，有人患病离世，这些都是可预见的。但是像公交坠江，飞机坠海这类的就是不可预见的。他们也许正在去上班的路上，也许正在去参加朋友聚会的路上，也许家中有正等待他们的亲人，可是突然一次事故。将他们与亲人，朋友永远的分开了。所以活着的人除了为他们惋惜，还应该珍惜当下，珍惜每一天，每一刻。我们有理由，有时间去学习。

废话不多说了，上篇文章搭建了一个两主两从异步的RocketMQ集群环境，这篇文章主要对这个集群环境进行可用性，可靠性的测试。我们仍然暂时不去讨论源码实现。

场景一：验证Producer发送消息是负载均衡的。

测试代码：我使用send(msg)方法发送消息，没有自定义MessageQueueSelector，一共发送了12条消息。

public class Producer {
    public static void main(String[] args) throws UnsupportedEncodingException {
        try {
            MQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
            ((DefaultMQProducer) producer).setNamesrvAddr("192.168.1.85:9876;192.168.99:9876");
            producer.start();

            for (int i = 0; i < 12; i++) {
                Message msg = new Message("TopicTest", "TagA", "KEY" + i,
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.send(msg);

                System.out.println(sendResult);
            }

            producer.shutdown();
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}

我们看一下Broker情况，此时broker-a和broker-b各有6条消息，并且成功把消息同步到slave。

场景二：消费测试，代码如下

public class Consumer {

    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");
        consumer.setNamesrvAddr("192.168.1.85:9876;192.168.99:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("TopicTest", "TagA");

        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                context.setAutoCommit(false);
                System.out.println(Thread.currentThread().getName()+"："+msgs);
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}

Consumer采用Push方式，其实RocketMQ的所谓推模式本质上也是对Pull模式的封装，这里先不去讨论。

Broker初始状态：

如下图：

1、启动Consumer

2、启动Producer，发送12条消息，此时看到broker中Today Produce Count加了6，Today Consume Count也加了6。说明消息都被消费了。

 场景三：如果两个broker master都挂掉，consumer还会拉取消息吗？

从上图可以看到，即使两个broker master都宕机了，并不影响consumer继续从slave拉取消息。只是此时无法写写入消息了。

接下来我们将两台master服务启动，看看会发生什么

和预想的效果一样，消费端继续从master拉取消息，这里有一个关键的问题，就是明显消息被重复消费了，RocketMQ之所以允许重复消费是因为master挂掉的概率很小，如果在broker里做消息去重操作，将会影响整个Broker的吞吐量，所以重复消费问题肯定需要业务方自己解决了。

场景四：如果两个namesrv都挂了，对Producer和Consumer有什么影响呢？

这个就不贴图，图片也反映不出来真实情况，其实真实情况是这样的。我将两个namesrv服务停止之后，原来已经建立连接Producer和Consumer依然可以正常工作，只是新加入的Producer和Consumer无法工作了。

场景五：两个slave都挂掉了，会有什么现象？

如上图：首先将Producer停止，不再发送新的消息，然后启动broker-a slave服务。刚启动时produce count数量是0。

如下图：继续观察，发现produce count数量开始上升，说明slave启动之后，master开始将消息向slave同步。

重点：通过上面测试，大家可能注意到开源版本的RocketMQ，Broker的master和slave是不能自动切换的。一旦master挂了，这条线路也就不能写消息了。根据Producer轮询规则，所有的消息都会发送到另一条Broker线路上。但是其实阿里自己内部使用的RocketMQ版本是支持master和slave自动切换的。如果你们用阿里云服务，也能享受到这个待遇。