c# - Azure EventHubs 抛出异常：端点的至少一个接收器是使用 '0' 的纪元创建的，因此不允许使用非纪元接收器

问题描述

介绍

大家好，我们目前正在开发一个使用 Azure EventHubs 和事件在服务之间发送数据的微服务平台。让我们将这些服务命名为：CustomerService、OrderService 和 MobileBFF。

CustomerService 主要发送更新（带有事件），然后由 OrderService 和 MobileBFF 存储，以便能够响应查询，而无需调用 CustomerService 获取此数据。

所有这 3 项服务 + 我们在 DEV 环境中的开发人员都使用相同的 ConsumerGroup 连接到这些事件中心。

我们目前仅使用 1 个分区，但计划稍后扩展到多个。（您可以看到我们的代码已经可以从多个分区中读取）

例外

不过，我们时不时会遇到一个异常（如果它启动，它通常会持续抛出这个错误一个小时左右）。不过，目前我们只在 DEV/TEST 环境中看到过这个错误。

例外：

Azure.Messaging.EventHubs.EventHubsException(ConsumerDisconnected): At least one receiver for the endpoint is created with epoch of '0', and so non-epoch receiver is not allowed. Either reconnect with a higher epoch, or make sure all epoch receivers are closed or disconnected.

EventHub 的所有消费者都将他们的 SequenceNumber 存储在他们自己的数据库中。这允许我们让每个消费者单独消费事件，并将最后处理的序列号存储在它自己的 SQL 数据库中。当服务（重新）启动时，它从数据库加载 SequenceNumber，然后从这里开始请求事件，直到找不到更多事件。然后它会休眠 100 毫秒，然后重试。这是（有些简化的）代码：

var consumerGroup = EventHubConsumerClient.DefaultConsumerGroupName;
string[] allPartitions = null;
await using (var consumer = new EventHubConsumerClient(consumerGroup, _inboxOptions.EventHubConnectionString, _inboxOptions.EventHubName))
{
    allPartitions = await consumer.GetPartitionIdsAsync(stoppingToken);
}

var allTasks = new List<Task>();

foreach (var partitionId in allPartitions)
{
    //This is required if you reuse variables inside a Task.Run();
    var partitionIdInternal = partitionId;

    allTasks.Add(Task.Run(async () =>
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                await using (var consumer = new EventHubConsumerClient(consumerGroup, _inboxOptions.EventHubConnectionString, _inboxOptions.EventHubName))
                {
                    EventPosition startingPosition;
                    using (var testScope = _serviceProvider.CreateScope())
                    {
                        var messageProcessor = testScope.ServiceProvider.GetService<EventHubInboxManager<T, EH>>();
                        //Obtains starting position from the database or sets to "Earliest" or "Latest" based on configuration
                        startingPosition = await messageProcessor.GetStartingPosition(_inboxOptions.InboxIdentifier, partitionIdInternal);
                    }

                    while (!stoppingToken.IsCancellationRequested)
                    {
                        bool processedSomething = false;
                        await foreach (PartitionEvent partitionEvent in consumer.ReadEventsFromPartitionAsync(partitionIdInternal, startingPosition, stoppingToken))
                        {
                            processedSomething = true;

                            startingPosition = await messageProcessor.Handle(partitionEvent);
                        }

                        if (processedSomething == false)
                        {
                            await Task.Delay(100, stoppingToken);
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                //Log error / delay / retry
            }

        }
    }
}

异常在以下行引发：

await using (var consumer = new EventHubConsumerClient(consumerGroup, _inboxOptions.EventHubConnectionString, _inboxOptions.EventHubName))

c# - Azure EventHubs 抛出异常：端点的至少一个接收器是使用 '0' 的纪元创建的，因此不允许使用非纪元接收器

问题描述

介绍

例外

更多调查

解决方案

推荐阅读