如何利用.NETCore向Azure EventHubs準實時批量傳送資料？
最近在做一個基於Azure雲的物聯網分析專案：
.netcore採集程式向Azure事件中心(EventHubs)傳送資料，通過Azure EventHubs Capture轉儲到Azure BlogStorage，供資料科學團隊分析。
為什麼使用Azure事件中心？
Azure事件中心是一種Azure上完全託管的實時資料攝取服務， 每秒可流式傳輸來自website、app、device任何源的數百萬個事件。提供的統一流式處理平臺和時間保留緩衝區，將事件生成者和事件使用者分開。
- 事件生成者： 可使用https、AQMP協議釋出事件
- 分割槽：事件中心通過
分割槽使用者模式提供訊息流式處理功能，提高可用性和並行化 - 事件接收者：所有事件中心使用者通過AMQP 1.0會話進行連線，讀取資料
例如，如果事件中心具有四個分割槽，並且其中一個分割槽要在負載均衡操作中從一臺伺服器移動到另一臺伺服器，則仍可以通過其他三個分割槽進行傳送和接收。 此外，具有更多分割槽可以讓更多併發讀取器處理資料，從而提高聚合吞吐量。 瞭解分散式系統中分割槽和排序的意義是解決方案設計的重要方面。 為了幫助說明排序與可用性之間的權衡，請參閱 CAP 定理
最直觀的方式：請在portal.azure.cn門戶站點---->建立事件中心名稱空間---> 建立事件中心
.NetCore 準實時批量傳送資料到事件中心
.NET庫 (Azure.Messaging.EventHubs)
我們使用Asp.NetCore以Azure App Service形式部署，依賴Azure App Service的自動縮放能錄應對物聯網的潮汐大流量。
通常推薦批量傳送到事件中心，能有效增加web服務的吞吐量和響應能力。
目前新版SDk： Azure.Messaging.EventHubs僅支援分批傳送。
- nuget上引入Azure.Messaging.EventHubs庫
EventHubProducerClient客戶端負責分批傳送資料到事件中心，根據傳送時指定的選項，事件資料可能會自動路由到可用分割槽或傳送到特定請求的分割槽。
在以下情況下，建議允許自動路由分割槽：
1） 事件的傳送必須高度可用
2） 事件資料應在所有可用分割槽之間平均分配。
自動路由分割槽的規則:
1）使用迴圈法將事件平均分配到所有可用分割槽中
2）如果某個分割槽不可用，事件中心將自動檢測到該分割槽並將訊息轉發到另一個可用分割槽。
我們要注意，根據選定的 命令空間定價層， 每批次發給事件中心的最大訊息大小也不一樣：
分段批量傳送策略
這裡我們就需要思考： web程式收集資料是以個數為單位； 但是我們分批傳送時要根據分批的位元組大小來切分。
我的方案是： 因引入TPL Dataflow 管道:
- web程式收到資料，立刻丟入
TransformBlock<string, EventData> - 轉換到EventData之後，使用
BatchBlock<EventData>按照個數打包 - 利用
ActionBlock<EventData[]>在包內 累積指定位元組大小批量傳送
- 最後我們設定一個定時器(5min)，強制在BatchBlock的前置佇列未滿時打包，併傳送。
核心的TPL Dataflow程式碼如下：
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;
}
}
總結
- Azure事件中心的基礎用法
- .NET Core準實時分批向Azure事件中心傳送資料，其中用到的TPL Dataflow是以actor模型：提供了粗粒度的資料流和流水線任務，提高了高併發程式的健壯性。