Jun 25

踩坑记#2:Go服务锁死 不指定

felix021 @ 2020-6-25 22:48 [IT » 程序设计] 评论(0) , 引用(0) , 阅读(1369) | Via 本站原创
接着 上一篇-内存暴涨坑 再挖个坟,讲讲去年踩的另一个坑。 

---

前方低能

那是去年7月的一天,被透过落地玻璃的宇宙中心五道口的夕阳照着的正在工位搬砖的我,突然听到一阵骚乱,转头一看,收到夺命连环call的D同学反馈,流量严重异常。

点开报警群,一串异常赫然在目: 
Jun 21

golang: bufio.Scanner 的坑 不指定

felix021 @ 2020-6-21 01:03 [IT » 程序设计] 评论(0) , 引用(0) , 阅读(1939) | Via 本站原创
之前从网上找的一段代码,按行读取文件:

inFile, err := os.Open("xxx.log")
if err != nil {
    fmt.Fprintf(os.Stderr, "open failed: %v\n", err)
    return
}
defer inFile.Close()

scanner := bufio.NewScanner(inFile)
for scanner.Scan() {
    line := scanner.Bytes()
    //do sth. with line
}


看起来没问题,用起来也没问题,直到踩了个坑:针对某个特定的文件,读取到某一行以后就不再继续了。

既然总能复现,那就好解决,我的一个常用方法是:制造一个总能复现的case,并不断缩小case的规模。

例如这个case,把那一行单独拿出来,通过二分找到出问题的位置。

原以为是该行有特殊字符导致触发了什么奇怪的逻辑,但经过不断尝试,发现临界点是该行长度 = 65536 的时候,正好会触发错误。

这么整的数字(2^16, 64KB)必然是代码里的特殊逻辑了,翻了一下 bufio 的源码,果然有一个

const (
  //...(一堆注释)...
  MaxScanTokenSize = 64 * 1024
)


搜索这个常量在代码里的引用:
func NewScanner(r io.Reader) *Scanner {
  return &Scanner{
    r:            r,
    split:        ScanLines,
    maxTokenSize: MaxScanTokenSize,
  }
}

...

func (s *Scanner) Scan() bool {
  ....
  if len(s.buf) >= s.maxTokenSize || len(s.buf) > maxInt/2 {
    s.setErr(ErrTooLong)
    return false
  }
  ...
}


在 for 循环后加上一句:
if scanner.Err() != nil {
  fmt.Fprintf(os.Stderr, "scan err: %v\n", scanner.Err())
}


实锤:
引用
scan err: bufio.Scanner: token too long



那怎么解决呢?

MaxScanTokenSize 上面的注释是这么写的:

引用
  // MaxScanTokenSize is the maximum size used to buffer a token
  // unless the user provides an explicit buffer with Scanner.Buffer.
  // The actual maximum token size may be smaller as the buffer
  // may need to include, for instance, a newline.


于是最终版的解决方案是这样:

...
scanner := bufio.NewScanner(inFile)
buf := make([]byte, 0, bufio.MaxScanTokenSize * 10) //根据自己的需要调整这个倍数
scanner.Buffer(buf, cap(buf))

for scanner.Scan() {
  line := scanner.Bytes()
  //do sth. with line
}

if scanner.Err() != nil {
  fmt.Fprintf(os.Stderr, "scan err: %v\n", scanner.Err())
}


真是丑陋的api啊。
Sep 21
最近线上服务压力很大,api的p99有点扛不住。

广告业务对延时的要求普遍比较严格,有些adx设置的超时时间低至100ms,因此亟需找出性能热点。

根据对目前系统情况的估计(和metrics埋点数据),大致估计问题出在广告的正排环节。

使用 pprof  也证明了这一块确实是热点:

引用
$ go tool pprof http://$IP:$PORT/debug/pprof/profile
...
(pprof) top 10
Showing nodes accounting for 25.50s, 32.63% of 78.14s total
Dropped 1533 nodes (cum <= 0.39s)
Showing top 10 nodes out of 284
      flat  flat%  sum%        cum  cum%
    4.56s  5.84%  5.84%      4.87s  6.23%  syscall.Syscall
    4.03s  5.16% 10.99%      4.03s  5.16%  runtime.aeshashbody
    3.50s  4.48% 15.47%      6.01s  7.69%  git.xxx.org/xxx/target.NewFilter
    2.78s  3.56% 19.03%      3.73s  4.77%  runtime.mapaccess2_fast64
    2.63s  3.37% 22.40%      4.52s  5.78%  runtime.mapiternext
    2.08s  2.66% 25.06%      2.16s  2.76%  runtime.heapBitsForObject
    1.65s  2.11% 27.17%      1.93s  2.47%  runtime.mapaccess1_fast64
    1.57s  2.01% 29.18%      2.96s  3.79%  runtime.mapaccess2
    1.43s  1.83% 31.01%      2.06s  2.64%  runtime.runqgrab
    1.27s  1.63% 32.63%      1.27s  1.63%  runtime.epollwait
(pprof) png
Generating report in profile001.png (使用生成的线框图查看耗时)


其中第三行 NewFilter 就是正排过滤函数。因为一些历史原因,系统里不是所有定向条件都使用了倒排,正排实现起来毕竟简单、容易理解,而一旦开了这个口子,就会有越来越多正排加进来,推测是这个原因导致了性能的逐渐衰退。

经过讨论,D同学花了一周多的时间,逐个梳理重写。在Libra(字节跳动内部的ABTest平台,参考谷歌分层实验框架方案)上开实验,平均耗时 -9%,从统计数据上来说,实验组比对照组有了显著的改进,但从最终结果上,整体的p95、p99超时都出现了进一步恶化。

这说明真正的问题不在于正排的计算,优化的思路出现了偏差。

考虑到晚高峰期间的cpu占用率也只是刚超过50%,也就是说有可能性能问题在于锁,但pprof的 block 和 mutex 都是空的,没有线索。

猜测问题有可能在日志,代码里确实用得不少。日志用的是 github.com/ngaut/logging 库,每一次调用都会用到两个全局mutex。但通过调整log level 为error级别,大幅减少了日志量,并没有看到性能的改善。

经过搜索,发现 uber 基于 pprof 开发了一个神器 go-torch,可以生成火焰图。安装好 go-torch 及依赖 FlameGraph 以后执行

引用
$ go-torch  -u http://$IP:$PORT -f cpu.svg
INFO[14:52:23] Run pprof command: go tool pprof -raw -seconds 30 http://$IP:$PORT/debug/pprof/profile
INFO[14:52:54] Writing svg to cpu.svg


用 Chrome 打开 cpu.svg,人肉排查:

点击在新窗口中浏览此图片

可以看到,在NewFilter旁边竟然还有一个耗时接近的 runtime.growslice ,结合实际代码(略作简化),可以推测是 slice 的初始化长度不足。

matchAds := make([]*ad, 0, 4096)
adsBitMap.GetList(func(seq int) {
    if NewFilter(ctx, ad) {
        matchAds = append(matchAds, adlist[seq])
    }
})

// 顺便提一下,bitmap是一个uint64数组,GetList(f) 是将每一个等于1的bit索引传给 f
// GetList方法里面用了cpu的BSF指令来提高性能。



实际上最终定向后得到的广告往往在数万甚至数十万的级别,而 go 的 slice 扩容在超过1024个元素以后是1.25倍,可想而知会出现大量的内存分配和拷贝,导致性能随着广告数量的增加逐渐恶化。最近的广告数量也确实有了大幅的上升 —— 逻辑上形成了闭环。

经过优化,使用更大的初始化长度,并且使用 sync.Pool 来进一步减少内存分配,最终上线后p95和p99都下降了超过50%,效果显著。

参考:
golang 使用pprof和go-torch做性能分析 https://www.cnblogs.com/li-peng/p/9391543.html
Jul 2
线上服务Panic,部分日志如下
引用
err: runtime error: index out of range
Traceback:
goroutine 19209941 [running]:
...
panic(0x191d0e0, 0x2e078d0)
  /usr/local/go/src/runtime/panic.go:502 +0x229
math/rand.(*rngSource).Uint64(...)
  /usr/local/go/src/math/rand/rng.go:246
math/rand.(*rngSource).Int63(0xc438bb2a00, 0x0)
  /usr/local/go/src/math/rand/rng.go:231 +0x8a
math/rand.(*Rand).Int63(0xc4279b3a70, 0x0)
  /usr/local/go/src/math/rand/rand.go:82 +0x33
math/rand.(*Rand).Int(0xc4279b3a70, 0x0)
  /usr/local/go/src/math/rand/rand.go:100 +0x2b
...


放狗搜了一下:math.Rand is not safe for concurrent use

from: https://github.com/golang/go/issues/3611

这个 issue 的 4 楼还提到 "top-level functions like strings.Split or fmt.Printf or rand.Int63 may be called from any goroutine at any time"

翻了一下源码,rand.Int() 用是自带 lock 的 globalRand 对象

func Int() int { return globalRand.Int() }

...

var globalRand = New(&lockedSource{src: NewSource(1).(Source64)})

...

type lockedSource struct {
  lk  sync.Mutex
  src Source64
}

...

func (r *lockedSource) Uint64() (n uint64) {
  r.lk.Lock()
  n = r.src.Uint64()
  r.lk.Unlock()
  return
}


看了下调用代码,之前的实现为了避免多个 goroutine 竞争同一个锁,所以 new 了一个 rand.Rand 对象,但没考虑到这个对象不支持并发。

最终的解决方案,是实现了一个 safeRander 。

具体代码不适合贴,核心逻辑是初始化 N 个 rand.Rand 对象和对应的 N 个锁,以及一个 index,每次调用 Int() 时,先 atomic.AddUint32(&index, 1) % N,加上对应的锁,再用对应的 rand.Rand 对象。

这样只要并发使用的goroutine不超过N个,就不会出现竞争;就算超过,竞争出现的频率也大幅减少了,而且也可以通过增加 N 来优化。
Mar 31

LeetCode 69 - x 的平方根 不指定

felix021 @ 2019-3-31 15:34 [IT » 程序设计] 评论(1) , 引用(0) , 阅读(2560) | Via 本站原创
这是罗凯同学布置的 Golang 学习作业。

这题之前用 Python 刷过,用的是二分法,在 [1, n / 2] 区间内,找到第一个 x,使得 x ^ 2 <= n < (x + 1) ^ 2 ,用的是 STL 中 lowerbound 的算法。

class Solution(object):
    def mySqrt(self, x):
        """
        :type x: int
        :rtype: int
        """
        if x < 0:
            raise Exception("invalid input")
        if x < 2:
            return x
       
        left = 1
        length = x / 2
        while length > 1:
            half = length / 2
            middle = left + half
            if middle * middle > x:
                length = half
            else:
                left = middle
                length = length - half
        return left


罗凯同学提到,应该使用牛顿迭代法来完成。这个方法是听说过的,但是早就忘了,于是到 wikipedia 去找了一下:

https://zh.wikipedia.org/wiki/%E7%89%9B%E9%A1%BF%E6%B3%95

求函数 f(x) 的零点,可以通过选取曲线上的任意一个点 x0 开始,然后计算 x1 = x0 - f(x1) / f'(x1) 的方式迭代,通常得到一个比 x0 更接近零点的 x1 。通过不断迭代,最终我们能找到一个零点 xn 。

对于求平方根,我们是要找到一个 x,使得 x ^2 - n = 0,也就是这里的 f(x) = x ^ 2 - n, f'(x) = 2 * x (勉强还记得这个求导公式……)

有了这个,答案就呼之欲出了:

import "math"

func mySqrt(x int) int {
    f := func (i float64) float64 {
        return i * i - float64(x)
    }
    g := func (i float64) float64 {
        return 2 * i
    }
    var i float64 = 1.0
    for math.Abs(f(i)) > 1e-6 {
        i = i - f(i) / g(i)
    }
    return int(math.Floor(i))
}


做完以后,我想起 Quake III 的作者 John Carmack 的 平方根倒数速算法,摘录一段内容:( src: https://blog.csdn.net/zyex1108/article/details/53540824 )

引用

Quake-III Arena (雷神之锤3)是90年代的经典游戏之一。该系列的游戏不但画面和内容不错,而且即使计算机配置低,也能极其流畅地运行。这要归功于它3D引擎的开发者约翰-卡马克(John Carmack)。事实上早在90年代初DOS时代,只要能在PC上搞个小动画都能让人惊叹一番的时候,John Carmack就推出了石破天惊的Castle Wolfstein, 然后再接再励,doom, doomII, Quake...每次都把3-D技术推到极致。他的3D引擎代码资极度高效,几乎是在压榨PC机的每条运算指令。


这个平方根倒数算法正是其中的一个例子。在3D游戏引擎中,求取照明和投影的波动角度与反射效果时,常需计算平方根倒数,而求平方根的常用算法效率较低。


Carmack 通过使用一个惊为天人的魔术常量 0x5f3759df,只需要做 1 次迭代(Quaker III源码中的为了提高精度的第二次迭代被注视掉了),就能得到一个足够精度的平方根,大幅提高了 3D 引擎的运行效率。

关于这个魔术常量,Carmack 表示并不是他自己发明的,至今为止仍未能确切知晓算法中所使用的特殊常数的起源。但 Carmack 凭一己之力,撑起了一个 3D 引擎的时代,以至于在1999年,登上了美国时代杂志评选出来的科技领域50大影响力人物榜单,并且名列第10位。

感兴趣的同学,可以在 Wikipedia 的 平方根倒数速算法 了解更多细节:

https://zh.wikipedia.org/wiki/%E5%B9%B3%E6%96%B9%E6%A0%B9%E5%80%92%E6%95%B0%E9%80%9F%E7%AE%97%E6%B3%95
Jan 29
TLDR版本:https://leetcode-cn.com/explore/ ,注册一个帐号开始做题就行了。

== 以下是正文 ==

作为一个程序员,编码能力是基础的基础。

我比较幸运,在大学的时候参加了学校的 ACM/ICPC 集训队,接触了 ACM/ICPC 比赛。这是一个针对大学生编程能力的世界级比赛,要求在几个小时的时间里完成若干道不同难度的题目,其中很多题目不仅需要复杂的算法、有各种特殊情况需要考虑,而且还有变态级的效率要求。强如楼教主(楼天城),也仅在 2009 年获得世界总决赛的第二名。

此外,从我观测到的结果来看,但凡从集训队走出去的成员(无论其竞赛成绩如何),**其毕业后的第一份工作(通常都是 BAT )乃至之后的发展,都显著高于计算机专业的平均水平**。

虽然在集训队里有教练,也有大神,但日常学习主要还是靠自己。看书学习固然是一种方式,但是比较枯燥,也不容易衡量自己的学习成果。另一方面,由于赛事多年的发展和积淀,国内参赛实力较强的大学(例如北大、杭州电子科技大学、华中科技大学)都创建了自己的在线测评系统(英文名叫 Online Judge,简称 OJ)。

OJ 上沉淀了多年来的竞赛题目,每一个题目都包含相应的题面、输入说明、输出要求、基础测试用例;用户按要求编写代码后,将代码提交给 OJ,系统会在后台启动自动化测试,告知测评结果。

由于 OJ 系统的存在,做题变成了一种乐趣,通过努力解决了一个问题,系统会给出红色的 Accepted 字样,就像一种奖赏;而在这个过程中,也可以直接地看到自己的进步。

工作以后,我非常庆幸当年自己在 OJ 系统刷过这些题,夯实了编程能力,在工作中能够完成更高质量的代码。而在过去几年的面试过程中,我发现很多来应聘的程序员,往往只能应对简单的情况,处理不好边界问题、例外情况、运行效率带来的挑战。

遗憾的是,由于学校自建的 OJ 往往都是学生自己开发、自己维护(我也写过一个,维护过几年,深有体会),体验较差,对存量题目的组织、整理也比较随意(往往只是简单的罗列),而且由于比赛是英文环境,题面往往也都是纯英文的,给竞赛圈之外的同学带来了一定门槛。

所幸,近年来,第三方(商业公司、志愿者社区)的 OJ 系统也逐渐完善,其中一个我很喜欢的平台是 LeetCode ,大约成立于 2008 年吧,上面的题多是业内 TOP 公司的面试题,很多人通过刷这些题来应聘喜欢面试算法的 NTMGBA 系列公司(注:Netease,Tencent,Microsoft,Google,Baidu,Alibaba/Amazon)。

相比各个学校维护的 OJ 平台,LeetCode 的体验令人称道:

* 支持多种语言,包括 PHP、Python、Go、Rust、Javascript,甚至还有基于 MySQL 的题目
* 推出了完整的中文版,包括纯中文的题面
* 对题目做了细致的整理,打上各种标签,包括难度(简单、中等、困难)、话题(字符串、堆/栈、贪心算法、动态规划等)
* 通过合集的方式,将题目整理归档(例如腾讯精选50题、初级算法、中级算法等)
* 对于错误的情况,给出明确的错误原因,及相应的输入输出数据,方便自我纠正
* 许多题目有详尽的官方解答,即使不会做也能够直接学习

LeetCode 上的题目大致可以分成两种(参考 CoolShell 博客说明):

1. 算法题。大多是套路题,每道题目都需要特定的算法,例如BFS、DFS、动态规划、回溯等。通过做这些题,能够让自己对这些最基础的算法的思路有非常扎实的了解和训练,也能很好地锻炼自己的思维能力(烧脑)。

2. 编程题。比如:atoi,strstr,add two num,括号匹配,字符串乘法,通配符匹配,文件路径简化,Text Justification,反转单词等等。这些题目的题面都很简单,大部分程序员都能读懂,但是魔鬼藏在细节中,具体的实现往往需要考虑多种情况。通过做这些题,可以非常好的训练自己对各种情况的考虑,以及对程序代码组织的掌控能力(其实就是其中的状态变量)。程序中的状态正是程序变得复杂难维护的直接原因。

每个程序员内心都有一个大神梦,但是别忘了,大神也是从菜鸟一步一个脚印走过来的,而 LeetCode 就是一个很好的垫脚石,共勉。
Sep 19
# 1. 什么是跳表

跳表(Skip List)是基于链表 + 随机化实现的一个有序数据结构,可以达到平均 O(logN) 的查找、插入、删除效率,在实际运行中的效率往往超过 AVL 等平衡二叉树,而且其实现相对更简单、内存消耗更低。

Redis 的 ZSET 底层实现就是用的 Skip List,这里是 [Antirez对此的说明](https://news.ycombinator.com/item?id=1171423)

这是一个典型的跳表:

[0] -> 0 -> 1 -> 3 -> 4 -> 5 -> 6 -> 7 -> 9 -> nil
[1] -> 0 ------> 3 ------> 5 ------> 7 ------> nil
[2]----------------------> 5-----------------> nil


解释一下:

1. SkipList 是一个多层的链表

2. 第[0]层的链表包含所有节点,其他层的链表包含部分节点,层次越高,节点越少

3. 每层链表之间会共享相同的节点(节省内存,但为了方便展示,每一层都输出了它的值)

4. 对于某个节点,在插入时通过概率判断它最高会出现在哪一层,并且也会出现在之下的每一层

通过这样的设计,当需要查找某个 key 时,可以从最高层的链表开始往前找,在这一层遇到末尾或者大于 key 的节点时往下走一个层,直到找到 key 节点。

例如:

引用
4 的查找路径为 [2] -> [1] -> 0 -> 3 -> 3@[0] -> 4
6 的查找路径为 [2] -> 5 -> 5@[1] -> 5@[0] -> 6
8 的查找路径为 [2] -> 5 -> 5@[1] -> 7 -> 7@[0] -> 9 (找不到)


# 2. 跳表的节点

从上面的描述,我们大概可以知道 (1) 每个节点需要保存一个 key; (2) 每个节点需要有多个next指针 (3) 其 next 指针的数量会在插入时确定

因此我们可以用下面这个 class 来表示节点:

class Node(object)
    def __init__(self, height, key):
        self.key = key
        self.next = [None] * height

    def height(self):
        return len(self.next)


# 3. 创建跳表

一个新创建的跳表是没有节点的。但为了实现的简单起见,可以添加一个头节点:

class SkipList(object):
    def __init__(self):
        self.head = Node(0, None) #头节点高度为0,不需要key


到目前为止都特别简单,但是还什么也干不了。

# 4. 创建节点

创建节点时,需要先按一定的概率分布确定其高度。

为了保证高层的节点比低层少,我们可以用这样的概率分布:

引用
Height(n) = p^n


实现其实非常简单:

import random

def randomHeight(self, p = 0.5):
    height = 1
    while random.uniform(0, 1) < p and self.head.height() >= height:
        height += 1
    return height


这样可以保证平均的路径长度是 log(n) 。

精确一点的话,实际上是 log(n-1, 1/p) / p,也就是说, p 的选择会影响跳表层数、平均路径长度。

具体的计算比较复杂,有兴趣可以参考跳表的原论文《Skip Lists: A Probabilistic Alternative to Balanced Trees》。(TL;DR)

然后我们就可以这样来创建一个新的节点:

    node = Node(self.randomHeight(), key)

# 5. 添加节点

如果只是为空跳表添加一个新的节点,只要更新头结点的每一个next指针:

def insertFirstNode(self, key):
    node = Node(self.randomHeight(), key)
    while node.height > self.head.height():
        self.head.next.append(None) #保证头节点的next数组覆盖所有层次的链表

    for level in range(node.height()):
        node.next[level] = self.head.next[level]
        self.head.next[level] = node


但很显然这个方法只能用一次。

如果跳表中已经有多个节点,那我们就必须找到每一层中适合插入的位置:

def getUpdateList(self, key):
    update = [None] * self.head.height()
    for level in range(len(update)):
        x = self.head
        while x.next[level] is not None and x.next[level].key < key:
            x = x.next[level]
        update[level] = x
    return update


这个函数返回一个 update 节点数组,其中的每个节点都是在这一层中小于 key 的最后一个节点。

也就是说,在 level = i 层,总是可以把新的节点插入 update[i] 之后:

def insert(self, key):
    node = Node(self.randomHeight(), key)
    while node.height > self.head.height():
        self.head.next.append(None) #保证头节点的next数组覆盖所有层次的链表

    update = self.getUpdateList(key)
    next0 = update[0].next[0]
    if next0 is not None and next0.key == key:
        return # 0层总是包含所有元素;如果 update[0] 的下一个节点与key相等,则无需插入。

    for level in range(node.height()):
        node.next[level] = update[level].next[level]
        update[level].next[level] = node


但是由于这一版 getUpdateList 是 O(n) 的,插入效率并没有达到跳表的设计目标。

# 6. 添加节点++

考虑这一点:跳表的每一层都是有序的。

也就是说,我们在找到 update[n] = x 以后,其实可以从节点 x 的 n - 1 层继续查找 update[n-1] 应该是哪个节点。

由于查找路径的平均长度是 log(N) ,所以我们可以实现一个更快的 getUpdateList 方法

注意,需要从最高层开始查

def getUpdateList(self, key):
    update = [None] * self.head.height()
    x = self.head
    for level in reversed(range(len(update))):
        while x.next[level] is not None and x.next[level].key < key:
            x = x.next[level]
        update[level] = x
    return update


# 7. 里程碑1

把上面的代码整合起来,我们就可以得到第一版跳表代码:能够插入节点。

为了更好地展示我们的成果,我们可以用这样一个函数,把链表按第1节的例子样式输出:

def dump(self):
    for i in range(self.head.height()):
        sys.stdout.write('[H]')
        x = self.head.next[0]
        y = self.head.next[i]
        while x is not None:
            s = ' -> %s' % x.key
            if x is y:
                y = y.next[i]
            else:
                s = '-' * len(s)
            x = x.next[0]
            sys.stdout.write(s)
        print ' -> <nil>'
    print


试试看:

sl = SkipList()
for i in range(10):
    sl.insert(sl)
    s1.dump()



[H] -> 0 -> 1 -> 2 -> 3 -> 4 -> 5 -> 6 -> 7 -> 8 -> 9 -> <nil>
[H]----- -> 1 -> 2 -> 3---------- -> 6 -> 7---------- -> <nil>
[H]---------- -> 2-------------------- -> 7---------- -> <nil>


多尝试几次,以及选择不同的 p 值,可以观察生成跳表的区别。

# 8. 查找节点

实际上查找节点的过程,已经包含在 insert 的实现里了:

def find(self, key):
    update = self.getUpdateList(key)
    if len(update) == 0:
        return None

    next0 = update[0].next[0]
    if next0 is not None and next0.key == key:
        return next0 # 0层总是包含所有元素;如果 update[0] 的下一个节点与key相等,则无需插入。
    else:
        return None


# 9. 删除节点

既然已经能找出 update 节点数组,在 level = i 层,只要判断 update[i].next[i] 是否等于要删除的 key 就可以了:

def remove(self, key):
    update = self.getUpdateList(key)
    for i, node in enumerate(update):
        if node.next[i] is not None and node.next[i].key == key:
            node.next[i] = node.next[i].next[i]


# 10. 里程碑2

整合 find 和 update 数组,就可以实现跳表的基础操作了,试试看:

node = sl.find(3)
print node

for i in range(7, 14):
    sl.remove(i)
    sl.dump()


# 11. 其他

我们在 Node 中只添加了一个 key 属性,在具体的实现中,我们往往可能需要针对 key 存储一个 value,例如 Python 自带的 dict 实现。改造起来也很简单:

1. node 中添加一个 value 属性,并且添加相应的初始化逻辑(__init__方法)

2. 将 SkipList.insert 修改为 `insert(self, key, value)`,在新建 Node 时指定其 value

3. 再添加一个 `update(self, key, value)` API,方便调用方的使用

4. 可以考虑针对语言适配,例如实现 python 的 __getitem__ 、 __setitem__ 等魔术方法


# 12. 完整代码

#coding:utf-8

import random

class Node(object):
    def __init__(self, height, key=None):
        self.key = key
        self.next = [None] * height

    def height(self):
        return len(self.next)

class SkipList(object):
    def __init__(self):
        self.head = Node(0, None) #头节点高度为0,不需要key

    def randomHeight(self, p = 0.5):
        height = 1
        while random.uniform(0, 1) < p and self.head.height() >= height:
            height += 1
        return height

    def insert(self, key):
        node = Node(self.randomHeight(), key)
        print node.height(), node.key
        while node.height() > self.head.height():
            self.head.next.append(None) #保证头节点的next数组覆盖所有层次的链表

        update = self.getUpdateList(key)
        if update[0].next[0] is not None and update[0].next[0].key == key:
            return # 0层总是包含所有元素;如果 update[0] 的下一个节点与key相等,则无需插入。

        for level in range(node.height()):
            node.next[level] = update[level].next[level]
            update[level].next[level] = node


    def getUpdateList(self, key):
        update = [None] * self.head.height()
        x = self.head
        for level in reversed(range(len(update))):
            while x.next[level] is not None and x.next[level].key < key:
                x = x.next[level]
            update[level] = x
        return update

    def dump(self):
        for i in range(self.head.height()):
            sys.stdout.write('[H]')
            x = self.head.next[0]
            y = self.head.next[i]
            while x is not None:
                s = ' -> %s' % x.key
                if x is y:
                    y = y.next[i]
                else:
                    s = '-' * len(s)
                x = x.next[0]
                sys.stdout.write(s)
            print ' -> <nil>'
        print

    def find(self, key):
        update = self.getUpdateList(key)
        if len(update) == 0:
            return None

        next0 = update[0].next[0]
        if next0 is not None and next0.key == key:
            return next0 # 0层总是包含所有元素;如果 update[0] 的下一个节点与key相等,则无需插入。
        else:
            return None

    def remove(self, key):
        update = self.getUpdateList(key)
        for i, node in enumerate(update):
            if node.next[i] is not None and node.next[i].key == key:
                node.next[i] = node.next[i].next[i]


完。
Mar 2
我注意到过去几个月我司有些同学还在踩一个简单的分布式事务Case的坑,而这个坑在两年以前就已经有同学踩过了,这里简单解析一下这个case和合适的处理方案,供参考。

1. 踩过的坑

这个case有很多变种,先从我们在XX业务踩过的坑开始,大约是16年9月,核心业务需求是很简单的:在用户发起支付请求的时候,从用户的银行卡扣一笔钱。负责这个需求的同学是这么写的代码(去除其他业务逻辑的简化版):
$dbTrans = $db->beginTransaction();
try {
    $order = PayRequest::model()->newPayRequest(...); #在数据库中插入一条支付请求记录,状态为待支付
    //其他业务改动
    $result = PaySvr::pay($order->id, $order->amount); #请求PaySvr(或第三方支付通道)扣款
    if ($result['code'] == PaySvr::E_SUCCESS) {
        $order->setAsSucceeded();
    } else {
        $order->setAsPending();
    }
    $dbTrans->commit();
} catch (Exception $e) {
    $dbTrans->rollback();
}

乍一看好像是没有什么毛病,测试的case都顺利提供过,也没有人去仔细review这一小段代码,于是就这么上线了。但问题很快就暴露出来,PaySvr在支付成功以后尝试回调,XX业务系统报错”订单不存在”。查询线上日志发现,这笔订单在请求第三方支付通道时网络超时,Curl抛了timeout异常,导致支付记录被回滚。有心的同学可以自己复现一下这个问题,观察BUG的发生过程。

代码修复起来倒是很简单,在请求PaySvr之前提交事务,将支付请求安全落库即可
$dbTrans = $db->beginTransaction();
try {
    $order = PayRequest::model()->newPayRequest(...);
    //其他业务改动
    $dbTrans->commit(); #先将支付请求落地
} catch (Exception $e) {
    $dbTrans->rollback();
}

#再请求PaySvr
$result = PaySvr::pay($order->id, $order->amount);

#根据PaySvr结果修改支付请求和其他业务记录的状态
$dbTrans = $db->beginTransaction();
try {
    if ($result['code'] == PaySvr::E_SUCCESS) {
        $order->setAsSucceeded();
        //其他业务改动
    } else {
        $order->setAsPending();
        //其他业务改动
    }
} catch (Exception $e) {
    $dbTrans->rollback();
}

把这个实现代入多个不同的业务下,还会衍生出更多问题,比如被动代扣业务,就可能因为重试导致用户被多次扣款,引起投诉(支付通道对投诉率的要求非常严格,甚至可能导致通道被关停);更严重的是放款业务,可能出现重复放款,给公司造成直接损失。据说某友商就是因为重复放款倒闭的,所以在实现类似业务时特别注意,考虑周全。

2. 归纳总结

我们往后退一步再审视这个case,这段简单的代码涉及了两个系统:XX业务系统(本地数据库)、PaySvr(外部系统)。可以看得出这段代码的本意,是期望将当前系统的业务和外部系统的业务,合并到一个事务里面,要么一起成功提交,要么一起失败回滚,从而保持两个系统的一致性。

之所以没能达到预期,直接原因是,在失败(异常)回滚的时候,只回滚了本地事务,而没有回滚远端系统的状态变化。按这个思路去考虑,似乎只要加一个 PaySvr::rollbackRequest($order->id) 好像就可以解决问题。但仔细想想就会发现远没这么简单,从业务上来说,如果已经给用户付款了,那实际上就是要给用户退款,而往往这时候是掉单(支付请求结果未知),我们又无法直接给用户退款;更极端一点,如果这个rollback请求也超时了呢,那本地可以rollback吗?

这就是分布式事务棘手的地方了,只靠这样的逻辑是无法保证跨系统的一致性的。解决这个问题的方法是引入两段式提交(2 Phase Commit,常常简写为2PC),其基本逻辑是,在两个系统分别完成业务,然后再分别提交。

例如我们上面的解决方案,实际上就是2PC的一个实现:我们把业务需求作为一整个事务,它可以拆成两个子事务(第三方支付通道完成代扣,在XX业务系统记录支付请求成功并修改相应业务状态),每个子事务又分成两个阶段:第一阶段,是在本地先记录支付请求(状态为待确认),并向第三方支付发出代扣请求(结果不确定);第二阶段,在确认第三方代扣成功以后,修改本地支付请求的状态修改为成功,或者是代扣结果为失败,本地支付请求状态记为失败。两个阶段都完成,这个事务才是真的完成了。

3. Case变种

仔细思考我们曾经实现过的需求,可能会在很多看似不起眼的地方发现分布式事务,例如我们在的存管匹配系统里面,就有这样一个Case。

由于与XX银行存管系统交互的延迟比较大,所以我们的匹配系统实现是异步的,匹配系统在撮合了资金和资产以后,会生成一条债权关系记录在本地,随后再发送到XX银行执行资金的划拨。为了提高执行的效率,我们希望在债权关系生成以后,尽快执行资金的划拨,因此我们会把资金划拨的指令通过LPush放进Redis的list里;List的另一端,那些使用BLPOP监听数据的worker会立刻被激活去执行。

如果没有仔细思考,代码可能会这么写:
#匹配系统
function matcher() {
    $dbTrans = $db->beginTransaction();
    try {
        foreach (matchCapitalAndProject() as $match_result) {
            list($capital_id, $project_id, $amount) = $match_result;
            $relation = Relation::model()->create($capital_id, $project_id, $amount);
            $redis->lPush($relation->id);
        }
        $dbTrans->commit();
    } catch (Exception $e) {
        $dbTrans->rollback();
    }
}

#Worker
function Worker() {
    while (true) {
        $id = $redis->brPop();
        $relation = Relation::model()->findByPk($id);
        if ($relation) {
            BankApi::invest($relation->capital_id, $relation->project_id, $amount);
        }
    }
}

在实际执行这段代码的时候,如果没有仔细测试(尤其是在有补单逻辑,捞出未执行成功的划拨指令再发送给银行),可能就不会发现,实际上有很多指令并不是马上被执行的,因为relation_id被送进list以后,worker马上就会读出来执行,但这时事务可能还没有提交。但这只是影响了业务的效率,还没有对业务的正确性产生影响。

为了修复这个问题,似乎可以这么做:把 [capital_id, project_id, amount] 发送到redis,worker直接取出执行,这样就不用从数据库读取relation,保证尽快将请求发送到银行。但如果因为某些原因,事务最终没有被提交呢?找银行rollback这些指令的执行,那就麻烦多了。

正确的做法是,在事务提交了以后,再lPush到Redis里:
#匹配系统
function matcher() {
    $arr_relation = [];
    $dbTrans = $db->beginTransaction();
    try {
        foreach (matchCapitalAndProject() as $match_result) {
            list($capital_id, $project_id, $amount) = $match_result;
            $relation = Relation::model()->create($capital_id, $project_id, $amount);
            $arr_relation[] = $relation;
        }
        $dbTrans->commit();
    } catch (Exception $e) {
        $dbTrans->rollback();
    }
    foreach ($arr_relation as $relation) {
        $redis->lPush($relation->id);
    }
}

注:foreach要放到try-catch后面。

最后想说,我相信有很多同学知道这个Case,或者就算不知道也不会犯这样的错误,因此也许会觉得没必要专门揪出来这样分享 —— 但“知识的诅咒”就是这样,“我会的东西都是简单的”,然而对于没有踩过坑的同学来说,其实都是宝贵的经验;另一方面,有些别人觉得简单的问题、踩过的坑,也许自己是不知道的。所以希望大家都能分享自己在工作学习中踩过的坑、解决过的问题,互相交流,互相提高。
分页: 2/23 第一页 上页 1 2 3 4 5 6 7 8 9 10 下页 最后页 [ 显示模式: 摘要 | 列表 ]