快捷導(dǎo)航

C#實現(xiàn)萬物皆可排序的隊列方法詳解

更新時間：2022年07月04日 08:08:55 作者：波斯馬

本文詳細(xì)講解了C#實現(xiàn)萬物皆可排序隊列的方法，文中通過示例代碼介紹的非常詳細(xì)。對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下

需求

產(chǎn)品中需要向不同的客戶推送數(shù)據(jù)，原來的實現(xiàn)是每條數(shù)據(jù)產(chǎn)生后就立即向客戶推送數(shù)據(jù)，走的的是HTTP協(xié)議。因為每條數(shù)據(jù)都比較小，而數(shù)據(jù)生成的頻次也比較高，這就會頻繁的建立HTTP連接，而且每次HTTP傳輸中攜帶的業(yè)務(wù)數(shù)據(jù)都很小，對網(wǎng)絡(luò)的實際利用率不高。希望能夠提高網(wǎng)絡(luò)的利用率，并降低系統(tǒng)的負(fù)載。

分析

一個很自然的想法就是將多條數(shù)據(jù)一起發(fā)送，這里有幾個關(guān)鍵點：

1、多條數(shù)據(jù)的聚合邏輯： 是攢夠幾條發(fā)送，還是按照時間周期發(fā)送。如果是攢夠幾條發(fā)送，在數(shù)據(jù)比較稀疏或者產(chǎn)生頻率不那么穩(wěn)定的時候，攢夠需要的數(shù)據(jù)條數(shù)可能比較困難，這時候還得需要一個過期時間，因為客戶可能接受不了太多的延遲。既然不管怎樣都需要使用時間進行控制，我這里索性就選擇按照時間周期發(fā)送了。思路是：自上次發(fā)送時間起，經(jīng)過了某個時長之后，就發(fā)送客戶在這段時間內(nèi)產(chǎn)生的所有數(shù)據(jù)。

2、數(shù)據(jù)到期判斷方法：既然選擇了按照時間周期發(fā)送，那么就必須有辦法判斷是否到了發(fā)送時間。一個很簡單的想法就是輪詢，把所有客戶輪詢一遍，看看誰的數(shù)據(jù)到期了，就發(fā)送誰的。這個算法的時間復(fù)雜度是O(N)，如果客戶比較多，就會消耗過多的時間在這上邊。還有一個辦法：如果客戶按照時間排序好了，那么只需要取時間最早的客戶的數(shù)據(jù)時間判斷就好了，滿足就發(fā)送，一直向后找，直到獲取的客戶數(shù)據(jù)時間不符合條件，則退出處理，然后等一會再進行判斷處理。這就需要有一個支持排序的數(shù)據(jù)結(jié)構(gòu)，寫入數(shù)據(jù)時自動排序，這種數(shù)據(jù)結(jié)構(gòu)的時間復(fù)雜度一般可以做到O(log(n))。對于這個數(shù)據(jù)結(jié)構(gòu)的讀寫操作原理上就是隊列的操作方式，只不過是個可排序的隊列。

3、區(qū)分客戶：不同客戶的數(shù)據(jù)接收地址不同，向具體某個客戶發(fā)送數(shù)據(jù)時，應(yīng)該能比較方便的聚合他的數(shù)據(jù)，最好是直接就能拿到需要發(fā)送的數(shù)據(jù)?？梢允褂米值鋽?shù)據(jù)結(jié)構(gòu)來滿足這個需求，取某個客戶數(shù)據(jù)的時間復(fù)雜度可以降低到O(1)。

4、數(shù)據(jù)的安全性問題：如果程序在數(shù)據(jù)發(fā)送成功之前退出了，未發(fā)送的數(shù)據(jù)怎么辦？是還能繼續(xù)發(fā)送，還是就丟掉不管了。如果要在程序重啟后恢復(fù)未發(fā)送成功的數(shù)據(jù)，則必須將數(shù)據(jù)同步到別的地方，比如持久化到磁盤。因為我這里的數(shù)據(jù)安全性要求不高，丟失一些數(shù)據(jù)也是允許的，所以要發(fā)送的數(shù)據(jù)收到之后放到內(nèi)存就行了。

實現(xiàn)

上文提到可排序的數(shù)據(jù)結(jié)構(gòu)，可以使用SortedList<TKey,TValue>，鍵是時間，值是這個時間產(chǎn)生了數(shù)據(jù)的客戶標(biāo)識列表。不過它的讀寫操作不是線程安全的，需要自己做同步，這里簡單點就使用lock了。

對于不同客戶的數(shù)據(jù)，為了方便獲取，使用Dictionary<TKey,TValue>來滿足，鍵是客戶的標(biāo)識，值是累積的未發(fā)送客戶數(shù)據(jù)。這個數(shù)據(jù)讀寫也不是線程安全的，可以和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();

插入數(shù)據(jù)的時候，需要先寫入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 });
            }
        }
    }

對于消費數(shù)據(jù)，這里采用拉數(shù)據(jù)的模式。最開始寫的方法邏輯是：讀取一條數(shù)據(jù)，處理它，然后從隊列中刪除。但是這個邏輯需要對隊列進行讀寫，所以必須加鎖。一般處理數(shù)據(jù)比較耗時，比如這里要通過HTTP發(fā)送數(shù)據(jù)，加鎖的話就可能導(dǎo)致寫數(shù)據(jù)到隊列時阻塞的時間比較長。所以這里實現(xiàn)的是把可以發(fā)送的數(shù)據(jù)全部提取出來，然后就釋放鎖，數(shù)據(jù)的處理放到鎖的外部實現(xiàn)，這樣隊列的讀寫性能就比較好了。

    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;
    }

這段代碼比較長一些，我梳理下邏輯：取隊列的第一條數(shù)據(jù)，判斷時間是否達到發(fā)送周期，未達到則直接退出，方法返回空列表。如果達到發(fā)送周期，則取出第一條數(shù)據(jù)中存儲的客戶標(biāo)識，然后根據(jù)這些標(biāo)識獲取對應(yīng)的客戶未發(fā)送數(shù)據(jù)，將這些數(shù)據(jù)按照客戶維度添加到返回列表中，將這些客戶及其數(shù)據(jù)從隊列中移除，返回有數(shù)據(jù)的列表。這里還增加了一個拉取數(shù)據(jù)的條數(shù)限制，方便根據(jù)業(yè)務(wù)實際情況進行控制。

再來看一下怎么使用這個隊列，這里模擬多個生產(chǎn)者加一個消費者，其實可以任意多個生產(chǎn)者和消費者：

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());

以上就是針對這個特定需求實現(xiàn)的一個按照時間進行排序的隊列。

萬物皆可排序的隊列
我們很容易想到，既然可以按照時間排序，那么按照別的數(shù)據(jù)類型排序也是可以的。這個數(shù)據(jù)結(jié)構(gòu)可以應(yīng)用的場景很多，比如按照權(quán)重排序的隊列、按照優(yōu)先級排序的隊列、按照年齡排序的隊列、按照銀行存款排序的隊列，等等。這就是一個萬物皆可排序的隊列。

我這里把主要代碼貼出來（完整代碼和示例請看文末）：

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;
    }
}

代碼邏輯還是比較簡單的，就不羅嗦了，如有問題歡迎留言交流。

再說數(shù)據(jù)安全

因為在這個實現(xiàn)中所有待處理的數(shù)據(jù)都在內(nèi)存中，丟失數(shù)據(jù)會帶來一定的風(fēng)險，因為我這個程序前邊還有一個隊列，即使程序崩潰了，也只損失沒處理的一小部分?jǐn)?shù)據(jù)，業(yè)務(wù)上可以接受，所以這樣做沒有問題。如果你對這個程序感興趣，需要慎重考慮你的應(yīng)用場景。

來看看數(shù)據(jù)丟失可能發(fā)生的兩種情況：

一是數(shù)據(jù)還在隊列中時程序重啟了：對于這種情況，前文提到將數(shù)據(jù)同步到其它地方，比如寫入Redis、寫入數(shù)據(jù)庫、寫入磁盤等等。不過因為網(wǎng)絡(luò)IO、磁盤IO較慢，這往往會帶來吞吐量的大幅下降，想要保證一定的吞吐量，還得引入一些分片機制，又因為分布式的不可靠，可能還得增加一些容錯容災(zāi)機制，比較復(fù)雜，可以參考Kafka。

二是數(shù)據(jù)處理的時候失敗了：對于這種情況，可以讓程序重試；但是如果異常導(dǎo)致程序崩潰了，數(shù)據(jù)已經(jīng)從內(nèi)存或者其它存儲中移除了，數(shù)據(jù)還是會發(fā)生丟失。這時候可以采用一個ACK機制，處理成功后向隊列發(fā)送一個ACK，攜帶已經(jīng)處理的數(shù)據(jù)標(biāo)識，隊列根據(jù)標(biāo)識刪除數(shù)據(jù)。否則消費者還能消費到這些數(shù)據(jù)。

這些問題并不一定要完全解決，還是得看業(yè)務(wù)場景，有可能你把數(shù)據(jù)持久化到Redis就夠了，或者你也不用引入ACK機制，記錄下處理到哪一條了就行了。

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: