美文网首页
RocketMQ源码阅读(十)-Consumer消费消息

RocketMQ源码阅读(十)-Consumer消费消息

作者: _呆瓜_ | 来源:发表于2018-04-10 16:41 被阅读450次

1.消费方式和消费者组

1.消费方式: 拉取和推送两种(事实上所有从远程获取数据都是这两种方式).
2.消费者组与消费模式
多个消费者组成一个消费组, 两种模式: 集群(消息被其中任何一个消息者消费), 广播模式(全部消费者消费).

2.Consumer消费消息的基本流程

RocketMQ 分别使用 DefaultMQPullConsumer 和 DefaultMQPushConsumer 实现了拉取和推送两种方式. 下面主要以DefaultMQPullConsumer为例进行分析.

先看源码中给出的Demo:

public class PullConsumerTest {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5"); //@1
        consumer.start(); //@2

        try {
            MessageQueue mq = new MessageQueue();
            mq.setQueueId(0);
            mq.setTopic("TopicTest3");
            mq.setBrokerName("vivedeMacBook-Pro.local");

            long offset = 26;

            long beginTime = System.currentTimeMillis();
            PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, offset, 32); //@3
            System.out.printf("%s%n", System.currentTimeMillis() - beginTime);
            System.out.printf("%s%n", pullResult);
        } catch (Exception e) {
            e.printStackTrace();
        }

        consumer.shutdown();
    }
}

首先在@1处构建Consumer并且制定其所属的消费者组. 在@2处启动Consumer, 并且在@3处拉取消息.

Consumer启动

事实上DefaultMQPullConsumer将所有操作都委托给DefaultMQPullConsumerImpl, 下面看DefaultMQPullConsumerImpl#start.

public void start() throws MQClientException {
    switch (this.serviceState) {
        case CREATE_JUST:
            this.serviceState = ServiceState.START_FAILED;

            this.checkConfig();  //@1 

            this.copySubscription(); //@2

            if (this.defaultMQPullConsumer.getMessageModel() == MessageModel.CLUSTERING) {
                this.defaultMQPullConsumer.changeInstanceNameToPID(); //@3
            }

            this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPullConsumer, this.rpcHook); //@4

            //@5            
            this.rebalanceImpl.setConsumerGroup(this.defaultMQPullConsumer.getConsumerGroup());
            this.rebalanceImpl.setMessageModel(this.defaultMQPullConsumer.getMessageModel());
            this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPullConsumer.getAllocateMessageQueueStrategy());
            this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);

            this.pullAPIWrapper = new PullAPIWrapper(//
                mQClientFactory, //
                this.defaultMQPullConsumer.getConsumerGroup(), isUnitMode());
            
            //@6            
            this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);
            //@7
            if (this.defaultMQPullConsumer.getOffsetStore() != null) {
                this.offsetStore = this.defaultMQPullConsumer.getOffsetStore();
            } else {
                switch (this.defaultMQPullConsumer.getMessageModel()) {
                    case BROADCASTING:
                        this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPullConsumer.getConsumerGroup());
                        break;
                    case CLUSTERING:
                        this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPullConsumer.getConsumerGroup());
                        break;
                    default:
                        break;
                }
            }

            this.offsetStore.load();

            //@8
            boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPullConsumer.getConsumerGroup(), this);
            if (!registerOK) {
                this.serviceState = ServiceState.CREATE_JUST;

                throw new MQClientException("The consumer group[" + this.defaultMQPullConsumer.getConsumerGroup()
                    + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
                    null);
            }

            //@9
            mQClientFactory.start();
            log.info("the consumer [{}] start OK", this.defaultMQPullConsumer.getConsumerGroup());
            //@10
            this.serviceState = ServiceState.RUNNING;
            break;
        case RUNNING:
        case START_FAILED:
        case SHUTDOWN_ALREADY:
            throw new MQClientException("The PullConsumer service state not OK, maybe started once, "//
                + this.serviceState//
                + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
                null);
        default:
            break;
    }
}

首先判断当前consumer的状态, 除了CREATE_JUST之外, 全部是非法状态.(这个容易理解, 因为时刚刚启动, 不应该处于其他状态).
状态合法后, 大体过程如下:

  • @1校验各配置项是否合法(consumerGroup, allocateMessageQueueStrategy, messageModel)
  • @2将当前defaultMQPullConsumer中的订阅关系复制到当前rebalanceImpl(负载均衡器, 主要负责决定, 当前的consumer应该从哪些Queue中消费消息)中.
  • @3如果是集群模式,则将当前defaultMQPullConsumer实例名改为线程ID.
  • @4实例化MQClientInstance(这个类是一个大杂烩,负责管理client(consumer, producer), 并提供多中功能接口供各个Service(Rebalance, PullMessage等)调用)
  • @5初始化rebalance变量
  • @6初始化pullAPIWrapper(长连接, 负责从broker处拉取消息, 然后利用ConsumeMessageService回调用户的Listener执行消息消费逻辑)
  • @7构建offsetStore消费进度存储对象(有两种实现, Local和Rmote, Local存储在本地磁盘上, 适用于BROADCASTING广播消费模式; 而Remote则将消费进度存储在Broker上, 适用于CLUSTERING集群消费模式).
  • @8向mqClientFactory注册本消费者
  • @9启动mqClientFactory(启动各种定时任务, 如定时获取nameserver地址, 定时清理下线的borker, 启动各种service, 如拉消息服务, 负载均衡服务)
  • @10将serviceState修改为ServiceState.RUNNING

Consumer获取消息

Consumer获取消息使用pullBlockIfNotFound方法, 方法签名如下:

PullResult pullBlockIfNotFound(MessageQueue mq, String subExpression, long offset, int maxNums) 
      throws MQClientException, RemotingException, MQBrokerException, InterruptedException

该方法一共有4个参数.

  • mq, 从哪个队列中拉取消息;
  • subExpression, SubscriptionData中的subString;
  • offset, 消息拉取的offset;
  • maxNums, 最大拉取的消息数目.
    跟踪该方法,最终会委托给DefaultMQPullConsumerImpl中的pullSyncImpl方法执行, 代码如下:
private PullResult pullSyncImpl(MessageQueue mq, String subExpression, long offset, int maxNums, boolean block, long timeout)
    throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
    this.makeSureStateOK();     //@1
    //@2
    if (null == mq) {
        throw new MQClientException("mq is null", null);

    }

    if (offset < 0) {
        throw new MQClientException("offset < 0", null);
    }

    if (maxNums <= 0) {
        throw new MQClientException("maxNums <= 0", null);
    }

    this.subscriptionAutomatically(mq.getTopic()); //@3

    int sysFlag = PullSysFlag.buildSysFlag(false, block, true, false); //@4
    //@5
    SubscriptionData subscriptionData;
    try {
        subscriptionData = FilterAPI.buildSubscriptionData(this.defaultMQPullConsumer.getConsumerGroup(), //
            mq.getTopic(), subExpression);
    } catch (Exception e) {
        throw new MQClientException("parse subscription error", e);
    }

    long timeoutMillis = block ? this.defaultMQPullConsumer.getConsumerTimeoutMillisWhenSuspend() : timeout;   //@6

    PullResult pullResult = this.pullAPIWrapper.pullKernelImpl(//
        mq, 
        subscriptionData.getSubString(), 
        0L, 
        offset, 
        maxNums, 
        sysFlag, 
        0, 
        this.defaultMQPullConsumer.getBrokerSuspendMaxTimeMillis(), 
        timeoutMillis, 
        CommunicationMode.SYNC, 
        null
    );   //@7
    this.pullAPIWrapper.processPullResult(mq, pullResult, subscriptionData);  //@8
    //@9
    if (!this.consumeMessageHookList.isEmpty()) {
        ConsumeMessageContext consumeMessageContext = null;
        consumeMessageContext = new ConsumeMessageContext();
        consumeMessageContext.setConsumerGroup(this.groupName());
        consumeMessageContext.setMq(mq);
        consumeMessageContext.setMsgList(pullResult.getMsgFoundList());
        consumeMessageContext.setSuccess(false);
        this.executeHookBefore(consumeMessageContext);
        consumeMessageContext.setStatus(ConsumeConcurrentlyStatus.CONSUME_SUCCESS.toString());
        consumeMessageContext.setSuccess(true);
        this.executeHookAfter(consumeMessageContext);
    }
    return pullResult;
}

过程概括如下:

  • @1检验当前consumer客户端是否处于RUNNING状态, 否则非法;
  • @2检查mq, offset, maxNums三个参数是否合法;
  • @3构建rebalanceImpl中的SubscriptionData;
  • @4构建sysFlag;
  • @5构建当前consumer的SubscriptionData(这一步和@3有点重复);
  • @6从broker拉取消息时的超时时间;
  • @7从broker拉取消息;
  • @8对pullresult进行处理, 这一步主要进行两个操作, a.更新消息队列拉取消息Broker编号的映射, b.解析消息,并根据订阅信息消息tagCode匹配合适消息.
  • @9如果HookList不为空, 执行HookList中的操作.

相关文章

网友评论

      本文标题:RocketMQ源码阅读(十)-Consumer消费消息

      本文链接:https://www.haomeiwen.com/subject/agrbcftx.html