將大於256KB的事件推送到Azure EventHub

Question

我正在使用Azure事件中心。我打算髮送活動並使用SendBatchAsync。我看到事件中心有256KB的限制（無論是單獨或批量發送）

所以，如果我的數據> 256KB，什麼是最佳實踐來處理呢？我們是否應該單獨發送消息（這將保證它是< 256KB）？

另外，如何將事件分成256KB塊併發送到Event Hub？ 我看了看Azure文檔，看到他們推薦使用EventHubClient.CreateBatch但我沒有看到足夠的示例。有人可以在我們將如何最終分裂爲256KB和SendBatchAsync

在這裏需要一些樣機或樣品或步驟提供是我做的（但不因子256KB限制）

await myEventHubClient.SendBatchAsync(
    events.Select( 
     iEvent => 
      new EventData(
       Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(iEvent))))); 

Answer 1

不要輕易嘗試並且將多個消息中的單個有效負載分塊，考慮到事件不能保證按順序到達，可能最終在不同的分區中並且可能到達一定的時間間隔，並且可以獨立於「總線」。這在規模上變得更具挑戰性。

選項1：如果您的有效負載是text/json，並且您的消息大小不超過256kb，請考慮壓縮有效負載。您可以從this online tool獲得尺寸結果的概念。如果您的有效負載中有大量空白（例如JSON），則也可以使用minify而不是壓縮。

選項2：寧可將您的有效內容存儲在外部存儲中（比如DocumentDB或Blob存儲）。

在事件中發送對您的有效負載的引用。這使您的事件保持精簡，但您的消費者需要了解如何檢索有效負載。一種簡單的方法是將有效負載的鏈接作爲URI呈現，顯然，您可能需要考慮有效負載存儲上的身份驗證和授權，以符合您的Event Hub訪問策略，以保持其光滑。

Answer 2

下面是如何使用發送批次不超過256KB限制的示例。 代碼來自這個回購協議（paolosalvatori/ServiceBusExtensions）

/// <summary> 
/// This class contains extensions methods for the <see cref="EventHubClient"/> class. 
/// </summary> 
public static class EventHubClientExtensions 
{ 
    private const string EventDataListCannotBeNullOrEmpty = "The eventDataEnumerable parameter cannot be null or empty."; 
    private const string SendPartitionedBatchFormat = "[EventHubClient.SendPartitionedBatch] Batch Sent: BatchSizeInBytes=[{0}] MessageCount=[{1}]"; 
    private const string SendPartitionedBatchAsyncFormat = "[EventHubClient.SendPartitionedBatchAsync] Batch Sent: BatchSizeInBytes=[{0}] MessageCount=[{1}]"; 
    private const int MaxBathSizeInBytes = 262144; 

    /// <summary> 
    /// Asynchronously sends a batch of event data to the same partition. 
    /// All the event data in the batch need to have the same value in the Partitionkey property. 
    /// If the batch size is greater than the maximum batch size, 
    /// the method partitions the original batch into multiple batches, 
    /// each smaller in size than the maximum batch size. 
    /// </summary> 
    /// <param name="eventHubClient">The current <see cref="EventHubClient"/> object.</param> 
    /// <param name="messages">An IEnumerable object containing event data instances.</param> 
    /// <param name="trace">true to cause a message to be written; otherwise, false.</param> 
    /// <returns>The asynchronous operation.</returns> 
    public static async Task SendPartitionedBatchAsync(this EventHubClient eventHubClient, IEnumerable<EventData> messages, bool trace = false) 
    { 
     var eventDataList = messages as IList<EventData> ?? messages.ToList(); 
     if (messages == null || !eventDataList.Any()) 
     { 
      throw new ArgumentNullException(EventDataListCannotBeNullOrEmpty); 
     } 

     var batchList = new List<EventData>(); 
     long batchSize = 0; 

     foreach (var eventData in eventDataList) 
     { 
      if ((batchSize + eventData.SerializedSizeInBytes) > MaxBathSizeInBytes) 
      { 
       // Send current batch 
       await eventHubClient.SendBatchAsync(batchList); 
       Trace.WriteLineIf(trace, string.Format(SendPartitionedBatchAsyncFormat, batchSize, batchList.Count)); 

       // Initialize a new batch 
       batchList = new List<EventData> { eventData }; 
       batchSize = eventData.SerializedSizeInBytes; 
      } 
      else 
      { 
       // Add the EventData to the current batch 
       batchList.Add(eventData); 
       batchSize += eventData.SerializedSizeInBytes; 
      } 
     } 
     // The final batch is sent outside of the loop 
     await eventHubClient.SendBatchAsync(batchList); 
     Trace.WriteLineIf(trace, string.Format(SendPartitionedBatchAsyncFormat, batchSize, batchList.Count)); 
    } 

    /// <summary> 
    /// Asynchronously sends a batch of event data to the same partition. 
    /// All the event data in the batch need to have the same value in the Partitionkey property. 
    /// If the batch size is greater than the maximum batch size, 
    /// the method partitions the original batch into multiple batches, 
    /// each smaller in size than the maximum batch size. 
    /// </summary> 
    /// <param name="eventHubClient">The current <see cref="EventHubClient"/> object.</param> 
    /// <param name="messages">An IEnumerable object containing event data instances.</param> 
    /// <param name="trace">true to cause a message to be written; otherwise, false.</param> 
    public static void SendPartitionedBatch(this EventHubClient eventHubClient, IEnumerable<EventData> messages, 
     bool trace = false) 
    { 
     var eventDataList = messages as IList<EventData> ?? messages.ToList(); 
     if (messages == null || !eventDataList.Any()) 
     { 
      throw new ArgumentNullException(EventDataListCannotBeNullOrEmpty); 
     } 

     var batchList = new List<EventData>(); 
     long batchSize = 0; 

     foreach (var eventData in eventDataList) 
     { 
      if ((batchSize + eventData.SerializedSizeInBytes) > MaxBathSizeInBytes) 
      { 
       // Send current batch 
       eventHubClient.SendBatch(batchList); 
       Trace.WriteLineIf(trace, string.Format(SendPartitionedBatchAsyncFormat, batchSize, batchList.Count)); 

       // Initialize a new batch 
       batchList = new List<EventData> { eventData }; 
       batchSize = eventData.SerializedSizeInBytes; 
      } 
      else 
      { 
       // Add the EventData to the current batch 
       batchList.Add(eventData); 
       batchSize += eventData.SerializedSizeInBytes; 
      } 
     } 
     // The final batch is sent outside of the loop 
     eventHubClient.SendBatch(batchList); 
     Trace.WriteLineIf(trace, string.Format(SendPartitionedBatchFormat, batchSize, batchList.Count)); 
    } 
}