C#實現一個萬物皆可排序的隊列
需求
產品中需要向不同的客戶推送數據，原來的實現是每條數據產生後就立即向客戶推送數據，走的的是HTTP協議。因為每條數據都比較小，而數據生成的頻次也比較高，這就會頻繁的建立HTTP連接，而且每次HTTP傳輸中攜帶的業務數據都很小，對網路的實際利用率不高。希望能夠提高網路的利用率，並降低系統的負載。
分析
一個很自然的想法就是將多條數據一起發送，這裡有幾個關鍵點：
1、多條數據的聚合邏輯： 是攢夠幾條發送，還是按照時間周期發送。如果是攢夠幾條發送，在數據比較稀疏或者產生頻率不那麼穩定的時候，攢夠需要的數據條數可能比較困難，這時候還得需要一個過期時間，因為客戶可能接受不了太多的延遲。既然不管怎樣都需要使用時間進行控制，我這裡索性就選擇按照時間周期發送了。思路是：自上次發送時間起，經過了某個時長之後，就發送客戶在這段時間內產生的所有數據。
2、數據到期判斷方法：既然選擇了按照時間周期發送，那麼就必須有辦法判斷是否到了發送時間。一個很簡單的想法就是輪詢，把所有客戶輪詢一遍，看看誰的數據到期了，就發送誰的。這個演算法的時間複雜度是O(N)，如果客戶比較多，就會消耗過多的時間在這上邊。還有一個辦法：如果客戶按照時間排序好了，那麼只需要取時間最早的客戶的數據時間判斷就好了，滿足就發送，一直向後找，直到獲取的客戶數據時間不符合條件，則退出處理，然後等一會再進行判斷處理。這就需要有一個支援排序的數據結構，寫入數據時自動排序，這種數據結構的時間複雜度一般可以做到O(log(n))。對於這個數據結構的讀寫操作原理上就是隊列的操作方式，只不過是個可排序的隊列。
3、區分客戶：不同客戶的數據接收地址不同，向具體某個客戶發送數據時，應該能比較方便的聚合他的數據，最好是直接就能拿到需要發送的數據。可以使用字典數據結構來滿足這個需求，取某個客戶數據的時間複雜度可以降低到O(1)。
4、數據的安全性問題：如果程式在數據發送成功之前退出了，未發送的數據怎麼辦？是還能繼續發送，還是就丟掉不管了。如果要在程式重啟後恢復未發送成功的數據，則必須將數據同步到別的地方，比如持久化到磁碟。因為我這裡的數據安全性要求不高，丟失一些數據也是允許的，所以要發送的數據收到之後放到記憶體就行了。
實現
上文提到可排序的數據結構，可以使用SortedList<TKey,TValue>，鍵是時間，值是這個時間產生了數據的客戶標識列表。不過它的讀寫操作不是執行緒安全的，需要自己做同步，這裡簡單點就使用lock了。
對於不同客戶的數據，為了方便獲取，使用Dictionary<TKey,TValue>來滿足，鍵是客戶的標識，值是累積的未發送客戶數據。這個數據讀寫也不是執行緒安全的，可以和SortedList的讀寫放到同一個lock中。
下邊是它們的定義：
SortedList<DateTime, List<TKey>> _queue = new SortedList<DateTime, List<TKey>>();
Dictionary<TKey, List<TValue>> _data = new Dictionary<TKey, List<TValue>>();
readonly object _lock = new object();
插入數據的時候，需要先寫入SortedList，然後再寫入Dictionary。程式碼邏輯比較簡單，請看：
public void Publish(TKey key, TValue value)
{
DateTime now = DateTime.Now;
lock (_lock)
{
if (_queue.TryGetValue(now, out List<TKey>? keys))
{
if (!keys!.Contains(key))
{
keys.Add(key);
}
}
else
{
_queue.Add(now, new List<TKey> { key });
}
if (_data.TryGetValue(key, out List<TValue>? values))
{
values.Add(value);
}
else
{
_data.Add(key, new List<TValue> { value });
}
}
}
對於消費數據，這裡採用拉數據的模式。最開始寫的方法邏輯是：讀取一條數據，處理它，然後從隊列中刪除。但是這個邏輯需要對隊列進行讀寫，所以必須加鎖。一般處理數據比較耗時，比如這裡要通過HTTP發送數據，加鎖的話就可能導致寫數據到隊列時阻塞的時間比較長。所以這裡實現的是把可以發送的數據全部提取出來，然後就釋放鎖，數據的處理放到鎖的外部實現，這樣隊列的讀寫性能就比較好了。
public List<(TKey key, List<TValue> value)> Pull(int maxNumberOfMessages)
{
List<(TKey, List<TValue>)> result = new List<(TKey, List<TValue>)>();
DateTime now = DateTime.Now;
lock (_lock)
{
int messageCount = 0;
while (true)
{
if (!_queue.Any())
{
break;
}
var first = _queue.First();
var diffMillseconds = now.Subtract(first.Key).TotalMilliseconds;
if (diffMillseconds < _valueDequeueMillseconds)
{
break;
}
var keys = first.Value;
foreach (var key in keys)
{
if (_data.TryGetValue(key, out List<TValue>? keyValues))
{
result.Add((key, keyValues));
_data.Remove(key);
messageCount += keyValues!.Count;
}
}
_queue.RemoveAt(0);
if (messageCount >= maxNumberOfMessages)
{
break;
}
}
}
return result;
}
這段程式碼比較長一些，我梳理下邏輯：取隊列的第一條數據，判斷時間是否達到發送周期，未達到則直接退出，方法返回空列表。如果達到發送周期，則取出第一條數據中存儲的客戶標識，然後根據這些標識獲取對應的客戶未發送數據，將這些數據按照客戶維度添加到返回列表中，將這些客戶及其數據從隊列中移除，返回有數據的列表。這裡還增加了一個拉取數據的條數限制，方便根據業務實際情況進行控制。
再來看一下怎麼使用這個隊列，這裡模擬多個生產者加一個消費者，其實可以任意多個生產者和消費者：
TimeSortedQueue<string, string> queue = new TimeSortedQueue<string, string>(3000);
List<Task> publishTasks = new List<Task>();
for (int i = 0; i < 4; i++)
{
var j = i;
publishTasks.Add(Task.Factory.StartNew(() =>
{
int k = 0;
while (true)
{
queue.Publish($"key_{k}", $"value_{j}_{k}");
Thread.Sleep(15);
k++;
}
}, TaskCreationOptions.LongRunning));
}
Task.Factory.StartNew(() =>
{
while (true)
{
var list = queue.Pull(100);
if (list.Count <= 0)
{
Thread.Sleep(100);
continue;
}
foreach (var item in list)
{
Console.WriteLine($"{DateTime.Now.ToString("mmss.fff")}:{item.key}, {string.Join(",", item.value)}");
}
}
}, TaskCreationOptions.LongRunning);
Task.WaitAll(publishTasks.ToArray());
以上就是針對這個特定需求實現的一個按照時間進行排序的隊列。
萬物皆可排序的隊列
我們很容易想到，既然可以按照時間排序，那麼按照別的數據類型排序也是可以的。這個數據結構可以應用的場景很多，比如按照權重排序的隊列、按照優先順序排序的隊列、按照年齡排序的隊列、按照銀行存款排序的隊列，等等。這就是一個萬物皆可排序的隊列。
我這裡把主要程式碼貼出來（完整程式碼和示例請看文末）：
public class SortedQueue<TSortKey, TKey, TValue>
where TSortKey : notnull, IComparable
where TKey : notnull
where TValue : notnull
{
Dictionary<TKey, List<TValue>> _data = new Dictionary<TKey, List<TValue>>();
SortedList<TSortKey, List<TKey>> _queue = new SortedList<TSortKey, List<TKey>>();
readonly object _lock = new object();
/// <summary>
/// Create a new instance of SortedQueue
/// </summary>
public SortedQueue(int maxNumberOfMessageConsumedOnce)
{
}
/// <summary>
/// Publish a message to queue
/// </summary>
/// <param name="sortKey">The key in the queue for sorting. Different messages can use the same key.</param>
/// <param name="key">The message key.</param>
/// <param name="value">The message value.</param>
public void Publish(TSortKey sortKey, TKey key, TValue value)
{
lock (_lock)
{
if (_queue.TryGetValue(sortKey, out List<TKey>? keys))
{
keys.Add(key);
}
else
{
_queue.Add(sortKey, new List<TKey> { key });
}
if (_data.TryGetValue(key, out List<TValue>? values))
{
values.Add(value);
}
else
{
_data.Add(key, new List<TValue> { value });
}
}
}
/// <summary>
/// Pull a batch of messages.
/// </summary>
/// <param name="maxNumberOfMessages">The maximum number of pull messages.</param>
/// <returns></returns>
public List<(TKey Key, List<TValue> Value)> Pull(int maxNumberOfMessages)
{
List<(TKey, List<TValue>)> result = new List<(TKey, List<TValue>)>();
lock (_lock)
{
int messageCount = 0;
while (true)
{
if (!_queue.Any())
{
break;
}
var keys = _queue.First().Value;
foreach (var key in keys)
{
if (_data.TryGetValue(key, out List<TValue>? keyValues))
{
result.Add((key, keyValues));
_data.Remove(key);
messageCount += keyValues!.Count;
}
}
_queue.RemoveAt(0);
if (messageCount >= maxNumberOfMessages)
{
break;
}
}
}
return result;
}
}
程式碼邏輯還是比較簡單的，就不羅嗦了，如有問題歡迎留言交流。
再說數據安全
因為在這個實現中所有待處理的數據都在記憶體中，丟失數據會帶來一定的風險，因為我這個程式前邊還有一個隊列，即使程式崩潰了，也只損失沒處理的一小部分數據，業務上可以接受，所以這樣做沒有問題。如果你對這個程式感興趣，需要慎重考慮你的應用場景。
來看看數據丟失可能發生的兩種情況：
一是數據還在隊列中時程式重啟了：對於這種情況，前文提到將數據同步到其它地方，比如寫入Redis、寫入資料庫、寫入磁碟等等。不過因為網路IO、磁碟IO較慢，這往往會帶來吞吐量的大幅下降，想要保證一定的吞吐量，還得引入一些分片機制，又因為分散式的不可靠，可能還得增加一些容錯容災機制，比較複雜，可以參考Kafka。
二是數據處理的時候失敗了：對於這種情況，可以讓程式重試；但是如果異常導致程式崩潰了，數據已經從記憶體或者其它存儲中移除了，數據還是會發生丟失。這時候可以採用一個ACK機制，處理成功後向隊列發送一個ACK，攜帶已經處理的數據標識，隊列根據標識刪除數據。否則消費者還能消費到這些數據。
這些問題並不一定要完全解決，還是得看業務場景，有可能你把數據持久化到Redis就夠了，或者你也不用引入ACK機制，記錄下處理到哪一條了就行了。
以上就是本文的主要內容了，完整程式碼和示例請訪問Github：//github.com/bosima/dotnet-demo/tree/main/CSharp-SortedList
