问题描述
使用EventHubConsumerClient消费Event Hub的消息时，遇见Azure.Messaging.EventHubs.EventHubsException(QuotaExceeded): Exceeded the maximum number of allowed receivers per partition in a consumer group which is 5.”异常。
代码片段为：
... private static String connectionString = "Event Hub Connection ..."; private static String consumerGroup = "xxxxxx"; private static EventHubConsumerClient consumer = new EventHubClientBuilder() .connectionString(connectionString) .consumerGroup(consumerGroup) .buildConsumerClient(); ... while (true) { IterableStream<PartitionEvent> events = consumer.receiveFromPartition(pId,100, sPosition, Duration.ofMillis(1000)); int count = 0; for (PartitionEvent partitionEvent : events) { EventData event = partitionEvent.getData(); ...
错误消息为：
com.azure.core.amqp.exception.AmqpException: Exceeded the maximum number of allowed receivers per partition in a consumer group which is 5.
List of connected receivers - ff7e6a04-43a4-4d50-ae86-e0586297f60c, ff7e6a04-43a4-4d50-ae86-e0586297f60c, ff7e6a04-43a4-4d50-ae86-e0586297f60c, ff7e6a04-43a4-4d50-ae86-e0586297f60c, ff7e6a04-43a4-4d50-ae86-e0586297f60c.
问题解答
这是因为Event Hub对于同一个使用者组的限制：建议在使用者组中一个分区上只有一个活动接收器。 但是，在某些情况下，每个分区最多可以使用 5 个使用者或接收器，其中所有接收器都会获取分区的所有事件。
根据文档中提到的非 Epoch Receiver的限制为5，而如果为每一个Receiver设置上Epoch值，当新的Receiver连接到Event Hub服务器后，服务端会保留最大Epoch值的Receiver，其它旧的则会被逐渐删除。
例如下面的代码就可以解决Azure.Messaging.EventHubs.EventHubsException(QuotaExceeded): Exceeded the maximum number of allowed receivers per partition in a consumer group which is 5.”异常。
import com.azure.messaging.eventhubs.models.ReceiveOptions; ... private static String connectionString = "Event Hub Connection ..."; private static String consumerGroup = "xxxxxx"; private static EventHubConsumerClient consumer = new EventHubClientBuilder() .connectionString(connectionString) .consumerGroup(consumerGroup) .buildConsumerClient(); ... ReceiveOptions receiveOptions ; receiveOptions = new ReceiveOptions(); receiveOptions.setOwnerLevel((new Date()).getTime()); // set epoch value while (true) { IterableStream<PartitionEvent> events = consumer.receiveFromPartition(pId,100, sPosition, Duration.ofMillis(1000),receiveOptions ); int count = 0; for (PartitionEvent partitionEvent : events) { EventData event = partitionEvent.getData(); ...
参考资料
- https://docs.azure.cn/zh-cn/event-hubs/event-hubs-features#consumer-groups
- https://docs.azure.cn/zh-cn/event-hubs/event-hubs-quotas
