在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

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。