如何利用.NETCore向Azure EventHubs准实时批量发送数据？

public class MsgBatchSender
    {
        private readonly EventHubProducerClient Client;
        private readonly TransformBlock<string, EventData> _transformBlock;
        private readonly BatchBlock<EventData> _packer;
        private readonly ActionBlock<EventData[]> _batchSender;
        private readonly DataflowOption _dataflowOption;
        private readonly Timer _trigger;
        private readonly ILogger _logger;
        public MsgBatchSender(EventHubProducerClient client, IOptions<DataflowOption> option,ILoggerFactory loggerFactory)
        {
            Client = client;
            _dataflowOption = option.Value;
            var dfLinkoption = new DataflowLinkOptions { PropagateCompletion = true };
            _transformBlock = new TransformBlock<string, EventData>(
                text => new EventData(Encoding.UTF8.GetBytes(text)),
                   new ExecutionDataflowBlockOptions
                   {
                       MaxDegreeOfParallelism = _dataflowOption.MaxDegreeOfParallelism
                   });
            _packer = new BatchBlock<EventData>(_dataflowOption.BatchSize);
            _batchSender = new ActionBlock<EventData[]>(msgs=> BatchSendAsync(msgs));
            _packer.LinkTo(_batchSender, dfLinkoption);
            _transformBlock.LinkTo(_packer, dfLinkoption, x => x != null);
            _trigger = new Timer(_ => _packer.TriggerBatch(), null, TimeSpan.Zero, TimeSpan.FromSeconds(_dataflowOption.TriggerInterval));
            _logger = loggerFactory.CreateLogger<DataTrackerMiddleware>();
        }
        private async Task BatchSendAsync(EventData[] msgs)
        {
            try
            {
                if (msgs != null)
                {
                    var i = 0;
                    while (i < msgs.Length)
                    {
                        var batch = await Client.CreateBatchAsync();
                        while (i < msgs.Length)
                        {
                            if (batch.TryAdd(msgs[i++]) == false)
                            {
                                break;
                            }
                        }
                        if(batch!= null && batch.Count>0)
                        {
                            await Client.SendAsync(batch);
                            batch.Dispose();
                        }
                    }
                }
            }
             catch (Exception ex)
            {
                // ignore and log any exception
                _logger.LogError(ex, "SendEventsAsync: {error}", ex.Message);
            }
        }
        public  async Task<bool> PostMsgsync(string txt)
        {
            return await _transformBlock.SendAsync(txt);
        }
        public async Task CompleteAsync()
        {
            _transformBlock.Complete();
            await _transformBlock.Completion;
            await _batchSender.Completion;
            await _batchSender.Completion;
        }
    }