手机网游 排行榜（网游手游排行榜第一）

先介绍下基本需求

1）游戏不分区服

2）每周刷新一些挑战副本，玩家完成副本，根据完成情况计算出得分，根据得分进行排名统计

3）玩家每天挑战副本的次数有限制

4）排行榜实时刷新，前100名记录玩家的详细信息，其余只显示排名以及排名的百分比

5）高效查询玩家自身排名

再来分析一下需求

1）游戏不分服，单一排行榜的数据量至少是百万级别，按千万数据量设计

2）副本挑战按1分钟每局计算，在百万同时在线的情况下，每个榜每秒钟更新量为1.6W/s左右。为了应对突发高峰，按10W+/s的标准设计。

3）对于前N名玩家，排名应保证完全正确，而N名以后的玩家，只要保证分数高的玩家必定比分数低的玩家排名高即可，具体是排10000还是10100没有太大关系。

思路1：redis

redis的ZSet可用于实现排行榜，并且性能不错，能满足我对更新的需求，但是，记录详细信息的小榜还是要单独维护，为了降低复杂性我希望在一个统一的进程内维护大榜和小榜，因此排除此方案。

思路2：数组

用有序数组存放rankItem,用hashmap记录id与rankItem的关系。

玩家第一次进榜时，通过2分查找在数组中查找合适的插入位置，将rankItem插入，积分变更后通过id找到rankItem,顺着当前位置向前比较(只有积分>原积分才更新排行榜)并交换位置。

package array import ( //"fmt" ) const maxItemCount int = 100000 type rankItem struct { score int id int idx int } type span struct { data []*rankItem } func newSpan() *span { s := &span{ data: make([]*rankItem, 0, maxItemCount), } return s } func (s *span) check() bool { if len(s.data) > 0 { max := s.data[0].score for _, v := range s.data { if v.score > max { return false } else { max = v.score } } } return true } //2分法查找插入点 func (s *span) binarySearch(score int, left int, right int) int { if left >= right { return left } mIdx := (right-left)/2 + left m := s.data[mIdx] if m.score > score { return s.binarySearch(score, mIdx+1, right) } else { pIdx := mIdx - 1 if pIdx < 0 || s.data[pIdx].score > score { return mIdx } return s.binarySearch(score, left, mIdx-1) } } //向span添加一个元素，如果添加后超过maxItemCount,将最小元素踢出并返回 func (s *span) add(item *rankItem) *rankItem { score := item.score if len(s.data) == cap(s.data) && score < s.data[len(s.data)-1].score { return item } else { var kick *rankItem var idx int if len(s.data) == 0 || score >= s.data[0].score { idx = 0 } else if score < s.data[len(s.data)-1].score { idx = len(s.data) } else { idx = s.binarySearch(score, 0, len(s.data)-1) } if len(s.data) == cap(s.data) { kick = s.data[len(s.data)-1] } else { s.data = s.data[:len(s.data)+1] } for i := len(s.data) - 1; i > idx; i-- { s.data[i] = s.data[i-1] s.data[i].idx = i } item.idx = idx s.data[idx] = item return kick } } func (s *span) update(item *rankItem) { for i := item.idx - 1; i >= 0; i-- { if item.score >= s.data[i].score { old := s.data[i] s.data[i] = item s.data[item.idx] = old old.idx = item.idx item.idx = i } } } type Rank struct { id2Item map[int]*rankItem spans *span } func NewRank() *Rank { return &Rank{ id2Item: map[int]*rankItem{}, spans: newSpan(), } } func (r *Rank) getRankItem(id int) *rankItem { return r.id2Item[id] } //返回排名和排名百分比 func (r *Rank) UpdateScore(id int, score int) { item := r.getRankItem(id) if nil == item { item = &rankItem{} r.id2Item[id] = item item.id = id item.score = score r.spans.add(item) } else { if item.score >= score { return } item.score = score r.spans.update(item) } } func (r *Rank) Check() bool { return r.spans.check() }

上面是实现代码，在10W数据规模下执行20万次更新，前10W次是首次插入，后10W次是积分变更，总耗时75.72秒，平均2641/s。查询到是非常快，直接返回rankItem的下标即可。

数组实现效率如此低的原因在于，新插入元素的时候，需要移动插入位置后面的所有元素，为新元素空出位置。当积分变更的时候，元素需要执行大量的冒泡比较和交换。

为了减少移动和交换的次数，可以考虑采用2级数组的方式实现。

第1级数组存放rankItem,但只存放很小数量的rankItem,例如上限100个。第2级为rankItem数组的数组。

插入时，通过score，采用2分查找在2级数组中检索出合适的一级数组，然后再将item插入到1级数组中。如果插入后1级数组超过容量限制，则将最小的元素弹出，新建一个1级数组，将溢出的元素插入到新的1数组中，然后再将新的1级数组插入到2级数组中。

实现如下

import ( //"fmt" ) const realRankCount int = 1000 const maxItemCount int = 100 type rankItem struct { score int id int idx int s *span } type span struct { max int min int idx int data []*rankItem } func newSpan(idx int) *span { s := &span{ idx: idx, data: make([]*rankItem, 0, maxItemCount), } return s } func (s *span) check(max int) int { if len(s.data) > 0 { for _, v := range s.data { if v.score > max { return -1 } else { max = v.score } } } return max } //2分法查找插入点 func (s *span) binarySearch(score int, left int, right int) int { if left >= right { return left } mIdx := (right-left)/2 + left m := s.data[mIdx] if m.score > score { return s.binarySearch(score, mIdx+1, right) } else { pIdx := mIdx - 1 if pIdx < 0 || s.data[pIdx].score > score { return mIdx } return s.binarySearch(score, left, mIdx-1) } } //向span添加一个元素，如果添加后超过maxItemCount,将最小元素踢出并返回 func (s *span) add(item *rankItem) *rankItem { score := item.score if len(s.data) == cap(s.data) && score < s.data[len(s.data)-1].score { return item } else { var kick *rankItem var idx int if len(s.data) == 0 || score >= s.data[0].score { idx = 0 } else if score < s.data[len(s.data)-1].score { idx = len(s.data) } else { idx = s.binarySearch(score, 0, len(s.data)-1) } if len(s.data) == cap(s.data) { kick = s.data[len(s.data)-1] } else { s.data = s.data[:len(s.data)+1] } for i := len(s.data) - 1; i > idx; i-- { s.data[i] = s.data[i-1] s.data[i].idx = i } item.idx = idx s.data[idx] = item item.s = s s.fixMinMax() return kick } } func (s *span) update(item *rankItem) { for i := item.idx - 1; i >= 0; i-- { if item.score >= s.data[i].score { old := s.data[i] s.data[i] = item s.data[item.idx] = old old.idx = item.idx item.idx = i } } s.fixMinMax() } func (s *span) remove(item *rankItem) { for i := item.idx + 1; i < len(s.data); i++ { s.data[i-1] = s.data[i] s.data[i-1].idx = i - 1 } s.data = s.data[:len(s.data)-1] item.s = nil s.fixMinMax() } func (s *span) fixMinMax() { if len(s.data) > 0 { s.max = s.data[0].score s.min = s.data[len(s.data)-1].score } else { s.max = 0 s.min = 0 } } type Rank struct { id2Item map[int]*rankItem spans []*span } func NewRank() *Rank { return &Rank{ id2Item: map[int]*rankItem{}, spans: make([]*span, 0, 8192), } } func (r *Rank) getRankItem(id int) *rankItem { return r.id2Item[id] } func (r *Rank) findSpan(score int) *span { var c *span if len(r.spans) == 0 { c = newSpan(len(r.spans)) r.spans = append(r.spans, c) } else { c = r.binarySearch(score, 0, len(r.spans)-1) } return c } func (r *Rank) binarySearch(score int, left int, right int) *span { if left >= right { return r.spans[left] } mIdx := (right-left)/2 + left m := r.spans[mIdx] if m.max > score { nIdx := mIdx + 1 if nIdx >= len(r.spans) || r.spans[nIdx].max < score { return m } return r.binarySearch(score, mIdx+1, right) } else { pIdx := mIdx - 1 if pIdx < 0 || r.spans[pIdx].min > score { return m } return r.binarySearch(score, left, mIdx-1) } } func (r *Rank) UpdateScore(id int, score int) { item := r.getRankItem(id) if nil == item { item = &rankItem{id: id} r.id2Item[id] = item } else { if item.score >= score { return } } item.score = score c := r.findSpan(score) if c == item.s { c.update(item) } else { oldC := item.s if item.s != nil { sl := item.s sl.remove(item) } var downItem *rankItem if downItem = c.add(item); nil != downItem { downCount := 0 downIdx := c.idx for nil != downItem { downIdx++ downCount++ if downCount <= 15 { if downIdx >= len(r.spans) { r.spans = append(r.spans, newSpan(downIdx)) } } else { //超过down次数，创建一个新的span接纳下降item终止下降过程 if downIdx >= len(r.spans) { r.spans = append(r.spans, newSpan(downIdx)) } else if len(r.spans[downIdx].data) >= maxItemCount { if len(r.spans) < cap(r.spans) { //还有空间,扩张containers,将downIdx开始的元素往后挪一个位置，空出downIdx所在位置 l := len(r.spans) r.spans = r.spans[:len(r.spans)+1] for i := l - 1; i >= downIdx; i-- { r.spans[i+1] = r.spans[i] r.spans[i+1].idx = i + 1 } r.spans[downIdx] = newSpan(downIdx) } else { //下一个container满了，新建一个 spans := make([]*span, 0, len(r.spans)+1) for i := 0; i <= downIdx-1; i++ { spans = append(spans, r.spans[i]) } spans = append(spans, newSpan(len(spans))) for i := downIdx; i < len(r.spans); i++ { c := r.spans[i] c.idx = len(spans) spans = append(spans, c) } r.spans = spans } } } downItem = r.spans[downIdx].add(downItem) } } if nil != oldC { if len(oldC.data) == 0 { //oldC已经被清空，需要删除 for i := oldC.idx + 1; i < len(r.spans); i++ { c := r.spans[i] r.spans[i-1] = c c.idx = i - 1 } r.spans[len(r.spans)-1] = nil r.spans = r.spans[:len(r.spans)-1] } } } } func (r *Rank) Check() bool { if len(r.spans) > 0 { max := r.spans[0].max for _, v := range r.spans { max = v.check(max) if max == -1 { return false } } return true } else { return true } }

跟上次一样的测试方法，本次数据量为100W，执行200W此，耗时16.89s。平均11.84W/s。比单一数组的实现高了几个数量级，对于100W数据级，这个实现完全可以接受。

思路3：数组链表

在基于数组的实现中，存放rankItem的1级数组无法避免元素移动，可以考虑使用链表来代替1级数组。

代码如下

package rank import ( "fmt" "strings" ) const maxItemCount int = 100 const realRankIdx int = 100 const vacancyRate int = 10 //空缺率10% const vacancy int = maxItemCount * vacancyRate / 100 type rankItemBlock struct { items []rankItem nextFree int } func (rb *rankItemBlock) get() *rankItem { if rb.nextFree >= cap(rb.items) { return nil } else { item := &rb.items[rb.nextFree] rb.nextFree++ return item } } func (rb *rankItemBlock) reset() { rb.nextFree = 0 } func newRankItemBlock() *rankItemBlock { return &rankItemBlock{ items: make([]rankItem, 100000), } } type rankItemPool struct { blocks []*rankItemBlock nextFree int } func newRankItemPool() *rankItemPool { return &rankItemPool{ blocks: []*rankItemBlock{newRankItemBlock()}, } } func (rp *rankItemPool) reset() { for _, v := range rp.blocks { v.reset() } rp.nextFree = 0 } func (rp *rankItemPool) get() *rankItem { item := rp.blocks[rp.nextFree].get() if nil == item { block := newRankItemBlock() rp.blocks = append(rp.blocks, block) rp.nextFree++ item = block.get() } item.c = nil return item } type rankItem struct { id uint64 score int pprev *rankItem pnext *rankItem c *span } type span struct { idx int max int min int count int head rankItem tail rankItem } func newSpan(idx int) *span { c := &span{ idx: idx, } c.head.pnext = &c.tail c.tail.pprev = &c.head return c } func (c *span) show() { fmt.Printf("------------------------idx:%d,count:%d,max:%d,min:%d-------------------------\n", c.idx, c.count, c.max, c.min) s := []string{} cur := c.head.pnext for cur != &c.tail { s = append(s, fmt.Sprintf("(id:%d,score:%d)", cur.id, cur.score)) cur = cur.pnext } fmt.Println(strings.Join(s, "->")) } func (c *span) remove(item *rankItem) { if c.count == 0 { panic("count == 0 1") } c.count-- p := item.pprev n := item.pnext p.pnext = n n.pprev = p c.fixMinMax() } func (c *span) update(item *rankItem) int { c.remove(item) _, realRank := c.add(item) return realRank } func (c *span) down(item *rankItem) *rankItem { c.count++ item.c = c n := c.head.pnext c.head.pnext = item item.pprev = &c.head n.pprev = item item.pnext = n var r *rankItem if c.count > maxItemCount { if c.count == 0 { panic("count == 0 2") } c.count-- //最后一个元素 r = c.tail.pprev r.pprev.pnext = &c.tail c.tail.pprev = r.pprev } c.fixMinMax() return r } func (c *span) add(item *rankItem) (*rankItem, int) { c.count++ item.c = c //寻找合适的插入位置 var cc *rankItem offset := 0 front := c.head.pnext back := c.tail.pprev for { if front.score <= item.score { cc = front offset += 1 break } else if back.score > item.score { cc = back.pnext offset = c.count - offset break } front = front.pnext back = back.pprev offset++ } //插入到cc前 p := cc.pprev p.pnext = item cc.pprev = item item.pprev = p item.pnext = cc var r *rankItem if c.count > maxItemCount { if c.count == 0 { panic("count == 0 3") } c.count-- //最后一个元素 r = c.tail.pprev r.pprev.pnext = &c.tail c.tail.pprev = r.pprev if item == r { offset = 1 } } c.fixMinMax() return r, offset } func (c *span) fixMinMax() { if c.count > 0 { c.max = c.head.pnext.score c.min = c.tail.pprev.score } } func (c *span) check(max int) int { cc := 0 cur := c.head.pnext for cur != &c.tail { if !(max >= cur.score) { return -1 } else { max = cur.score } cur = cur.pnext cc++ } if cc != c.count { return -1 } return max } func (c *span) append(item *rankItem) { p := c.tail.pprev p.pnext = item c.tail.pprev = item item.pprev = p item.pnext = &c.tail c.count++ item.c = c } func (c *span) pop() *rankItem { f := c.head.pnext if f == &c.tail { return nil } else { f.pnext.pprev = &c.head c.head.pnext = f.pnext c.count-- f.c = nil return f } } //将other中的元素吸纳进来 func (c *span) merge(o *span) { for c.count < maxItemCount { if item := o.pop(); nil != item { c.append(item) } else { break } } c.fixMinMax() o.fixMinMax() } type Rank struct { id2Item map[uint64]*rankItem spans []*span itemPool *rankItemPool nextShink int cc int } func NewRank() *Rank { return &Rank{ id2Item: map[uint64]*rankItem{}, spans: make([]*span, 0, 65536/2), itemPool: newRankItemPool(), } } func (r *Rank) Reset() { r.id2Item = map[uint64]*rankItem{} r.spans = make([]*span, 0, 65536/2) r.itemPool.reset() } func (r *Rank) GetPercentRank(id uint64) int { item := r.getRankItem(id) if nil == item { return -1 } else { return 100 - 100*item.c.idx/(len(r.spans)-1) } } func (r *Rank) getFrontSpanItemCount(item *rankItem) int { c := 0 if item.c.idx < realRankIdx { for i := 0; i < item.c.idx; i++ { c += r.spans[i].count } } else { c = 100 * item.c.idx } return c } func (r *Rank) getExactRank(item *rankItem) int { c := r.getFrontSpanItemCount(item) count := 0 cc := item.c pprev := item.pprev pnext := item.pnext for { if pprev == &cc.head { c += count + 1 break } if pnext == &cc.tail { c += cc.count - count break } pprev = pprev.pprev pnext = pnext.pnext count++ } return c } func (r *Rank) GetExactRank(id uint64) int { item := r.getRankItem(id) if nil == item { return -1 } else { return r.getExactRank(item) } } func (r *Rank) GetScore(id uint64) int { item := r.getRankItem(id) if nil == item { return -1 } else { return item.score } } // 获取排名前 n func (r *Rank) GetTopN(n int) []uint64 { ids := make([]uint64, 0, n) for _, span := range r.spans { item := span.head.pnext for item != &span.tail && n > 0 { n-- ids = append(ids, item.id) item = item.pnext } if n <= 0 { break } } return ids } // 获取整个排名数据 func (r *Rank) GetRankList() []uint64 { return r.GetTopN(r.Size()) } // 获取指定名次的分数 1... func (r *Rank) GetScoreByIdx(idx int) (uint64, int) { curIdx := 0 var span *span for _, s := range r.spans { if curIdx+s.count < idx { curIdx += s.count } else { span = s break } } if span == nil { return 0, -1 } item := span.head.pnext curIdx++ for item != &span.tail && curIdx < idx { curIdx++ item = item.pnext } return item.id, item.score } func (r *Rank) Size() int { return len(r.id2Item) } func (r *Rank) Check() bool { if len(r.spans) > 0 { max := r.spans[0].max for _, v := range r.spans { max = v.check(max) if max == -1 { return false } } } return true } func (r *Rank) Show() { for _, v := range r.spans { v.show() } } func (r *Rank) getRankItem(id uint64) *rankItem { return r.id2Item[id] } func (r *Rank) binarySearch(score int, left int, right int) *span { if left >= right { return r.spans[left] } mIdx := (right-left)/2 + left m := r.spans[mIdx] if m.max > score { nIdx := mIdx + 1 if nIdx >= len(r.spans) || r.spans[nIdx].max < score { return m } return r.binarySearch(score, mIdx+1, right) } else { pIdx := mIdx - 1 if pIdx < 0 || r.spans[pIdx].min > score { return m } return r.binarySearch(score, left, mIdx-1) } } func (r *Rank) findSpan(score int) *span { var c *span if len(r.spans) == 0 { c = newSpan(len(r.spans)) r.spans = append(r.spans, c) } else { c = r.binarySearch(score, 0, len(r.spans)-1) } return c } func (r *Rank) UpdateScore(id uint64, score int) int { r.cc++ defer func() { if r.cc%100 == 0 { r.shrink(vacancy, nil) } }() var realRank int item := r.getRankItem(id) if nil == item { item = r.itemPool.get() item.id = id item.score = score r.id2Item[id] = item } else { if item.score == score { return r.getExactRank(item) } item.score = score } if item.c != nil && item.c.max > score && item.c.min <= score { realRank = item.c.update(item) } else { c := r.findSpan(score) oldC := item.c if item.c != nil { item.c.remove(item) } var downItem *rankItem if downItem, realRank = c.add(item); nil != downItem { downCount := 0 downIdx := c.idx for nil != downItem { downIdx++ downCount++ if downIdx < realRankIdx || downCount <= 15 { if downIdx >= len(r.spans) { r.spans = append(r.spans, newSpan(downIdx)) } } else { //超过down次数，创建一个新的span接纳下降item终止下降过程 if downIdx >= len(r.spans) { r.spans = append(r.spans, newSpan(downIdx)) } else if r.spans[downIdx].count >= maxItemCount { if len(r.spans) < cap(r.spans) { //还有空间,扩张containers,将downIdx开始的元素往后挪一个位置，空出downIdx所在位置 l := len(r.spans) r.spans = r.spans[:len(r.spans)+1] for i := l - 1; i >= downIdx; i-- { r.spans[i+1] = r.spans[i] r.spans[i+1].idx = i + 1 } r.spans[downIdx] = newSpan(downIdx) } else { //下一个container满了，新建一个 spans := make([]*span, 0, len(r.spans)+1) for i := 0; i <= downIdx-1; i++ { spans = append(spans, r.spans[i]) } spans = append(spans, newSpan(len(spans))) for i := downIdx; i < len(r.spans); i++ { c := r.spans[i] c.idx = len(spans) spans = append(spans, c) } r.spans = spans } } } downItem = r.spans[downIdx].down(downItem) } } if nil != oldC { if oldC.count == 0 { //oldC已经被清空，需要删除 for i := oldC.idx + 1; i < len(r.spans); i++ { c := r.spans[i] r.spans[i-1] = c c.idx = i - 1 } r.spans[len(r.spans)-1] = nil r.spans = r.spans[:len(r.spans)-1] } else if oldC.idx != item.c.idx && maxItemCount-oldC.count > vacancy { r.shrink(vacancy, oldC) } } } return realRank + r.getFrontSpanItemCount(item) } func (r *Rank) shrink(vacancy int, s *span) { if nil == s { if r.nextShink >= len(r.spans)-1 { r.nextShink = 0 return } else { s = r.spans[r.nextShink] r.nextShink++ } } if maxItemCount-s.count > vacancy && s.idx+1 < len(r.spans) { n := r.spans[s.idx+1] s.merge(n) if n.count == 0 { //n已经空了，删除 for i := n.idx + 1; i < len(r.spans); i++ { c := r.spans[i] r.spans[i-1] = c c.idx = i - 1 } r.spans[len(r.spans)-1] = nil r.spans = r.spans[:len(r.spans)-1] } } }

同样的测试，100W数据，基于链表的实现耗时8.76s。平均22.83W/s。

把数据量放大到1000W，执行2000W次操作，耗时223.68秒，平均8.9W/s。虽说达不到10W+,但也是可以接受的。

经过性能分析，消耗最大的地方在InsertNode,因为要遍历链表，查找合适的插入位置，要想继续提高性能，就必须减少遍历的次数。

思路4：skiplists

如前面讨论的，普通链表在插入元素时，需要从头开始遍历以找到合适的插入位置，skiplists通过随机算法，可以一次跳过多个节点，加快检索过程。

我们首先还是实现一个单一skiplists的排行榜

package skiplists import ( "fmt" "math/rand" "strings" ) const maxLevel int = 15 type link struct { pnext *node pprev *node skip int //next跳过了多少个节点 } type node struct { key int value uint64 links [maxLevel]link } type skiplists struct { level int size int head node tail node step int } func newSkipLists() *skiplists { sl := &skiplists{} for i := 0; i < maxLevel; i++ { sl.head.links[i].pnext = &sl.tail sl.tail.links[i].pprev = &sl.head //sl.head.links[i].skip = 1 } return sl } func (sl *skiplists) show() { for i := 0; i <= sl.level; i++ { cur := sl.head.links[i].pnext s := []string{} s = append(s, fmt.Sprintf("head skip:%d", sl.head.links[i].skip)) for &sl.tail != cur { s = append(s, fmt.Sprintf("(%d,%d,%d)", cur.key, cur.value, cur.links[i].skip)) cur = cur.links[i].pnext } fmt.Println("level", i+1, strings.Join(s, ",")) } fmt.Println("--------------------------------------------------------------------------") } func (sl *skiplists) randomLevel() int { lvl := 0 for rand.Float32() < 0.4 && lvl < maxLevel-1 { lvl++ } return lvl } /* func (sl *skiplists) Search(key int) *node { x := &sl.head for i := sl.level; i >= 0; i-- { for nil != x.links[i].pnext && x.links[i].pnext.key < key { x = x.links[i].pnext } } x = x.links[0].pnext if nil != x && x.key == key { return x } else { return nil } }*/ /////////////////////////使用node的接口 //由上层确保n不在skiplists里 func (sl *skiplists) InsertNode(n *node) int { key := n.key update := [maxLevel]*node{} //n插入到update后面，n的offset=updateOffset[0] + 1 updateOffset := [maxLevel]int{} //update节点的offset updateOffset[sl.level] = 1 x := &sl.head for i := sl.level; i >= 0; i-- { if i != sl.level { updateOffset[i] = updateOffset[i+1] } for &sl.tail != x.links[i].pnext && x.links[i].pnext.key < key { updateOffset[i] += x.links[i].skip x = x.links[i].pnext sl.step++ } update[i] = x } offset0 := updateOffset[0] + 1 //新节点在level1的位置 //fmt.Println("offset0", offset0) lvl := sl.randomLevel() if lvl > sl.level { for i := sl.level + 1; i <= lvl; i++ { update[i] = &sl.head updateOffset[i] = 1 } sl.level = lvl } x = n for i := 0; i <= sl.level; i++ { if i <= lvl { x.links[i].pprev = update[i] x.links[i].pnext = update[i].links[i].pnext update[i].links[i].pnext = x x.links[i].pnext.links[i].pprev = x oldSkip := update[i].links[i].skip update[i].links[i].skip = offset0 - updateOffset[i] if x.links[i].pnext != &sl.tail { x.links[i].skip = (updateOffset[i] + oldSkip + 1) - offset0 if x.links[i].skip < 0 { panic("3") } } else { x.links[i].skip = 0 } } else if update[i].links[i].skip > 0 { update[i].links[i].skip++ } } sl.size++ return offset0 - 1 } func (sl *skiplists) isInLink(lvl int, head *node, n *node) bool { cur := head.links[lvl].pnext for cur != &sl.tail { if cur == n { return true } else { cur = cur.links[lvl].pnext } } return false } func (sl *skiplists) DeleteNode(n *node) { update := [maxLevel]*node{} x := n lvl := 0 for x != nil { pprev := x.links[lvl].pprev if (x.links[lvl].pprev != nil || x.links[lvl].pnext != nil) && nil == update[lvl] { //当前节点是lvl链上的节点 if x == n { update[lvl] = pprev } else { update[lvl] = x } if update[lvl].links[lvl].pnext != n && update[lvl].links[lvl].skip == 0 { break } } if lvl < sl.level && (nil != x.links[lvl+1].pprev || nil != x.links[lvl+1].pnext) { //当前节点也在lvl+1的链上 lvl++ } else { x = pprev } } for i := 0; i <= sl.level; i++ { if nil == update[i] { break } else { if update[i].links[i].pnext != n { if update[i].links[i].skip > 0 { update[i].links[i].skip-- if update[i].links[i].skip < 1 { panic("1") } } else { break } } else { update[i].links[i].pnext = n.links[i].pnext n.links[i].pnext.links[i].pprev = update[i] if update[i].links[i].pnext != &sl.tail { update[i].links[i].skip = update[i].links[i].skip + n.links[i].skip - 1 if update[i].links[i].skip < 0 { panic("2") } } else { update[i].links[i].skip = 0 } n.links[i].pnext = nil n.links[i].pprev = nil n.links[i].skip = 0 } } } for sl.level > 0 && sl.head.links[sl.level].pnext == &sl.tail { sl.level-- } } func (sl *skiplists) GetNodeRank(n *node) int { rank := 0 x := n var pprev *node var lvl int for pprev != &sl.head { for i := sl.level; i >= 0; i-- { pprev = x.links[i].pprev if nil != pprev { lvl = i break } } rank += pprev.links[lvl].skip x = pprev } return rank } func (sl *skiplists) GetNodeRankCheck(n *node) int { rank := 1 cur := sl.head.links[0].pnext for cur != &sl.tail { if cur == n { break } else { rank++ cur = cur.links[0].pnext } } if cur == &sl.tail { return -1 } return rank }

1000W数据执行2000W次，耗时127.81秒，平均15.64W/s。

最后是skiplists的数组，这个实现比单一skiplists有更好的更新以及排名查询性能，代码在下面的仓库中

思路5：有序平衡树

假设当前子树A的score=a，规定score>a的子树位于A的左侧，score>=a的子树位于A的右侧。

我们来看一下如何计算A的排名。

如果A没有父节点，根据定义A左子树的score>a， 因此A的排名就是A->left->size + 1。

如果A有父节点，则存在两种情况：

A == A->parent->left：A->score > A->parent->score，因此A的排名=A->parent的排名-1。

A == A->parent->right：A->score <= A->parent->score，因此A的排名=A->parent的排名+1。

type tree struct { score int parent *tree left *tree right *tree size int } func rank(t *tree) int { if nil == t->parent { if t->left == nil { return 1 } else { return t->left->size + 1 } } else { if t == t->parent->left { return rank(t->parent) - 1 } else { return rank(t->parent) + 1 } } }

有兴趣的朋友可以搜索下SBTree。它是使用根据左右子树的size来做平衡的，实现难度比红黑树小多了。

总体来说，有序平衡树的时间复杂度跟skiplists是一个等级的。skiplists可以看作一棵利用概率进行平衡的有序平衡树。在同样的数据量下，有序平衡树的内存占用比skiplists更低。