如何设计一个好的自旋锁？-摩杜云开发者社区

最近在阅读ants源码时，偶然发现ants有自己研发的自旋锁，之前ants一直用的官方的mutex，后面才改为用自旋锁。

微信公众号原文地址：自旋锁设计

根据自旋锁的定义，当一个线程在获取锁的时候，如果锁已经被其它线程获取，那么该线程将循环判断锁是否能够被成功获取，直到获取到锁才会退出循环。那么问题来了，单个协程需要不断循环去检测锁状态，这样看，似乎会有一定的资源浪费。那为什么还需要自旋锁呢？

这就要说到自旋锁的应用场景了，对于很小的原子操作，单个协程不太可能长时间持有锁的情况，比如并发场景下对切片的操作。在这种场景下，使用mutex 这种重锁去解决数据竞争问题显然是不可取的，使用自旋锁性能更好，因为自选锁不会导致上下文切换。

聊完自旋锁的定义和使用场景，我们聊聊自旋锁的实现。自旋锁按照定义来其实很好实现，原理也简单，就是一个for循环，不断轮询某个变量的状态。简化版代码（go）如下：

type sLock uint32

//自旋，轮询锁状态
func (sl *sLock) Lock() {
    for !atomic.CompareAndSwapUint32((*uint32)(sl), 0, 1) {
        runtime.Gosched()
    }
}
func (sl *sLock) Unlock() {
    atomic.StoreUint32((*uint32)(sl), 0)
}
func NewSpinLock() sync.Locker {
    var l sLock
    return &l
}

看代码发现整个逻辑其实挺简单的，不过有个地方需要注意，就是第五行的runtime.Gosched()，这个函数主要是将当前协程的时间片让出来，通过GMP调度，让其他协程占用时间片，至于何时恢复，取决于当前协程下次被GMP调度，绑定线程这个时机点。

然后我们谈谈自旋锁的优化，假设现在机器CPU核心数很多，能保证runtime.Gosched()后，立刻通过GMP绑定新的核心，占用时间片，那和直接CAS轮询锁状态没什么区别了，这种情况下就会导致锁冲突次数过多。

所以为了减少锁冲突，我们可以引入指数退避算法。即每次查询完锁状态后，使让出CPU次数不断变大，成指数形式，比如第一次查询完，让出CPU次数为1，第二次查询完，让出CPU次数为2。。。

一般来说最高让出CPU次数最高为16，太高了的话也不好，太高和互斥锁的性能损耗差不多了。

这里给出优化后的自旋锁代码（ants的自旋锁实现）：

type spinLock uint32

const maxBackoff = 16

func (sl *spinLock) Lock() {
    backoff := 1
    for !atomic.CompareAndSwapUint32((*uint32)(sl), 0, 1) {
        // Leverage the exponential backoff algorithm, see https://en.wikipedia.org/wiki/Exponential_backoff.
        for i := 0; i < backoff; i++ {
            runtime.Gosched()
        }
        if backoff < maxBackoff {
            backoff <<= 1
        }
    }
}

func (sl *spinLock) Unlock() {
    atomic.StoreUint32((*uint32)(sl), 0)
}

// NewSpinLock instantiates a spin-lock.
func NewSpinLock() sync.Locker {
    return new(spinLock)
}

优化后使用指数退避算法，能很好的适用各种场景。另外，指数退避算法其实在很多业务场景也用得到，比如登录。

总结

自旋锁就是让单个协程不断去轮询锁状态，直到获取锁。
自旋锁在极端情况下会存在锁冲突过多，可以通过指数退避算法进行优化。