聊聊 Pulsar：Producer 源码解析

@Test
public void testSimpleProducerEvents() throws Exception {
    final String topicName = "persistent://prop/ns-abc/topic0";

    // 1. producer connect
    Producer<byte[]> producer = pulsarClient.newProducer()
        .topic(topicName)
        .enableBatching(false)
        .messageRoutingMode(MessageRoutingMode.SinglePartition)
        .create();

    PersistentTopic topicRef = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topicName).get();
    assertNotNull(topicRef);
    assertEquals(topicRef.getProducers().size(), 1);

    // 2. producer publish messages
    for (int i = 0; i < 10; i++) {
        String message = "my-message-" + i;
        producer.send(message.getBytes());
    }

    rolloverPerIntervalStats();
    assertTrue(topicRef.getProducers().values().iterator().next().getStats().msgRateIn > 0.0);

    // 3. producer disconnect
    producer.close();

    Thread.sleep(ASYNC_EVENT_COMPLETION_WAIT);
    assertEquals(topicRef.getProducers().size(), 0);
}

/**
 * 异步创建消息生产者
 * @param topic 主题名称
 * @param conf 生产者的配置数据
 * @param schema 消息的模式（Schema）
 * @param interceptors 生产者的拦截器列表
 * @return 返回一个 CompletableFuture，表示生产者的异步创建过程。
 * @param <T> 生产者的消息类型
 */
private <T> CompletableFuture<Producer<T>> createProducerAsync(String topic,
                                                               ProducerConfigurationData conf,
                                                               Schema<T> schema,
                                                               ProducerInterceptors interceptors) {
    CompletableFuture<Producer<T>> producerCreatedFuture = new CompletableFuture<>();
    // 取主题的分区元数据
    getPartitionedTopicMetadata(topic).thenAccept(metadata -> {
        if (log.isDebugEnabled()) {
            log.debug("[{}] Received topic metadata. partitions: {}", topic, metadata.partitions);
        }
        // 创建生产
        ProducerBase<T> producer;
        if (metadata.partitions > 0) {
            // 分区主题: 如果 metadata.partitions 大于 0，表示这是一个分区主题，调用 newPartitionedProducerImpl 方法创建分区生产者。
            producer = newPartitionedProducerImpl(topic, conf, schema, interceptors, producerCreatedFuture,
                    metadata);
        } else {
            // 非分区主题: 否则，调用 newProducerImpl 方法创建普通生产者。
            producer = newProducerImpl(topic, -1, conf, schema, interceptors, producerCreatedFuture);
        }
        // 将创建的生产者添加到 producers 集合中。
        producers.add(producer);
    }).exceptionally(ex -> {
        log.warn("[{}] Failed to get partitioned topic metadata: {}", topic, ex.getMessage());
        producerCreatedFuture.completeExceptionally(ex);
        returnnull;
    });

    return producerCreatedFuture;
}

public PartitionedProducerImpl(PulsarClientImpl client, String topic, ProducerConfigurationData conf, int numPartitions,
        CompletableFuture<Producer<T>> producerCreatedFuture, Schema<T> schema, ProducerInterceptors interceptors) {
    super(client, topic, conf, producerCreatedFuture, schema, interceptors);
    this.producers = new ConcurrentOpenHashMap<>();
    this.topicMetadata = new TopicMetadataImpl(numPartitions);
    this.routerPolicy = getMessageRouter();
    stats = client.getConfiguration().getStatsIntervalSeconds() > 0 ? new ProducerStatsRecorderImpl() : null;

    // MaxPendingMessagesAcrossPartitions doesn't support partial partition such as SinglePartition correctly
    // 计算最大待处理消息数: 根据配置计算每个分区的最大待处理消息数。
    int maxPendingMessages = Math.min(conf.getMaxPendingMessages(),
            conf.getMaxPendingMessagesAcrossPartitions() / numPartitions);
    conf.setMaxPendingMessages(maxPendingMessages);

    // 确定要创建的分区索引列表: 根据配置确定要创建的分区索引列表。
    final List<Integer> indexList;
    // 懒启动且共享访问模式: 只创建一个分区的生产者，选择的分区由消息路由策略决定。
    if (conf.isLazyStartPartitionedProducers() &&
            conf.getAccessMode() == ProducerAccessMode.Shared) {
        // try to create producer at least one partition
        indexList = Collections.singletonList(routerPolicy
                .choosePartition(((TypedMessageBuilderImpl<T>) newMessage()).getMessage(), topicMetadata));
    } else {
        // try to create producer for all partitions
        indexList = IntStream.range(0, topicMetadata.numPartitions()).boxed().collect(Collectors.toList());
    }
    // 设置第一个分区索引: 设置第一个分区的索引。
    firstPartitionIndex = indexList.get(0);
    // 启动生产者: 启动指定分区索引列表中的生产者。
    start(indexList);

    // start track and auto subscribe partition increasement
    // 自动更新分区: 如果配置了自动更新分区，则创建监听器并启动定时任务定期检查和更新分区。
    if (conf.isAutoUpdatePartitions()) {
        topicsPartitionChangedListener = new TopicsPartitionChangedListener();
        partitionsAutoUpdateTimeout = client.timer()
            .newTimeout(partitionsAutoUpdateTimerTask, conf.getAutoUpdatePartitionsIntervalSeconds(), TimeUnit.SECONDS);
    }
}

// 键为分区索引，值为 ProducerImpl 实例。
privatefinal ConcurrentOpenHashMap<Integer, ProducerImpl<T>> producers;

private ProducerImpl<T> createProducer(final int partitionIndex) {
    return producers.computeIfAbsent(partitionIndex, (idx) -> {
        String partitionName = TopicName.get(topic).getPartition(idx).toString();
        // 创建一个新的 ProducerImpl 实例，传入分区名称、分区索引、配置、模式、拦截器和一个 CompletableFuture。
        return client.newProducerImpl(partitionName, idx,
                conf, schema, interceptors, new CompletableFuture<>());
    });
}

protected <T> ProducerImpl<T> newProducerImpl(String topic, int partitionIndex,
                                              ProducerConfigurationData conf,
                                              Schema<T> schema,
                                              ProducerInterceptors interceptors,
                                              CompletableFuture<Producer<T>> producerCreatedFuture) {
    returnnew ProducerImpl<>(PulsarClientImpl.this, topic, conf, producerCreatedFuture, partitionIndex, schema,
            interceptors);
}

protected void grabCnx() {
    // 检查客户端连接是否已设置
    if (CLIENT_CNX_UPDATER.get(this) != null) {
        log.warn("[{}] [{}] Client cnx already set, ignoring reconnection request", state.topic, state.getHandlerName());
        return;
    }
    // 检查当前状态是否允许重新连接
    if (!isValidStateForReconnection()) {
        // Ignore connection closed when we are shutting down
        log.info("[{}] [{}] Ignoring reconnection request (state: {})", state.topic, state.getHandlerName(), state.getState());
        return;
    }

    try {
        state.client.getConnection(state.topic) // 获取与指定主题的连接
                .thenAccept(cnx -> connection.connectionOpened(cnx)) // 处理连接打开事件
                .exceptionally(this::handleConnectionError);
    } catch (Throwable t) {
        log.warn("[{}] [{}] Exception thrown while getting connection: ", state.topic, state.getHandlerName(), t);
        reconnectLater(t);
    }
}

@Override
CompletableFuture<MessageId> internalSendWithTxnAsync(Message<?> message, Transaction txn) {
    if (txn == null) {
        return internalSendAsync(message);
    } else {
        return ((TransactionImpl) txn).registerProducedTopic(topic)
                    .thenCompose(ignored -> internalSendAsync(message));
    }
}