我想确保，如果我的 eventhub 客户端崩溃(目前是一个控制台应用程序)，它只会拾取尚未从 eventhub 获取的事件。实现这一目标的一种方法是利用偏移量。但是，这(据我所知)需要客户端存储最新的偏移量(此外，事件似乎不一定会命中由 SequenceNumber 排序的 ProcessEventsAsync 方法的 foreach 循环)。

另一种方法是使用检查点。我认为它们是使用提供的存储帐户凭据通过服务器 (eventhub) 保留的。这是正确的吗？

这是我目前正在使用的一些初步代码:

public class SimpleEventProcessor : IEventProcessor
{
    private Stopwatch _checkpointStopWatch;

    async Task IEventProcessor.CloseAsync(PartitionContext context, CloseReason reason)
    {
        Console.WriteLine("Processor Shutting Down. Partition '{0}', Reason: '{1}'.", context.Lease.PartitionId, reason);
        if (reason == CloseReason.Shutdown)
        {
            await context.CheckpointAsync();
        }
    }

    Task IEventProcessor.OpenAsync(PartitionContext context)
    {
        Console.WriteLine("SimpleEventProcessor initialized.  Partition: '{0}', Offset: '{1}'", context.Lease.PartitionId, context.Lease.Offset);
        _checkpointStopWatch = new Stopwatch();
        _checkpointStopWatch.Start();
        return Task.FromResult<object>(null);
    }

    async Task IEventProcessor.ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
    {
        foreach (var eventData in messages)
        {
            // do something                    
        }

        //Call checkpoint every 5 minutes, so that worker can resume processing from 5 minutes back if it restarts.
        if (_checkpointStopWatch.Elapsed > TimeSpan.FromMinutes(5))
        {
            await context.CheckpointAsync();
            _checkpointStopWatch.Restart();
        }
    }
}

我相信它会每 5 分钟向服务器发送一个检查点。服务器如何知道哪个客户端提交了检查点(通过上下文)？另外，如果客户端重新启动，如何防止再次处理事件？此外，仍然可能有最多 5 分钟的窗口来再次处理事件。根据我的要求，也许我应该使用队列/主题？

附言:

这似乎就足够了:

async Task IEventProcessor.ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
{
    foreach (var eventData in messages)
    {
        // do something
    }
    await context.CheckpointAsync();
}

最佳答案

Lemme 在回答之前提出一些基本术语:

EventHubs 是高吞吐量的持久事件摄取管道。简而言之 - 它是云上可靠的事件流。

Offset EventData(流中的一个事件)实际上是流上的游标。拥有此游标 - 将启用类似的操作 - 从该游标(又名偏移量)重新开始读取 - 包含或排他。

EventProcessor library 是 EventHubs 团队构建的框架，在 ServiceBus SDK 之上，使“eventhub receiver gu”看起来更容易。 Kafka 的 ZooKeeper <-> 事件中心的 EPH。它将确保在特定分区上运行 EventProcessor 的进程何时死亡/崩溃 - 它将在其他可用的 EventProcessorHost 实例中从最后一个检查点偏移恢复。

CheckPoint:截至今天 - EventHubs 仅支持客户端检查点。当您从客户端代码调用 Checkpoint 时:

await context.CheckpointAsync();

- 它将转换为存储调用(直接来自客户端)- 它将当前偏移量存储在您提供的存储帐户中。 EventHubs 服务不会与存储对话以进行检查。

答案

EventProcessor 框架旨在实现您正在寻找的东西。

检查点不会通过服务器(又名 EVENTHUBS 服务)保留。它纯粹是客户端。你正在与 Azure 存储对话。这就是 EventProcessor 库引入新的附加依赖项的原因 - AzureStorageClient .您可以连接到存储帐户和检查点写入到的容器——我们维护所有权信息——EPH 实例(名称)到他们拥有的 EventHubs 的分区，以及他们当前读取/处理到的检查点。

根据基于计时器的检查点模式 - 您最初拥有 - 如果流程出现故障 - 您将在最后 5 分钟的窗口中重新执行事件。 这是一个健康的模式:

1. 基本假设是故障是罕见的事件 - 所以你 很少会处理重复事件
2. 你最终会赚得更少 调用存储服务(你很容易被 经常检查点)。我会更进一步 实际上，会异步触发检查点调用。处理事件 如果检查点失败，则不必失败!

如果您绝对不想重复任何事件 - 您将需要在下游管道中构建此重复数据删除逻辑。

• 每次 EventProcessorImpl 启动时 - 向下游查询最后一个序列号。它得到并继续丢弃事件，直到当前序列号。

here's more general reading on Event Hubs...

关于c# - 了解 eventhub 中的检查点，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/35464192/