如何配置EventProcessorClient仅读取Azure Event Hub的特定分区事件

结论 #

EventProcessorClient 的设计定位是分布式多实例消费场景，自动实现分区负载均衡、故障转移和checkpoint管理，原生未提供直接限制仅读取指定分区的配置项。如果你的场景需要固定消费指定分区，有两种可行实现方案：

方案1：自定义EventProcessorClient分区分配器 #

如果必须保留EventProcessorClient的checkpoint管理能力，可以通过自定义分区分配器仅给当前实例分配目标分区，示例代码如下：

// 定义需要消费的目标分区ID集合
List<String> targetPartitions = List.of("0", "3");

EventProcessorClient eventProcessorClient = new EventProcessorClientBuilder()
        .connectionString(connectionString, eventHubName)
        .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME)
        .processEvent(IncomingEventProcessor.processEvent)
        .processError(processError)
        .checkpointStore(new BlobCheckpointStore(blobContainerAsyncClient))
        // 自定义分区分配逻辑，仅返回目标分区
        .partitionAssigner(() -> new PartitionAssigner() {
            @Override
            public Map<String, List<String>> assign(PartitionAssignmentContext context) {
                return Collections.singletonMap(context.getOwnership().getOwnerId(), targetPartitions);
            }
        })
        .buildEventProcessorClient();

注意：该方案仅适用于单实例消费场景，多实例部署会出现分区所有权冲突，导致消费异常。

方案2：使用EventHubConsumerClient直接消费指定分区（更推荐） #

如果不需要分布式负载均衡能力，直接用单分区专用消费客户端是性能最优、逻辑最简单的方案，不需要依赖Blob存储做分区分配，示例代码如下：

EventHubConsumerClient consumerClient = new EventHubClientBuilder()
        .connectionString(connectionString, eventHubName)
        .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME)
        .buildConsumerClient();

// 指定分区ID和消费起始位置
String partitionId = "1";
consumerClient.receiveFromPartition(
        partitionId,
        EventPosition.latest(),
        IncomingEventProcessor.processEvent,
        processError
);

内容的提问来源于stack exchange，提问作者Ziggy000