问题背景
最近有同事说平台的某个服务出现超时异常，让我帮忙看下原因。我进入平台后触发了该服务，并没有发现超时异常，那可能是在特定操作场景下会出现或者是一个非必现问题。跟同事沟通之后，找到了复现的流程。
接下来开始定位问题可能出现在哪个模块，进入后台之后，触发服务通过top命令检查后台资源使用情况，发现CPU、内存和负载情况都是正常的。嗯，此种方法分析不出什么。
既然已知道异常服务，那可以从这里入手进行分析，又与同事沟通一番，确定了与该服务相关的一些后台模块，接下来重点排查这些模块。首先通过pprof抓取了这些模块的堆栈日志,Go提供了net/http/pprof和runtime/pprof两个包用于性能测评分析，前者通常用于web服务器的性能分析，后者通常用于普通代码的性能分析。下面是出现问题的参考日志，关键点已包含其中，因为原日志不方便展示。
排查方法
日志中出现了sync.(*Mutex).Lock的关键字，怀疑是程序死锁了，结合代码，在mutex.go 138行进入了lockSlow,然后goroutine被挂起了，也就是说在doConcat中获取锁失败了。然后搜索下程序中有哪些地方用到了doConcat中的锁，这些都是可疑的地方。最后分析这些的可疑的地方，基本可以确定是在哪里死锁了。当然这里的日志只是个示例，实际堆栈文件可能很大，有很多地方需要分析，分析起来也比较麻烦，我们可以写个脚本提取出关键的goroutine以及上下文信息，再进行分析。不过github上已有动态检测死锁的工具https://github.com/sasha-s/go-deadlock，使用方法见下面的死锁检测工具小节。
问题本质
上面问题的根因是死锁导致的，死锁也是计算机中常见出现的问题。这里再来回顾下死锁的定义：
❝死锁是指两个或两个以上的进程或线程在执行的过程中，由于竞争资源或者彼此通信而造成的一种阻塞程序不能推进的现象，如果不借助外部作用，它们会无法推进下去。 ❞
产生死锁的原因，通常是系统资源不足、程序的执行顺序不当和资源分配不当等导致的。
产生死锁有四大必要条件，注意是必要条件，就是说如果出现了死锁，下面四个条件必定成立，如果有其中至少一个条件不满足，则不会出现死锁。
- 互斥条件：一个资源每次只能被一个进程或线程使用
- 请求与保持条件：一个进程或线程因请求资源而阻塞时，对已获得的资源保持占有不释放
- 不可剥夺条件：进行或线程已获得的资源，在没有使用完之前，不能强行剥夺
- 循环等待条件：若干进程或线程之间形成一种头尾相接的循环等待资源状态
如何预防死锁？可以通过破坏死锁产生的4个必要条件来预防死锁，打破其中之一就可以避免死锁产生。那挑选比较容易的进行破坏，由于资源互斥是资源使用时的固有特性无法改变，所以破坏互斥条件这条直接放弃。
- 破坏循环等待条件，在申请资源获取锁时保持一致的顺序，例如下面的程序,goroutine 1先获取A锁然后获取B锁，而goroutine 2先获取B锁然后获取A锁，会形成环形依赖，我们可以调整程序的顺序，让它们获取锁的顺序保持一致。
// goroutine 1
A.Lock()
...
B.Lock()
// goroutine 2
B.Lock()
...
A.Lock()
- 破坏不可剥夺条件，通过外部程序检测，如果出现了死锁，强行释放掉锁，例如在数据库中，检测到死锁之后强制让某个事务回滚。
- 破坏请求与保持条件，进程或线程执行前，先一次性申请其在整个运行期间所需的全部资源，或者一段时间获取不到资源尝试主动放弃已经获得的资源。
如何避免
上个小节只是从原理层面说了怎么预防死锁，在我们的实际工作中，怎么做才能够尽早发现死锁问题并进行消除呢？小编想到了下面三种方法，codereview、编写单元测试和通过死锁检测工具检查。
codereview
通过代码评审，有经验的工程师对代码进行review之后，一些比较明显的存在死锁代码逻辑是容易发现的，当然有些逻辑隐藏的比较深，一般很难发现。往往改动代码引发的死锁问题比较容易出现，像本文中出现的问题就是代码改动导致的，添加功能需求的时候关注点集中在了业务逻辑上，容易忽视锁的问题。
编写单元测试
编写单元测试，执行单元测试对于死锁问题是很容易发现的，因为在运行单元测试的过程中，程序会卡死结束不了，可以很快暴露问题。
死锁检测工具
上面两种方法是通过流程制度来约束减少死锁问题的发生，通过死锁检测工具自动帮助我们检测也是一种有效的手段。这里介绍一款在程序运行的时候检测是否可能存在死锁的工具，代码地址https://github.com/sasha-s/go-deadlock。注意这个检测工具不要在生产环境中使用，因为它的实现是有性能开销的。
使用实例如下，像用dead.Mutex替代sync.Mutex,go-deadlock提供了dead.Mutex和dead.RWMutex. 下面的程序存在锁重入，如果用的是sync.Mutex会出现卡死，下面程序会检测了存在死锁，直接退出了。
package main
import (
"fmt"
"github.com/sasha-s/go-deadlock"
"time"
)
// 锁重入测试
func reentrantLockTest(mu *deadlock.Mutex){
fmt.Println("called reentrantLockTest")
mu.Lock()
defer mu.Unlock()
do(mu)
}
func do(mu *deadlock.Mutex){
fmt.Println("called do")
mu.Lock()
defer mu.Unlock()
// do something
fmt.Println("do something")
}
func main(){
var mu deadlock.Mutex
reentrantLockTest(&mu)
time.Sleep(time.Minute)
}
上面的程序运行结果如下，输出结果指出了代码中存在Recursive locking，并指明了死锁的逻辑存在的两处位置。
再来看另一种情况，两个goroutine加锁顺序不当导致的死锁问题。goroutine 1和goroutine 2都在对lock1和lock2加锁，不过它们获取锁的顺序是不同的，一个先获取lock1在获取lock2,另一个先获取lock2在获取lock1.这会导致它们形成了一个环，都无法推进。
代码语言：javascript复制package main
import (
"fmt"
"github.com/sasha-s/go-deadlock"
"time"
)
func main(){
var (
lock1,lock2 deadlock.Mutex
)
go func(){
lock1.Lock()
defer lock1.Unlock()
time.Sleep(time.Second)
lock2.Lock()
defer lock2.Unlock()
fmt.Println("func1 end")
}()
go func(){
lock2.Lock()
defer lock2.Unlock()
time.Sleep(time.Second)
lock1.Lock()
defer lock1.Unlock()
fmt.Println("func2 end")
}()
time.Sleep(time.Second*10)
}
运行上面的程序，也检测出了存在死锁的问题，并提示是加锁顺序不一致导致的死锁。
还有一种情况，程序上锁之后忘了释放，导致其他获取此锁的goroutine一直卡死，这种情况go-deadlock是通过设置goroutine卡死的时间来提示可能存在死锁，默认超时时间是30秒。之所以说提示可能存在死锁，是因为存在加锁之后之后的处理逻辑执行时间很长，然后才释放，会被误判为死锁。
代码语言：javascript复制package main
import (
"fmt"
"github.com/sasha-s/go-deadlock"
"time"
)
func main(){
var lock deadlock.Mutex
go func(){
lock.Lock()
// 不释放锁
// defer lock.Unlock()
fmt.Println("func1 end")
}()
time.Sleep(time.Second)
go func(){
lock.Lock()
defer lock.Unlock()
fmt.Println("func2 end")
}()
select {
}
}
上面的程序加锁之后没有释放锁，会导致第二个goroutine一直无法运行，运行输出结果如下：
做一个总结，对上面三种情形，go-deadlock的处理方法如下：
- 情形1，存在锁重入，即goroutine 1获取了A互斥锁，在没有释放前，然后又在后序处理中尝试获取A互斥锁。
处理方法：go-deadlock用一个map记录了到当前为止所有还未释放的锁，map的key为*deadlock.Mutex类型，value为堆栈信息和gid信息。然后检查如果map中锁已存在，并且当前尝试获取锁的goroutine id即gid相同，说明存在重入获取锁。
func (l *lockOrder) preLock(stack []uintptr, p interface{}) {
...
gid := goid.Get()
l.mu.Lock()
for b, bs := range l.cur {
if b == p {
if bs.gid == gid {
// 处理重入情形
}
continue
}
...
}
l.mu.Unlock()
}
- 情形2，获取锁的顺序不当导致的死锁，goroutine 1先获取A锁然后获取B锁，goroutine 2先获取B锁，然后获取A锁。处理方法：go-deadlock中用order map记录了获取锁的先后顺序，key为有序的两个锁，A锁-B锁会存入order,当又存在B锁-A锁时，说明存在顺序不一致。
func (l *lockOrder) preLock(stack []uintptr, p interface{}) {
...
gid := goid.Get()
l.mu.Lock()
for b, bs := range l.cur {
...
if s, ok := l.order[beforeAfter{p, b}]; ok {
....
// 情形2，存在顺序不一致
}
// 记录下顺序
l.order[beforeAfter{b, p}] = ss{bs.stack, stack}
...
}
l.mu.Unlock()
}
在如下的环路中，lockA --> lock B --> lock C, l.order会记录两两lock之前的先后关系，例如会存储下面的顺序，现在goroutine 1重新调度时，会检查l.order中是否存 lock A-lock B或C, 如果有说明顺序不一致。
代码语言：javascript复制l.order[lock A-lock B]
l.order[lock A-lock C]
l.order[lock B-lock C]
3.情形3：获取了锁但没有释放，会导致其他goroutine一直拿不到，这种情况没法严格检查，go-deadlock通过检查goroutine卡主时间来判断的，默认是30秒。
代码语言：javascript复制func lock(lockFn func(), ptr interface{}) {
...
if Opts.DeadlockTimeout <= 0 {
lockFn()
} else {
ch := make(chan struct{})
currentID := goid.Get()
// 开启一个goroutine 启动定时器检查
go func() {
for {
t := time.NewTimer(Opts.DeadlockTimeout)
defer t.Stop()
select {
case <-t.C:
...
case <-ch:
return
}
}
}()
lockFn()
postLock(stack, ptr)
close(ch)
return
}
postLock(stack, ptr)
}
