关于rocketmq:RoketMq源码Consumer的构造

DefaultMQPushConsumer的结构

实例代码

        // 设置group
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_consumer_group");
        // 注册生产监听
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                ConsumeConcurrentlyContext context) {
                System.out.println(" Receive New Messages: " + msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    // 启动
    consumer.start();

consumer结构

public DefaultMQPushConsumer(final String namespace, final String consumerGroup, RPCHook rpcHook,
    AllocateMessageQueueStrategy allocateMessageQueueStrategy) {
    this.consumerGroup = consumerGroup;
    this.namespace = namespace;
    // 设置调配策略
    this.allocateMessageQueueStrategy = allocateMessageQueueStrategy;
    defaultMQPushConsumerImpl = new DefaultMQPushConsumerImpl(this, rpcHook);
}

defaultMQPushConsumerImpl的结构

public DefaultMQPushConsumerImpl(DefaultMQPushConsumer defaultMQPushConsumer, RPCHook rpcHook) {
    this.defaultMQPushConsumer = defaultMQPushConsumer;
    this.rpcHook = rpcHook;
    // 有一个pull的产生异样后提早多久再接着解决的一个工夫
    this.pullTimeDelayMillsWhenException = defaultMQPushConsumer.getPullTimeDelayMillsWhenException();
}

总结

consumer结构次要是结构外部的defaultMQPushConsumerImpl，而后须要理解一下调配的策略，有一个细节，就是这个pull过程中产生异样了的一个提早解决的工夫