首页 > 其他 > 详细

RocketMQ学习笔记(9)----RocketMQ的Producer 顺序消息

时间:2018-11-13 22:32:06      阅读:232      评论:0      收藏:0      [点我收藏+]

1. 顺序消息原理图

技术分享图片

2. 什么是顺序消息?

  消费消息的顺序要求同发送消息的顺序一致,在RocketMQ中,主要指的是局部顺序,即一类消息为满足顺序性,必须Producer单线程顺序发送,并且发送给到同一队列,这样Consumer就可以按照Producer发送的顺序去消费消息。

  2.1 普通顺序消息

  正常情况下可以保证完全的顺序消息,但是一旦发生通信异常,Broker重启,由于队列总数发生变化,哈希取模后定位的队列会变化,产生短暂的消息顺序不一致。

  如果业务能够容忍在集群异常(如某个Broker宕机或者重启)下,消息短暂乱序,使用普通顺序方式比较合适。

  2.2 严格顺序消息

  无论正常异常都保证顺序,但是牺牲了分布式Failover特性,即Broker集群中只要有一台机器不可用,则整个集群都不可用,服务可用性大大降低。

  如果服务器部署方式为同步双写模式,此缺陷可通过备机自动切换为主避免,不过仍然会存在短暂时间服务不可用。

  目前已知的应用只有数据库binlog同步强依赖严格顺序消息。其他应用绝大部分都可以容忍短暂乱序,推荐使用普通顺序消息。

3. 顺序消息缺陷 

  1. 发送顺序消息无法利用集群FailOver 特性
  2.  消费顺序消息的幵行度依赖亍队列数量
  3.  队列热点问题,个别队列由亍哈希丌均导致消息过多,消费速度跟丌上,产生消息堆积问题
  4.  遇到消息失败的消息,无法跳过,当前队列消费暂停

5. 顺序消息的使用方式

  跟普通消息相比,顺序消息的使用需要在producer的send()方法中添加MessageQueueSelector接口的实现类,并重写select选择使用的队列,因为顺序消息局部顺序,需要将所有消息指定发送到同一队列中。

  Producer实现如下:

package com.wangx.rocketmq.order;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;

public class OrderProducer {

    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        DefaultMQProducer producer = new DefaultMQProducer("my-order-producer-group");

        //2. 设置namesrvAddr,集群环境多个nameserver用;分割

        producer.setNamesrvAddr("47.105.149.61:9876;47.105.145.123:9876");
        //3. 启动
        producer.start();

        for (int i = 0; i < 4; i++) {
            Message message = new Message("OrderTopic", "tabA", ("Hello World " + i).getBytes());
            /**send() 方法中的MessageQueueSelector.select()方法用于指定消息队列,
             * send() 的最后一个参数为队列下表,一个topic默认初始化四个队列
             * select()方法中的第一个参数为队列list,里面存放队列,第二个为send()方法传入的message
             * 第三个参数为send()方法传入的队列下标
             *
             */
            SendResult result = producer.send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    Integer qID = (Integer) o;
                    MessageQueue queue = list.get(qID);
                    return queue;
                }
            },1);
            System.out.println(result);
        }
        producer.shutdown();
    }
}

  与普通消息相比,Consumer的使用只需要在注册监听时使用MessageListenerOrderly对象,并重写consumeMessage即可,需要注意的是consumeMessage的返回值与普通消息不同,具体实现如下:

package com.wangx.rocketmq.order;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class OrderConsumer {

    public static void main(String[] args) throws MQClientException {
        //实例化一个consumer组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my-order-consumer-group");
        //设置setNamesrvAddr,同生产者
        consumer.setNamesrvAddr("47.105.149.61:9876;47.105.145.123:9876");

        //设置消息读取方式,这里设置的是队尾开始读取
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        //设置订阅主题,第二个参数为过滤tabs的条件,可以写为tabA|tabB过滤Tab,*表示接受所有
        consumer.subscribe("OrderTopic", "*");

        //注册消息监听
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                try{
                    MessageExt ext = list.get(0);
                    int quID = ext.getQueueId();
                    String body = new String(ext.getBody(),"UTF-8");
                    System.out.println("quID: " + quID + "接收到消息:" + body);
                }catch (Exception e) {
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
    }
}

  启动consumer,在启动producer,控制台打印如下:

quID: 1接收到消息:Hello World 0
quID: 1接收到消息:Hello World 1
quID: 1接收到消息:Hello World 2
quID: 1接收到消息:Hello World 3

  可以看到所有消息都在队列1中,且都是按照发送消息的顺序进行消费的。

  那么当向多个消息队列中分别顺序发送多条消息时是什么情况呢?

  分别下下标为1,2,3的三个队列中发送四条消息,代码如下:

package com.wangx.rocketmq.order;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;

public class OrderProducer {

    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        DefaultMQProducer producer = new DefaultMQProducer("my-order-producer-group");

        //2. 设置namesrvAddr,集群环境多个nameserver用;分割

        producer.setNamesrvAddr("47.105.149.61:9876;47.105.145.123:9876");
        //3. 启动
        producer.start();

        for (int i = 0; i < 4; i++) {
            Message message = new Message("OrderTopic", "tabA", ("Hello World " + i).getBytes());
            /**send() 方法中的MessageQueueSelector.select()方法用于指定消息队列,
             * send() 的最后一个参数为队列下表,一个topic默认初始化四个队列
             * select()方法中的第一个参数为队列list,里面存放队列,第二个为send()方法传入的message
             * 第三个参数为send()方法传入的队列下标
             *
             */
            SendResult result = producer.send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    Integer qID = (Integer) o;
                    MessageQueue queue = list.get(qID);
                    return queue;
                }
            },1);
            System.out.println(result);
        }
        for (int i = 0; i < 4; i++) {
            Message message = new Message("OrderTopic", "tabA", ("Hello World " + i).getBytes());
            /**send() 方法中的MessageQueueSelector.select()方法用于指定消息队列,
             * send() 的最后一个参数为队列下表,一个topic默认初始化四个队列
             * select()方法中的第一个参数为队列list,里面存放队列,第二个为send()方法传入的message
             * 第三个参数为send()方法传入的队列下标
             *
             */
            SendResult result = producer.send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    Integer qID = (Integer) o;
                    MessageQueue queue = list.get(qID);
                    return queue;
                }
            },2);
            System.out.println(result);
        }
        for (int i = 0; i < 4; i++) {
            Message message = new Message("OrderTopic", "tabA", ("Hello World " + i).getBytes());
            /**send() 方法中的MessageQueueSelector.select()方法用于指定消息队列,
             * send() 的最后一个参数为队列下表,一个topic默认初始化四个队列
             * select()方法中的第一个参数为队列list,里面存放队列,第二个为send()方法传入的message
             * 第三个参数为send()方法传入的队列下标
             *
             */
            SendResult result = producer.send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    Integer qID = (Integer) o;
                    MessageQueue queue = list.get(qID);
                    return queue;
                }
            },3);
            System.out.println(result);
        }
        producer.shutdown();
    }
}

  消费者不变,控制台打印如下:

quID: 1接收到消息:Hello World 0
quID: 1接收到消息:Hello World 1
quID: 1接收到消息:Hello World 2
quID: 1接收到消息:Hello World 3
quID: 2接收到消息:Hello World 0
quID: 2接收到消息:Hello World 1
quID: 2接收到消息:Hello World 2
quID: 2接收到消息:Hello World 3
quID: 3接收到消息:Hello World 0
quID: 3接收到消息:Hello World 1
quID: 3接收到消息:Hello World 2
quID: 3接收到消息:Hello World 3

  这里很明显的看到,每个队列中所有消息均是按照顺序消费的,这里可能消息队列并不一定为1,2,3这样的方式顺序消费(因为每个队列对应单个线程,多个线程在跑时并不一定能够保证队列的顺序),但是针对单个队列中的消息,确实顺序的,这就是rockemq局部顺序的概念的实现。

  

RocketMQ学习笔记(9)----RocketMQ的Producer 顺序消息

原文:https://www.cnblogs.com/Eternally-dream/p/9955179.html

(0)
(0)
   
举报
评论 一句话评论(0
关于我们 - 联系我们 - 留言反馈 - 联系我们:wmxa8@hotmail.com
© 2014 bubuko.com 版权所有
打开技术之扣,分享程序人生!