Azure Event Hub：Event Processor Host与Direct Receivers选型及技术咨询

一、Event Processor Host vs Direct Receivers 选择建议 #

没有绝对的“更优”，需结合业务场景判断：

优先选择 Event Processor Host（EPH）的场景 #

• 自动负载均衡：EPH会在多消费者实例间自动分配分区，无需手动管理分区分配逻辑
• 内置 checkpoint 管理：自动记录消费进度，重启后可从上次中断处继续消费，无需自行实现状态存储
• 简化分布式部署：多实例部署时，自动协调实例间的分区租赁，避免重复消费或消息遗漏
• 降低开发复杂度：无需处理故障转移、租赁过期等底层逻辑，可专注于业务消费代码

适合使用 Direct Receivers 的场景 #

• 需完全控制消费逻辑：比如自定义分区分配策略、手动管理 checkpoint，或是有特殊负载均衡需求
• 低延迟要求：EPH的租赁和 checkpoint 机制会带来少量额外开销，对延迟敏感的场景可考虑直接接收器
• 单实例简单消费：单个消费者进程处理所有分区时，Direct Receivers 更轻量，无需依赖Azure存储账户（EPH需要存储账户存储租赁和检查点）

二、Java 消费代码示例 #

1. Event Processor Host 实现（依赖azure-messaging-eventhubs-checkpointstore-blob） #

import com.azure.messaging.eventhubs.*;
import com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore;
import com.azure.storage.blob.BlobContainerClient;
import com.azure.storage.blob.BlobContainerClientBuilder;

public class EventProcessorSample {
    public static void main(String[] args) {
        String eventHubConnStr = "<你的Event Hub连接字符串>";
        String eventHubName = "<你的Event Hub名称>";
        String storageConnStr = "<你的Azure存储账户连接字符串>";
        String containerName = "<存储容器名称>";

        // 初始化Blob容器客户端，用于存储检查点和租赁信息
        BlobContainerClient blobContainerClient = new BlobContainerClientBuilder()
            .connectionString(storageConnStr)
            .containerName(containerName)
            .buildClient();

        // 创建检查点存储
        BlobCheckpointStore checkpointStore = new BlobCheckpointStore(blobContainerClient);

        // 构建并启动事件处理器
        EventProcessorClient processor = new EventProcessorClientBuilder()
            .connectionString(eventHubConnStr, eventHubName)
            .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME)
            .checkpointStore(checkpointStore)
            .processEvent(eventContext -> {
                EventData event = eventContext.getEventData();
                System.out.printf("收到事件：序列号 %s，偏移量 %s，内容：%s%n",
                    event.getSequenceNumber(), event.getOffset(), new String(event.getBody()));

                // 每处理10个事件更新一次检查点
                if (event.getSequenceNumber() % 10 == 0) {
                    eventContext.updateCheckpoint();
                    System.out.println("检查点已更新");
                }
            })
            .processError(errorContext -> {
                System.err.printf("分区 %s 发生错误：%s%n",
                    errorContext.getPartitionContext().getPartitionId(),
                    errorContext.getThrowable().getMessage());
            })
            .buildEventProcessorClient();

        System.out.println("启动事件处理器...");
        processor.start();

        // 保持运行60秒（实际场景中可长期运行）
        try {
            Thread.sleep(60000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        processor.stop();
        System.out.println("事件处理器已停止");
    }
}

2. Direct Receivers 实现（依赖azure-messaging-eventhubs） #

import com.azure.messaging.eventhubs.*;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class DirectReceiverSample {
    public static void main(String[] args) {
        String eventHubConnStr = "<你的Event Hub连接字符串>";
        String eventHubName = "<你的Event Hub名称>";
        String consumerGroup = EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME;

        // 创建Event Hub客户端
        EventHubClient client = new EventHubClientBuilder()
            .connectionString(eventHubConnStr, eventHubName)
            .buildClient();

        // 获取所有分区ID
        List<String> partitionIds = client.getPartitionIds();

        // 为每个分区创建独立的接收器
        for (String partitionId : partitionIds) {
            EventHubConsumerAsyncConsumer consumer = new EventHubClientBuilder()
                .connectionString(eventHubConnStr, eventHubName)
                .consumerGroup(consumerGroup)
                .buildAsyncConsumerClient();

            // 从最新位置开始消费
            consumer.receiveFromPartition(partitionId, EventPosition.latest())
                .subscribe(eventContext -> {
                    EventData event = eventContext.getData();
                    System.out.printf("分区 %s 收到事件：序列号 %s，内容：%s%n",
                        partitionId, event.getSequenceNumber(), new String(event.getBody()));
                }, error -> {
                    System.err.printf("分区 %s 消费出错：%s%n", partitionId, error.getMessage());
                });
        }

        // 保持运行1分钟
        try {
            TimeUnit.MINUTES.sleep(1);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        client.close();
    }
}

三、分区与消费者组是否为更优方案 #

是的，合理使用分区和消费者组是Event Hub消费的最佳实践，核心优势如下：

分区的价值 #

• 水平扩展吞吐量：每个分区是独立的处理单元，增加分区数量可提升Event Hub的整体吞吐量，多个消费者可并行消费不同分区，提高处理效率
• 保证事件顺序：同一分区内的事件按发送顺序消费，若业务需要顺序处理，分区是必须的（跨分区无法保证顺序）
• 故障隔离：单个分区的消费故障不会影响其他分区的正常处理

消费者组的价值 #

• 独立消费视图：每个消费者组拥有独立的消费进度（checkpoint），多个组可同时消费同一Event Hub的事件，互不干扰（比如一个组用于实时分析，另一个组用于数据归档）
• 业务逻辑隔离：不同业务场景的消费者使用不同组，避免相互影响消费进度
• 灵活定制策略：可为特定业务场景创建专属消费者组，定制专属的消费逻辑和进度管理

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