Azure EventHub Producer 的 TPL .Net 并发问题

Question

我正在使用 Azure 事件中心生产者客户端并从 kafka 流中读取消息，然后将其传递给反序列化/映射，然后传递给事件中心。我有一个消耗循环，它为每个消耗创建一个任务，然后有两种方法进行处理（从 kafka 滞后的角度来看，这似乎大大提高了速度。但是，事件中心让你创建一个我不创建的事件批处理一定要使用。我现在只想一次发送一条消息。为了创建一个新批次，我必须调用 Dispose()。我遇到了一个问题，即另一个函数调用当我调用 Dispose() 时，我收到一条错误消息，指出事件中心正在使用该对象。

我也尝试过使用 eventHubProducerClient.SendAsync 的重载，它允许您传入一个 IEnumerable，但我遇到了同样的问题。

所以我认为这是一个同步问题，或者我可能需要在某处加锁？

任何帮助将不胜感激。

       public void Execute()
                {
                    using (_consumer)
                    {
                        try
                        {
                            _consumer.Subscribe(_streamConsumerSettings.Topic);
                            while (true)
                            {
                                var result = _consumer.Consume(1000);
        
                                if (result == null)
                                {
                                    continue;
                                }
                                var process = Task.Factory.StartNew(() => ProcessMessage(result?.Message?.Value));
                                var send = process.ContinueWith(t => SendMessage(process.Result));                        
                            }
        
                        }
                        catch (ConsumeException e)
                        {
                            _logger.LogError(e, e.StackTrace ?? e.Message);
                            _cancelConsume = true;
                            _consumer.Close();
                            RestartConsumer();
                        }
                    }
                }
    
         public static EquipmentJson ProcessMessage(byte[] result)
         {
                    var json = _messageProcessor.DeserializeAndMap(result);
                    return json;
         }
        
         public static void SendMessage(EquipmentJson message)
         {
                    try 
                    {   
        
                        _eventHubClient.AddToBatch(message);             
                        
                    }
                    catch (Exception e)
                    {
                        _logger.LogError(e, e.StackTrace ?? e.Message);
                    }
          }
    
     
    
        public async Task AddToBatch(EquipmentJson message)
                {
                    if 
      (!string.IsNullOrEmpty(message.EquipmentLocation))
                    {
                        try
                        {
                            var batch = await _equipmentLocClient.CreateBatchAsync();
                            batch.TryAdd(new EventData(Encoding.UTF8.GetBytes(message.EquipmentLocation)));
                            await _eventHubProducerClient.SendAsync(batch);
                            batch.Dispose();
                            _logger.LogInformation($"Data sent {DateTimeOffset.UtcNow}");
                        }
                        catch (Exception e)
                        {
                            _logger.LogError(e, e.StackTrace ?? e.Message);
                        }
                    }
                }

 public class EventHubClient : IEventHubClient
    {
        private readonly ILoggerAdapter<EventHubClient> _logger;
        private readonly EventHubClientSettings _eventHubClientSettings;
        private IMapper _mapper;

        
        private static EventHubProducerClient _equipmentLocClient;


        public EventHubClient(ILoggerAdapter<EventHubClient> logger, EventHubClientSettings eventHubClientSettings, IMapper mapper)
        {
            _logger = logger;
            _eventHubClientSettings = eventHubClientSettings;
            _mapper = mapper;
            _equipmentLocClient = new EventHubProducerClient(_eventHubClientSettings.ConnectionString, _eventHubClientSettings.EquipmentLocation);

        }
    }
}

Answer 1

根据我对 cme​​ts 的推测，我很好奇重构为使用 async/await 而不是主循环中的显式延续可能会有所帮助。也许类似于以下 LinqPad sn-p：

async Task Main()
{
    while (true)
    {
        var message = await Task.Factory.StartNew(() => GetText());
        var events = new[] { new EventData(Encoding.UTF8.GetBytes(message)) };
        
        await Send(events).ConfigureAwait(false);
    }
}

public EventHubProducerClient client = new EventHubProducerClient("<< CONNECTION STRING >>");

public async Task Send(EventData[] events)
{
    try
    {
        await client.SendAsync(events).ConfigureAwait(false);
        "Sent".Dump();
    }
    catch (Exception ex)
    {
        ex.Dump();
    }
}

public string GetText()
{
    Thread.Sleep(250);
    return "Test";
}

如果您打算保持延续，我想知道在延续中进行轻微的结构重构是否会有所帮助，既可以推动事件的创建，也可以兑现 await 声明。也许类似于以下 LinqPad sn-p：

async Task Main()
{
    while(true)
    {
        var t = Task.Factory.StartNew(() => GetText());
        var _ = t.ContinueWith(async q =>
        {
            var events = new[] { new EventData(Encoding.UTF8.GetBytes(t.Result)) };
            await Send(events).ConfigureAwait(false);
        });
        
        await Task.Yield();
    }
}

public EventHubProducerClient client = new EventHubProducerClient("<< CONNECTION STRING >>");

public async Task Send(EventData[] events)
{
    try
    {
        await client.SendAsync(events).ConfigureAwait(false);
        "Sent".Dump();
    }
    catch (Exception ex)
    {
        ex.Dump();
    }
}

public string GetText()
{
    Thread.Sleep(250);
    return "Test";
}