Понимание контрольной точки в eventhub

Я хочу убедиться, что в случае сбоя моего клиента Eventhub (в настоящее время это консольное приложение), он получает только те события, которые еще не получены из Eventhub. Один из способов добиться этого - использовать смещения. Однако это (на мой взгляд) требует, чтобы клиент сохранял последнее смещение (кроме того, что события не обязательно попадают в цикл foreach метода ProcessEventsAsync, упорядоченного по SequenceNumber).

Альтернативой является использование контрольных точек. Я думаю, что они сохраняются через сервер (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 минут, в котором события обрабатываются снова. Возможно, мне лучше использовать очередь/тему, учитывая мои требования?

PS:

Это кажется достаточным:

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

Ответы

Ответ 1

Лемм выдвинул несколько основных терминов, прежде чем ответить:

EventHubs - это высокопроизводительный конвейер приема долговременных событий. Проще говоря - это надежный поток событий в облаке.

Смещение на EventData (одно событие в потоке) буквально является курсором в потоке. Наличие этого курсора - включит такие операции, как - возобновить чтение с этого курсора (он же Offset) - включительно или эксклюзивно.

Библиотека EventProcessor - это фреймворк, созданный командой EventHubs поверх пакета ServiceBus SDK для создания "приемника событийного гу" - выглядеть проще. ZooKeeper для Kafka <-> EPH для Event Hub. Он удостоверится, что процесс, запускающий EventProcessor на определенном разделе, умирает/падает - он будет возобновлен с последнего смещения Checkpointed - в другом доступном экземпляре EventProcessorHost.

CheckPoint: на сегодняшний день - EventHubs поддерживает только проверку на стороне клиента. Когда вы звоните в Checkpoint с вашего кода клиента:

await context.CheckpointAsync();

- он преобразуется в вызов хранилища (напрямую от клиента), который будет хранить текущее смещение в предоставленной вами учетной записи хранения. Сервис EventHubs не будет связываться с хранилищем для проверки чека.

ОТВЕТ

EventProcessor Framework предназначен для достижения именно того, что вы ищете.

Контрольные точки не сохраняются через сервер (он же EVENTHUBS Service). Это чисто на стороне клиента. Вы разговариваете с хранилищем Azure. По этой причине библиотека EventProcessor вносит новую дополнительную зависимость - AzureStorageClient. Вы можете подключиться к учетной записи хранения и к контейнеру, в который записаны контрольные точки - мы сохраняем информацию о владельце - экземпляры (имена) EPH к разделам концентраторов EventHub, которыми они владеют, и к какой контрольной точке они в настоящее время считываются/обрабатываются до тех пор.

В соответствии с шаблоном проверки контрольных точек на основе таймера - у вас изначально было - если Процесс остановится - вы будете заново делать события в последнем 5-минутном окне. Это здоровый образец, как:

  1. фундаментальное предположение состоит в том, что неисправности являются редкими событиями, поэтому вы будете иметь дело с дублирующимися событиями
  2. в конечном итоге вы будете делать меньше звонков в службу хранилища (что вы можете легко переполнить, часто проверяя). Я бы сделал еще один шаг и фактически произвел бы вызов контрольной точки асинхронно. OnProcessEvents не нужно проваливать, если контрольная точка терпит неудачу!

если вы хотите, чтобы абсолютно без событий повторялись - вам нужно будет построить эту логику дедупликации в нисходящем конвейере.

  • каждый раз, когда запускается EventProcessorImpl - запрашивать у вашего нисходящего потока последнюю последовательность нет. он получил и продолжает отбрасывать события до текущей последовательности нет.

здесь более общее чтение на Event Hubs...