Блог

Я пытаюсь заставить service fabric последовательно извлекать сообщения из azure event hub. Кажется, у меня все подключено, но я замечаю, что мой потребитель просто перестает извлекать события.

У меня есть концентратор с парой тысяч событий, которые я к нему подключил. Настроил концентратор с 1 разделом, и у моей службы service fabric также есть только 1 раздел, чтобы упростить отладку.

Служба запускается, создает EventHubClient, оттуда использует его для создания PartitionReceiver. Получатель передается в «EventLoop», который вводит «бесконечный», когда вызывает receiver.ReceiveAsync. Ниже приведен код для EventLoop.

Что я наблюдаю, так это то, что в первый раз в цикле я почти всегда получаю 1 сообщение. Во второй раз я получаю где-то от 103 до 200 сообщений. После этого я не получаю никаких сообщений. Также кажется, что если я перезапущу службу, я снова получу те же сообщения, но это потому, что при перезапуске службы она запускается с начала потока.

Ожидал бы, что это будет продолжаться до тех пор, пока мои 2000 сообщений не будут использованы, а затем он будет ждать меня (опрос случайно).

Есть ли что-то конкретное, что мне нужно сделать с Azure.Обмен сообщениями.Пакет EventHubs 5.3.0, позволяющий ему продолжать извлекать события?

 //Here is how I am creating the EventHubClient:
var connectionString = "something secret";
var connectionStringBuilder = new EventHubsConnectionStringBuilder(connectionString)
{
   EntityPath = "NameOfMyEventHub"
};
try
{
   m_eventHubClient = EventHubClient.Create(connectionStringBuilder);
}

//Here is how I am getting the partition receiver
var receiver = m_eventHubClient.CreateReceiver("$Default", m_partitionId, EventPosition.FromStart());

//The event loop which the receiver is passed to
private async Task EventLoop(PartitionReceiver receiver)
  {
     m_started = true;
     while (m_keepRunning)
     {
        var events = await receiver.ReceiveAsync(m_options.BatchSize, TimeSpan.FromSeconds(5));
        if (events != null) //First 2/3 times events aren't null. After that, always null and I know there are more in the partition/
        {
           var eventsArray = events as EventData[] ?? events.ToArray();
           m_state.NumProcessedSinceLastSave  = eventsArray.Count();

           foreach (var evt in eventsArray)
           {
              //Process the event
              await m_options.Processor.ProcessMessageAsync(evt, null);

              string lastOffset = evt.SystemProperties.Offset;

              if (m_state.NumProcessedSinceLastSave >= m_options.BatchSize)
              {
                 m_state.Offset = lastOffset;
                 m_state.NumProcessedSinceLastSave = 0;
                 await m_state.SaveAsync();
              }
           }
        }
     }

     m_started = false;
  }
 

** РЕДАКТИРОВАТЬ, был задан вопрос о количестве разделов. Концентратор событий имеет один раздел, и служба SF также имеет один раздел.

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

Прослушиватели разделов создаются для каждого раздела. Я получаю разделы, подобные этому:

 public async Task StartAsync()
  {
     // slice the pie according to distribution
     // this partition can get one or more assigned Event Hub Partition ids
     string[] eventHubPartitionIds = (await m_eventHubClient.GetRuntimeInformationAsync()).PartitionIds;
     string[] resolvedEventHubPartitionIds = m_options.ResolveAssignedEventHubPartitions(eventHubPartitionIds);

     foreach (var resolvedPartition in resolvedEventHubPartitionIds)
     {
        var partitionReceiver = new EventHubListenerPartitionReceiver(m_eventHubClient, resolvedPartition, m_options);
        await partitionReceiver.StartAsync();
        m_partitionReceivers.Add(partitionReceiver);
     }
  }
 

Когда partitionListener.Вызывается startAsync, он фактически создает PartitionListener, например, так (на самом деле это немного больше, чем это, но выбранная ветвь — это эта:

 m_eventHubClient.CreateReceiver(m_options.EventHubConsumerGroupName, m_partitionId, EventPosition.FromStart());
 

Спасибо за любые советы.
Будет

Сколько у вас разделов? Я не вижу в вашем коде, как вы убедитесь, что прочитали все разделы в группе потребителей по умолчанию.

Любая конкретная причина, по которой вы используете PartitionReceiver вместо использования EventProcessorHost?

Мне кажется, SF идеально подходит для использования узла обработки событий. Я вижу, что уже существует интегрированное решение SF, которое использует службы с отслеживанием состояния для контрольных точек.