Я хочу убедиться, что в случае сбоя моего клиента 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();
}
