Pulsar源码解析-客户端-单个消费者ConsumerImpl创建底层实现

Pulsar源码解析-客户端-单个消费者ConsumerImpl创建底层实现

pulsar消费者有多种
ConsumerImpl：一个消费者客户端连接
MultiTopicsConsumerImpl：多个消费者，topic配置有多个，如果开启重试队列也是当前实现，具体看MultiTopicsConsumerImpl源码解析
PatternMultiTopicsConsumerImpl：配置正则表达式的topic，本质也是MultiTopicsConsumerImpl，主要实现了topic的自动发现。
ZeroQueueConsumerImpl：没有缓冲队列的消费者 即receiverQueueSize=0

本篇介绍的是相对最好理解的ConsumerImpl，创建方式：

Consumer subscribe = PulsarClient.builder()
             	.serviceUrl("pulsar://127.0.0.1:6650")
             	.build()
              	.newConsumer()
                .topic("public/default/topic")
                .enableRetry(false)
                ...

重点是：topic(...)只配一个，enableRetry(fasle)

之前介绍过生产者的创建，入口是PulsarClientImpl，消费者也是这里。

消费者的介绍跟以前一样，省略参数校验，日志等代码

一、客户端ConsumerImpl创建入口

public class PulsarClientImpl {

public  CompletableFuture> subscribeAsync(ConsumerConfigurationData conf, Schema schema, ConsumerInterceptors interceptors) {
        // 配置的正则
        if (conf.getTopicsPattern() != null) {
            return patternTopicSubscribeAsync(conf, schema, interceptors);
        } 
      	// 单个消费者
		else if (conf.getTopicNames().size() == 1) {
            return singleTopicSubscribeAsync(conf, schema, interceptors);
        } 
        // 多个
        else {
            return multiTopicSubscribeAsync(conf, schema, interceptors);
        }
    }
}

singleTopicSubscribeAsync(...)中先处理schema，然后调用了doSingleTopicSubscribeAsync(...)
直接看 doSingleTopicSubscribeAsync(...)

public class PulsarClientImpl {

    private  CompletableFuture> doSingleTopicSubscribeAsync(ConsumerConfigurationData conf, Schema schema, ConsumerInterceptors interceptors) {
    
        CompletableFuture> consumerSubscribedFuture = new CompletableFuture<>();

        String topic = conf.getSingleTopic();
        // 获取分区数
        getPartitionedTopicmetadata(topic).thenAccept(metadata -> {
            Consumerbase consumer;
			// 如果是分区topic
            if (metadata.partitions > 0) {
            	// 多个分区创建MultiTopicsConsumerImpl
            	// 当前方法doSingle命名是不是不太准确...我也想不出更好的命名了
                consumer = MultiTopicsConsumerImpl.createPartitionedConsumer(PulsarClientImpl.this, conf,
                        externalExecutorProvider, consumerSubscribedFuture, metadata.partitions, schema, interceptors);
            } else {
            	// 非分区topic，partitionIndex是-1
                int partitionIndex = TopicName.getPartitionIndex(topic);
                consumer = ConsumerImpl.newConsumerImpl(PulsarClientImpl.this, topic, conf, externalExecutorProvider,
                        partitionIndex, false, consumerSubscribedFuture, null, schema, interceptors,
                        true);
            }
            consumers.add(consumer);
        }).exceptionally(ex -> {
            return null;
        });

        return consumerSubscribedFuture;
    }
}

继续ConsumerImpl.newConsumerImpl(...)

二、ConsumerImpl构造

public class ConsumerImpl {

protected ConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurationData conf,
           ExecutorProvider executorProvider, int partitionIndex, boolean hasParentConsumer,
           CompletableFuture> subscribeFuture, MessageId startMessageId,
           long startMessageRollbackDurationInSec, Schema schema, ConsumerInterceptors interceptors,
           boolean createTopicIfDoesNotExist) {
           // 封装消费者常用的几乎所有api
        super(client, topic, conf, conf.getReceiverQueueSize(), executorProvider, subscribeFuture, schema, interceptors);
        // 唯一标识
        this.consumerId = client.newConsumerId();
        // 订阅模式
        this.subscriptionMode = conf.getSubscriptionMode();
        // 批量消息开始id
        this.startMessageId = startMessageId != null ? new BatchMessageIdImpl((MessageIdImpl) startMessageId) : null;
        this.initialStartMessageId = this.startMessageId;
        // 消费者从什么时候开始恢复消息，一般是reader api指定 其它是0
        this.startMessageRollbackDurationInSec = startMessageRollbackDurationInSec;
        // 控制什么时候向服务端发起请求拉取
        AVAILABLE_PERMITS_UPDATeR.set(this, 0);
        // 创建订阅超时控制
        this.subscribeTimeout = System.currentTimeMillis() + client.getConfiguration().getOperationTimeoutMs();
        this.partitionIndex = partitionIndex;
        this.hasParentConsumer = hasParentConsumer;
        // 当AVAILABLE_PERMITS_UPDATER超过当前值 向服务端发起请求拉取
        this.receiverQueueRefillThreshold = conf.getReceiverQueueSize() / 2;
        // 消费者优先级
        this.priorityLevel = conf.getPriorityLevel();
        // 压缩
        this.readCompacted = conf.isReadCompacted();
        // 默认最新的位置
        this.subscriptionInitialPosition = conf.getSubscriptionInitialPosition();
        // 跟踪拒绝的消息，延迟投递
        this.negativeAcksTracker = new NegativeAcksTracker(this, conf);
        // reader api使用
        this.resetIncludeHead = conf.isResetIncludeHead();
        // 自动创建topic
        this.createTopicIfDoesNotExist = createTopicIfDoesNotExist;
        // 分块相关
        this.maxPendingChunkedMessage = conf.getMaxPendingChunkedMessage();
        this.pendingChunkedMessageUuidQueue = new GrowableArrayBlockingQueue<>();
        this.expireTimeOfIncompleteChunkedMessageMillis = conf.getExpireTimeOfIncompleteChunkedMessageMillis();
        this.autoAckOldestChunkedMessageonQueueFull = conf.isAutoAckOldestChunkedMessageOnQueueFull();
        // 用户自己释放消息堆外内存
        this.poolMessages = conf.isPoolMessages();
		// 消费指标打印频率
        if (client.getConfiguration().getStatsIntervalSeconds() > 0) {
            stats = new ConsumerStatsRecorderImpl(client, conf, this);
        } else {
            stats = ConsumerStatsDisabled.INSTANCE;
        }

        duringSeek = new AtomicBoolean(false);
		
		// 未ack和超时消息跟踪
        if (conf.getAckTimeoutMillis() != 0) {
            if (conf.getTickDurationMillis() > 0) {
                this.unAckedMessageTracker = new UnAckedMessageTracker(client, this, conf.getAckTimeoutMillis(),
                        Math.min(conf.getTickDurationMillis(), conf.getAckTimeoutMillis()));
            } else {
                this.unAckedMessageTracker = new UnAckedMessageTracker(client, this, conf.getAckTimeoutMillis());
            }
        } else {
            this.unAckedMessageTracker = UnAckedMessageTracker.UNACKED_MESSAGE_TRACKER_DISABLED;
        }

        // 加密
        if (conf.getCryptoKeyReader() != null) {
            if (conf.getMessageCrypto() != null) {
                this.msgCrypto = conf.getMessageCrypto();
            } else {
                // default to use MessageCryptoBc;
                MessageCrypto msgCryptoBc;
                try {
                    msgCryptoBc = new MessageCryptoBc(
                            String.format("[%s] [%s]", topic, subscription),
                            false);
                } catch (Exception e) {
                    log.error("MessageCryptoBc may not included in the jar. e:", e);
                    msgCryptoBc = null;
                }
                this.msgCrypto = msgCryptoBc;
            }
        } else {
            this.msgCrypto = null;
        }

        if (conf.getProperties().isEmpty()) {
            metadata = Collections.emptyMap();
        } else {
            metadata = Collections.unmodifiableMap(new HashMap<>(conf.getProperties()));
        }
		// 连接管理
        this.connectionHandler = new ConnectionHandler(this,
        		        new BackoffBuilder()
                                .setInitialTime(client.getConfiguration().getInitialBackoffIntervalNanos(), TimeUnit.NANOSECONDS)
                                .setMax(client.getConfiguration().getMaxBackoffIntervalNanos(), TimeUnit.NANOSECONDS)
                                .setMandatoryStop(0, TimeUnit.MILLISECONDS)
                                .create(),
                this);

        this.topicName = TopicName.get(topic);
        // 分组ack
        if (this.topicName.isPersistent()) {
            this.acknowledgmentsGroupingTracker =
                new PersistentAcknowledgmentsGroupingTracker(this, conf, client.eventLoopGroup());
        } else {
            this.acknowledgmentsGroupingTracker =
                NonPersistentAcknowledgmentGroupingTracker.of();
        }
		// 死信队列
        if (conf.getDeadLetterPolicy() != null) {
            possibleSendToDeadLetterTopicMessages = new ConcurrentHashMap<>();
            if (StringUtils.isNotBlank(conf.getDeadLetterPolicy().getDeadLetterTopic())) {
                this.deadLetterPolicy = DeadLetterPolicy.builder()
                        .maxRedeliverCount(conf.getDeadLetterPolicy().getMaxRedeliverCount())
                        .deadLetterTopic(conf.getDeadLetterPolicy().getDeadLetterTopic())
                        .build();
            } else {
                this.deadLetterPolicy = DeadLetterPolicy.builder()
                        .maxRedeliverCount(conf.getDeadLetterPolicy().getMaxRedeliverCount())
                        .deadLetterTopic(String.format("%s-%s" + RetryMessageUtil.DLQ_GROUP_TOPIC_SUFFIX, topic, subscription))
                        .build();
            }

            if (StringUtils.isNotBlank(conf.getDeadLetterPolicy().getRetryLetterTopic())) {
                this.deadLetterPolicy.setRetryLetterTopic(conf.getDeadLetterPolicy().getRetryLetterTopic());
            } else {
                this.deadLetterPolicy.setRetryLetterTopic(String.format("%s-%s" + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX,
                        topic, subscription));
            }

        } else {
            deadLetterPolicy = null;
            possibleSendToDeadLetterTopicMessages = null;
        }

        topicNameWithoutPartition = topicName.getPartitionedTopicName();
		// 重点：之前讲producer也说过改方法
        grabCnx();
    }
}

继续grabCnx()，之前讲producer介绍过，先获取一个连接，然后打开连接创建生产者。
这里同理，先获取连接，然后打开连接创建消费者。
所以直接看打开连接的实现connectionOpened

三、打开连接创建消费者

public class ConsumerImpl {

 public void connectionOpened(final ClientCnx cnx) {
 		...
        long requestId = client.newRequestId();
        int currentSize;
        synchronized (this) {
        	// 等待推送给用户的消息数量
            currentSize = incomingMessages.size();
            // incomingMessages的第一条
            startMessageId = clearReceiverQueue();
        }
		
        boolean isDurable = subscriptionMode == SubscriptionMode.Durable;
        MessageIdData startMessageIdData = null;
        // 如果是持久化，服务端存了消费的位置，不用指定
        if (isDurable) {
            startMessageIdData = null;
        } 
        // 非持久化需要指定初始位置
        else if (startMessageId != null) {
            startMessageIdData = new MessageIdData()
                    .setLedgerId(startMessageId.getLedgerId())
                    .setEntryId(startMessageId.getEntryId())
                    .setBatchIndex(startMessageId.getBatchIndex());
        }
		// 指定恢复的时间，一般是0，reader api指定
        long startMessageRollbackDuration = (startMessageRollbackDurationInSec > 0
                && startMessageId != null && startMessageId.equals(initialStartMessageId)) ? startMessageRollbackDurationInSec : 0;
                // 构建创建消费者请求
        ByteBuf request = Commands.newSubscribe(topic, subscription, consumerId, requestId, getSubType(), priorityLevel,
                consumerName, isDurable, startMessageIdData, metadata, readCompacted,
                conf.isReplicateSubscriptionState(), InitialPosition.valueOf(subscriptionInitialPosition.getValue()),
                startMessageRollbackDuration, si, createTopicIfDoesNotExist, conf.getKeySharedPolicy());
		// 请求创建
        cnx.sendRequestWithId(request, requestId).thenRun(() -> {
            synchronized (ConsumerImpl.this) {
            	// 更新消费者状态ready
                if (changeToReadyState()) {
                    consumerIsReconnectedToBroker(cnx, currentSize);
                } else {
                    setState(State.Closed);
                    deregisterFromClientCnx();
                    client.cleanupConsumer(this);
                    cnx.channel().close();
                    return;
                }
            }
			// 重置
            resetBackoff();
            // 完成
            boolean firstTimeConnect = subscribeFuture.complete(this);
            // 重点
            // firstTimeConnect肯定是true
            // hasParentConsumer单个消费者是false
            // 如果是MultiTopicsConsumerImpl 则内部创建的Consumer都是true
            // 所以这里只有单个消费者的场景和ReceiverQueue不为0的可以进来
            // ReceiverQueueSize默认1000
            if (!(firstTimeConnect && hasParentConsumer) && conf.getReceiverQueueSize() != 0) {
                increaseAvailablePermits(cnx, conf.getReceiverQueueSize());
            }
        }).exceptionally((e) -> {
            return null;
        });
    }
}

继续increaseAvailablePermits

三、重点方法increaseAvailablePermits(ClientCnx currentCnx, int delta)

public class ConsumerImpl {

	// delta上面默认1000带进来
    protected void increaseAvailablePermits(ClientCnx currentCnx, int delta) {
    	// 本身是0，available=1000
        int available = AVAILABLE_PERMITS_UPDATER.addAndGet(this, delta);
		// receiverQueueRefillThreshold是1000/2=500
		// paused是挂起消费者 默认false
		// 如果1000>=500 并且没有挂起消费者
        while (available >= receiverQueueRefillThreshold && !paused) {
        	// 把1000更新成0
            if (AVAILABLE_PERMITS_UPDATER.compareAndSet(this, available, 0)) {
            	// 重点
            	
                sendFlowPermitsToBroker(currentCnx, available);
                break;
            } else {
                available = AVAILABLE_PERMITS_UPDATER.get(this);
            }
        }
    }
}

sendFlowPermitsToBroker的内容：
1、建立长连接，向服务端发送一个拉取请求
2、如果服务端没有消息则挂起请求，消息到来时唤醒并推送
3、上面2的挂起请求就是：上一节讲解生产者发送消息服务端实现时说过一个跟消费者有关的回调。就是那里唤醒消费者的

private void sendFlowPermitsToBroker(ClientCnx cnx, int numMessages) {
        if (cnx != null && numMessages > 0) {
            cnx.ctx().writeAndFlush(Commands.newFlow(consumerId, numMessages), cnx.ctx().voidPromise());
        }
    }

以上比较重要是在最后这个代码increaseAvailablePermits(...)，以后会很常见。
需要记住它的作用。
1、当值大于receiverQueue的一半时向服务端发送一次拉取请求，当前值更新成0
什么时候累加？
当服务端向客户端推送消息时，客户端接收到后调用用户的监听回调或者放入等待队列用户主动获取后+1
这个实现后面讲解消费者服务端推送客户端的实现时介绍

什么时候减？
没有减哈，当值大于receiverQueue的一半时直接更新成0

默认值1000
例如，启动一个消费者，发起请求后，服务端推送1000条到客户端，客户端不停的消费，剩500时再发起拉取请求。目前这个阈值控制是代码固定死的。

以上就是单个消费者创建的过程。