本地 Azure Event Hubs emulator 的消费者为何收不到消息？

2026-11-04数据小陈

优先检查 Spring 消费端的连接配置，以及监听器是否已启动。ProducerService 中的连接字符串只用于手动创建的 EventHubProducerClient，不会自动传给 @EventHubsListener。

现有代码未展示 application.yml、checkpoint store 或监听器启动日志，暂时无法确定唯一根因。控制台没有报错，也不能证明消息已发送并被消费者读取。

1. 为 Spring 监听器配置模拟器连接

如果尚未配置，可以在 application.yml 中添加：

spring:
  cloud:
    azure:
      eventhubs:
        connection-string: "Endpoint=sb://localhost;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;"

spring.cloud.azure.eventhubs.connection-string 是 Spring Event Hubs 集成使用的连接属性。即使手动创建的 producer 已经能够连接，监听器仍需要自己的基础设施配置。Spring Messaging Event Hubs 配置说明

上述 localhost 适用于应用运行在宿主机、模拟器端口已映射到宿主机的情况。如果应用也运行在 Docker 容器中，localhost 指向应用容器自身。此时，同一 Docker 网络内通常应使用模拟器服务名或网络别名，例如 eventhubs-emulator。还需检查 AMQP 端口映射，官方部署示例使用 5672:5672。模拟器连接方式

2. 核对 Event Hub、consumer group 和监听器注册

检查模拟器实际加载的配置，确认其中存在 eh1，并核对 consumer group 名称。如果使用默认组，先将代码中的 $default 改为 SDK 默认常量，排除名称拼写差异：

import com.azure.messaging.eventhubs.EventHubClientBuilder;
import com.azure.spring.messaging.eventhubs.implementation.core.annotation.EventHubsListener;
import org.springframework.stereotype.Service;

@Service
public class ConsumerService {

    @EventHubsListener(
        destination = "eh1",
        group = EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME
    )
    public void handleMessageFromEventHub(String message) {
        System.out.printf("New message received: %s%n", message);
    }
}

默认组的标准名称是 $Default。仅凭原代码中的大小写差异，还不能断定它就是故障原因，应结合模拟器配置和服务端错误判断。

还要确认 ConsumerService 位于 Spring 组件扫描范围内，且对应的 listener container 已创建并启动。官方文档说明，较新的 Spring Cloud Azure 自动配置可以自动启用 @EnableAzureMessaging。对于当前 6.1.0 项目，应先检查自动配置条件是否满足。如果使用了自定义配置或排除了相关自动配置，再核对是否需要显式启用。监听器启用说明

3. 检查 checkpoint store 是否可用

Spring 监听器使用的处理器还涉及分区协调和 checkpoint store。官方集成示例包含 azure-messaging-eventhubs-checkpointstore-blob 依赖及存储配置。如果缺少这些配置，或存储连接失败，需要先解决处理器启动或分区协调错误。官方依赖与配置示例

如果采用 Blob checkpoint store，可通过匹配的 BOM 管理依赖版本：

implementation platform('com.azure.spring:spring-cloud-azure-dependencies:6.1.0')
implementation 'com.azure.spring:spring-messaging-azure-eventhubs'
implementation 'com.azure.spring:spring-cloud-azure-starter'
implementation 'com.azure:azure-messaging-eventhubs-checkpointstore-blob'

如果项目已有统一的依赖管理，应在现有配置中补充所需依赖。

使用 Azurite 作为 checkpoint store 时，可以在前面的 Event Hubs 配置下补充：

spring:
  cloud:
    azure:
      eventhubs:
        connection-string: "Endpoint=sb://localhost;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;"
        processor:
          checkpoint-store:
            connection-string: ${AZURITE_STORAGE_CONNECTION_STRING}
            container-name: eventhubs-checkpoints
            create-container-if-not-exists: true

AZURITE_STORAGE_CONNECTION_STRING 需要替换为应用实际可访问的 Azurite Storage 连接字符串，其中应包含正确的账号、密钥和 Blob endpoint。这里使用的是 Storage 连接字符串，用途不同于 Event Hubs 的连接字符串。checkpoint store 配置属性

即使模拟器启动了 Azurite，Spring 应用也不会因此自动获得它的连接配置。如果已提供自定义 CheckpointStore，则检查该实现及其注入情况。

4. 确认发送方法确实执行，并检查 tryAdd

代码中定义了 send()，但没有展示调用位置。先确认是否调用了该方法，再检查事件是否成功加入 batch：

public void send(String message) {
    EventDataBatch batch = producer.createBatch();
    if (!batch.tryAdd(new EventData(message))) {
        throw new IllegalArgumentException("Message does not fit in an empty batch");
    }

    producer.send(batch);
    log.info("Event Hubs send completed, eventCount={}", batch.getCount());
}

tryAdd() 返回 false 表示事件未加入 batch，需要处理这个结果。同步 send() 正常返回也不代表消费回调已经执行。EventDataBatch API

5. 用独立 SDK 消费者定位故障范围

如果监听器仍然没有输出，可以先停止 Spring 消费者，再运行下面的独立诊断程序。它会枚举所有分区，从各分区当前保留的最早事件开始读取，避免只检查一个分区而漏掉消息：

import com.azure.messaging.eventhubs.EventHubClientBuilder;
import com.azure.messaging.eventhubs.EventHubConsumerClient;
import com.azure.messaging.eventhubs.models.EventPosition;
import com.azure.messaging.eventhubs.models.PartitionEvent;
import java.time.Duration;

public class EventHubsReadCheck {
    public static void main(String[] args) {
        String connectionString =
            "Endpoint=sb://localhost;"
            + "SharedAccessKeyName=RootManageSharedAccessKey;"
            + "SharedAccessKey=SAS_KEY_VALUE;"
            + "UseDevelopmentEmulator=true;";

        try (EventHubConsumerClient consumer = new EventHubClientBuilder()
                .connectionString(connectionString, "eh1")
                .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME)
                .buildConsumerClient()) {

            for (String partitionId : consumer.getPartitionIds()) {
                for (PartitionEvent event : consumer.receiveFromPartition(
                        partitionId,
                        100,
                        EventPosition.earliest(),
                        Duration.ofSeconds(5))) {
                    System.out.printf(
                        "partition=%s, body=%s%n",
                        partitionId,
                        event.getData().getBodyAsString());
                }
            }
        }
    }
}

这是一次有限读取，每个分区最多读取 100 条，等待时间为 5 秒。它用于检查已保留的事件，并非持续监听程序。直接使用 EventHubConsumerClient 无需配置 Blob checkpoint store。同步消费者 API

根据结果继续排查：

独立消费者能读到消息：重点检查 Spring 监听器注册、连接属性、checkpoint store 和消息转换错误。
获取分区列表失败：先检查地址、端口、模拟器状态、连接字符串及实际解析到的 SDK 依赖。
能获取分区但读不到消息：确认 send() 已执行、batch 非空，以及生产者和诊断消费者连接的是同一个模拟器中的 eh1。

恢复 Spring 监听器后，等待其完成启动，再发送一条带唯一标识的新消息。如果处理器已有 checkpoint，或初始位置配置为只读取新事件，之前发送的消息可能不会触发当前回调。排查时同时查看 Spring、Event Hubs SDK 和模拟器日志中的完整异常，尤其是连接失败、实体不存在及 checkpoint store 访问错误。