1. 程式人生 > >在C#中使用二叉樹實時計算海量使用者積分排名的實現

在C#中使用二叉樹實時計算海量使用者積分排名的實現

從何說起

前些天和朋友討論一個問題,他們的應用有幾十萬會員然後對應有積分,現在想做積分排名的需求,問有沒有什麼好方案。這個問題也算常見,很多地方都能看到,常規做法一般是資料定時跑批把計算結果到中間表然後直接查表就行,或者只顯示個TOP N的排行榜,名次高的計算真實名次,名次比較低的直接顯示在xxx名開外這種。但是出於探索問題的角度,我還是想找一下有沒有實時計算的辦法,並且效率能夠接受。
在部落格園搜到一篇不錯的文章,基本羅列了常用的方案,每種演算法詳細介紹了具體思路,其中基於二叉樹的演算法是個非常不錯的方案,文章中只給了思路沒有給出程式碼,於是我決定自己用C#實現出來。

這裡只討論具體演算法實現,不考慮業務需求是否合理。


思路解析

關於演算法核心思想前面的文章中寫的很詳細,我不再重複描述,這裡只用一個具體示例演示這個過程。
假設積分範圍是0-5,我們對它不斷進行中位分割槽直到不能分為止,形成如下一棵二叉樹:

其中每個樹節點包含2個資訊:節點範圍range[min,max) 和命中數量計數器count ,可以看到葉子節點的range一定是相鄰的2個數。
假如現在有一個積分3要插入到樹中,該如何操作呢?當前節點從根節點開始,分別判斷是否包含於左右子節點,如果包含的話當前節點改為這個子節點,同時計數器加1,然後再次進行相同判斷,直到遍歷到葉子節點為止,遍歷順序如下:

再依次插入1和4,二叉樹的演變情況為:


資料放進去後怎麼判斷它是排名多少呢?還是從根節點開始,判斷它是否包含於左子節點,如果包含的話說明它比右子節點中count個數小(在count名之外),然後再往下一級做同樣的判斷;如果包含於右子節點那就繼續往下判斷,直到碰到葉子節點為止。依次累加count最後加上葉子節點佔的一位就得到了它在這棵樹裡的排名,以1為例演示判斷步驟(排名為2+1=3):

好了,一切就緒,只欠程式碼。


擼碼實現

樹結構由節點構成,那首先設計一個節點類:

    /// <summary>
    /// 樹節點物件
    /// </summary>
    public class TreeNode
    {
        /// <summary>
        /// 節點的最小值
        /// </summary>
        public int ValueFrom { get; set; }

        /// <summary>
        /// 節點的最大值
        /// </summary>
        public int ValueTo { get; set; }

        /// <summary>
        /// 在節點範圍內的數量
        /// </summary>
        public int Count { get; set; }

        /// <summary>
        /// 節點高度(樹的層級)
        /// </summary>
        public int Height { get; set; }

        /// <summary>
        /// 父節點
        /// </summary>
        public TreeNode Parent { get; set; }

        /// <summary>
        /// 左子節點
        /// </summary>
        public TreeNode LeftChildNode { get; set; }

        /// <summary>
        /// 右子節點
        /// </summary>
        public TreeNode RightChildNode { get; set; }
    }

樹節點的屬性主要包含範圍值ValueFrom、ValueTo、計數器Count、左子節點LeftChildNode和右子節點RightChildNode,由此組成一個有層次的樹結構。
然後就是定義我們的樹物件了,它的核心欄位就是代表源頭的根節點:

    public class RankBinaryTree
    {
        /// <summary>
        /// 根節點
        /// </summary>
        private TreeNode _root;

    }

根據前面的演算法思想,建立樹的時候要用積分範圍初始化所有節點,這裡約定了最小積分為0,通過建構函式傳入最大值並建立樹結構:

        /// <summary>
        /// 建構函式初始化根節點
        /// </summary>
        /// <param name="max"></param>
        public RankBinaryTree(int max)
        {
            _root = new TreeNode() { ValueFrom = 0, ValueTo = max+1, Height = 1 };
            _root.LeftChildNode = CreateChildNode(_root, 0, max / 2);
            _root.RightChildNode = CreateChildNode(_root, max / 2, max);
        }

        /// <summary>
        /// 遍歷建立子節點
        /// </summary>
        /// <param name="current"></param>
        /// <param name="min"></param>
        /// <param name="max"></param>
        /// <returns></returns>
        private TreeNode CreateChildNode(TreeNode current, int min, int max)
        {
            if (min == max) return null;
            var node = new TreeNode() { ValueFrom = min, ValueTo = max, Height = current.Height + 1 };
            node.Parent = current;
            int center = (min + max) / 2;
            if (min < max - 1)
            {
                node.LeftChildNode = CreateChildNode(node, min, center);
                node.RightChildNode = CreateChildNode(node, center, max);
            }
            return node;
        }

有了樹以後下一步就是往裡面插入資料,根據前面介紹的邏輯:

        /// <summary>
        /// 往樹中插入一個值
        /// </summary>
        /// <param name="value"></param>
        public void Insert(int value)
        {
            InnerInsert(_root, value);
            _data.Add(value);
        }

        /// <summary>
        /// 子節點判斷範圍遍歷插入
        /// </summary>
        /// <param name="node"></param>
        /// <param name="value"></param>
        private void InnerInsert(TreeNode node, int value)
        {
            if (node == null) return;
            //判斷是否在這個節點範圍內
            if (value >= node.ValueFrom && value < node.ValueTo)
            {
                //更新節點總數資訊
                node.Count++;
                //更新左子節點
                InnerInsert(node.LeftChildNode, value);
                //更新右子節點
                InnerInsert(node.RightChildNode, value);
            }
        }

下一步提供方法獲取指定值在樹中的排名:

        /// <summary>
        /// 從樹中獲取總排名
        /// </summary>
        /// <param name="value"></param>
        /// <returns></returns>
        public int GetRank(int value)
        {
            if (value < 0) return 0;
            return InnerGet(_root, value);
        }

        /// <summary>
        /// 遍歷子節點獲取累計排名
        /// </summary>
        /// <param name="node"></param>
        /// <param name="value"></param>
        /// <returns></returns>
        private int InnerGet(TreeNode node, int value)
        {
            if (node.LeftChildNode == null || node.RightChildNode == null) return 1;
            if (value >= node.LeftChildNode.ValueFrom && value < node.LeftChildNode.ValueTo)
            {
                //當這個值存在於左子節點中時,要累加右子節點的總數(表示這個數在多少名之後)
                return node.RightChildNode.Count + InnerGet(node.LeftChildNode, value);
            }
            else
            {
                //如果在右子節點中就繼續遍歷
                return InnerGet(node.RightChildNode, value);
            }
        }

到這裡,核心功能已經實現了。考慮到有積分更新的情況,我們可以加上節點更新和刪除的方法。刪除很容易,和插入逆向操作就行,更新就更容易了,把舊節點刪除再計算出新值插入即可,完整程式碼已經上傳到Github。
這棵樹究竟效率如何,下面我們跑個分看看。


測試走起來

在測試程式中,我模擬了積分範圍0-1000000的場景,這個範圍幾乎覆蓋了真實業務中90%的積分值,100萬積分以上的會員系統應該比較少見了。
而會員的積分值分佈也是不均勻的,一般來說擁有小額積分的使用者比例最大,積分值越高所佔使用者比例越小。
在程式中我假設有100萬個會員,其中50W使用者積分都在100以內,30W使用者積分在100-10000,15W使用者積分在10000-50000,5W使用者積分在50000以上。
下面是各個操作的耗時時間:

可以看到,這個效率不是一般的快啊,其中獲取排名的查詢時間幾乎可以忽略不計。
這時候有人問了,這麼多資料會不會非常吃記憶體,下面用工作管理員分別檢視不使用樹和使用樹的記憶體情況:

執行環境是.NetCore3.0 Console,測試主機配置情況:

100萬資料只有130M記憶體佔用,對現代計算機來說簡直是灑灑水~

業務環境中使用務必注意執行緒安全問題!!!


寫在最後

以上的二叉樹演算法處理排名問題確實比較巧妙,實現起來也不算特別複雜,如果上述程式碼有缺陷或有其他更好的方案,歡迎探討,也算拋磚引玉了~

完整程式碼及測試用例請戳這裡https://github.com/hey-hoho/NetCoreDemo/tree/master/ConsoleApp/ScoreRank