用 mutex、channel、自旋锁进行同步

水深无声 2022-04-11 04:44 122阅读 0赞

## 背景 ##

Goroutine 是原生支持的一种轻量级线程——协程。协程的优势在于上下文切换的代价非常小，但进程中执行数以万计的协程，依旧能够保持很高的性能。

进程、线程、协程的关系和区别：

*  进程拥有独立的堆和栈，既不共享堆，也不共享栈，由操作系统负责调度。
 *  线程拥有独立的栈和共享的堆，由操作系统负责调度（内核线程）。
 *  协程拥有独立的栈和共享的堆，有 golang 的调度器负责调度。

想象一个场景，假设需要将 for 循环遍历的值填充到 map 中，你会怎么做？如下的填充方式吗？

m := make(map[int]int, 20)
    for i := 0; i < 20; i++ { 
        m[i] = i
    }

既然 go 已经为开发者提供了 goroutine 这么便利的工具，本着将工具应用的极限的精神，我猜你肯定也想到了这么写：

m := make(map[int]int, 20)
    mu := sync.Mutex{ }
    for i := 0; i < 20; i++ {   
        go func(i int){ 
            m.Lock()
            m[i] = i
            m.Unlock()
        }(i)
    }

上述问题可以概括为数据同步的问题，可以使用 mutex、channel、自旋锁等多种解法，对于求知欲有渴求的你，肯定好奇哪种解法性能更好。

## mutex、channel、自旋锁对比 ##

### 测试结果 ###

goos: darwin
    goarch: amd64
    pkg: sync/test
    BenchmarkSyncMapByMutex100-4          	   30000	     54633 ns/op
    BenchmarkSyncMapByMutex1000-4         	    2000	    609157 ns/op
    BenchmarkSyncMapByMutex10000-4        	     100	  11865318 ns/op
    BenchmarkSyncMapByMutex100000-4       	      20	  85326451 ns/op
    
    BenchmarkSyncMapByChannel100-4        	   30000	     51579 ns/op
    BenchmarkSyncMapByChannel1000-4       	    5000	    439112 ns/op
    BenchmarkSyncMapByChannel100000-4     	      50	  45009265 ns/op
    BenchmarkSyncMapByChannel1000000-4    	       1	18901138625 ns/op
    
    BenchmarkSyncMapBySpinLock100-4       	   20000	     62983 ns/op
    BenchmarkSyncMapBySpinLock1000-4      	    2000	    625292 ns/op
    BenchmarkSyncMapBySpinLock100000-4    	      20	  88903819 ns/op
    BenchmarkSyncMapBySpinLock1000000-4   	       1	1265857442 ns/op
    PASS
    ok  	sync/test	43.544s

为将测试结果展示的更加清晰，笔者强行对结果进行了分割，可以看出：

*  mutex 的解决方案更优于自旋锁的解决方案
 *  channel 的解决方案在使用的 goroutine 较少的时候优于 mutex、自旋锁，但当 goroutine 的数量达到几十万到几百万的时候性能下降非常严重
 *  具体使用哪种解决方式，可以根据场景进行选择

### 测试代码 ###

1.  bench\_test.go

package test
    
    import (
       "example/sync/spin_lock"
       "strconv"
       "sync"
       "testing"
    )
    
    func syncMapByMutex(count int) { 
       var (
       	m  = make(map[string]int, 0)
       	mu = sync.Mutex{ }
       	wg = sync.WaitGroup{ }
       )
       for i := 0; i < count; i++ { 
       	wg.Add(1)
       	go func(i int) { 
       		defer wg.Done()
       		mu.Lock()
       		key := strconv.Itoa(i)
       		m[key] = i
       		mu.Unlock()
       	}(i)
       }
       wg.Wait()
    }
    func syncMapBySpinLock(count int) { 
       var (
       	m  = make(map[string]int, 0)
       	mu = spinlock.NewSpinLock()
       	wg = sync.WaitGroup{ }
       )
       for i := 0; i < count; i++ { 
       	wg.Add(1)
       	go func(i int) { 
       		defer wg.Done()
       		mu.Lock()
       		key := strconv.Itoa(i)
       		m[key] = i
       		mu.Unlock()
       	}(i)
       }
       wg.Wait()
    }
    
    func syncMapByChannel(count int) { 
       type result struct { 
       	key   string
       	value int
       }
       var (
       	m  = make(map[string]int, 0)
       	ch = make(chan result, count)
       )
       for i := 0; i < count; i++ { 
       	go func(i int) { 
       		key := strconv.Itoa(i)
       		ch <- result{ 
       			key:   key,
       			value: i,
       		}
       	}(i)
       }
       for r := range ch { 
       	count--
       	m[r.key] = r.value
       	if count == 0 { 
       		break
       	}
       }
    }
    
    func BenchmarkSyncMapByMutex100(b *testing.B) { 
       for i := 0; i < b.N; i++ { 
       	syncMapByMutex(100)
       }
    }
    func BenchmarkSyncMapByMutex1000(b *testing.B) { 
       for i := 0; i < b.N; i++ { 
       	syncMapByMutex(1000)
       }
    }
    func BenchmarkSyncMapByMutex10000(b *testing.B) { 
       for i := 0; i < b.N; i++ { 
       	syncMapByMutex(10000)
       }
    }
    func BenchmarkSyncMapByMutex100000(b *testing.B) { 
       for i := 0; i < b.N; i++ { 
       	syncMapByMutex(100000)
       }
    }
    
    func BenchmarkSyncMapByChannel100(b *testing.B) { 
       for i := 0; i < b.N; i++ { 
       	syncMapByChannel(100)
       }
    }
    func BenchmarkSyncMapByChannel1000(b *testing.B) { 
       for i := 0; i < b.N; i++ { 
       	syncMapByChannel(1000)
       }
    }
    func BenchmarkSyncMapByChannel100000(b *testing.B) { 
       for i := 0; i < b.N; i++ { 
       	syncMapByChannel(100000)
       }
    }
    func BenchmarkSyncMapByChannel1000000(b *testing.B) { 
       for i := 0; i < b.N; i++ { 
       	syncMapByChannel(1000000)
       }
    }
    
    func BenchmarkSyncMapBySpinLock100(b *testing.B) { 
       for i := 0; i < b.N; i++ { 
       	syncMapBySpinLock(100)
       }
    }
    func BenchmarkSyncMapBySpinLock1000(b *testing.B) { 
       for i := 0; i < b.N; i++ { 
       	syncMapBySpinLock(1000)
       }
    }
    func BenchmarkSyncMapBySpinLock100000(b *testing.B) { 
       for i := 0; i < b.N; i++ { 
       	syncMapBySpinLock(100000)
       }
    }
    func BenchmarkSyncMapBySpinLock1000000(b *testing.B) { 
       for i := 0; i < b.N; i++ { 
       	syncMapBySpinLock(1000000)
       }
    }

1.  spin\_lock.go 自旋锁的实现

package spinlock
    
    import (
    	"runtime"
    	"sync/atomic"
    )
    
    type SpinLock uint32
    
    func (sl *SpinLock) Lock() { 
    	for !atomic.CompareAndSwapUint32((*uint32)(sl), 0, 1) { 
    		runtime.Gosched()
    	}
    }
    
    func (sl *SpinLock) Unlock() { 
    	atomic.StoreUint32((*uint32)(sl), 0)
    }
    
    func NewSpinLock() *SpinLock { 
    	var lock SpinLock
    	return &lock
    }

笔者不记得在哪里看到过一句话 go 的调度机制实现了将 IO 密集型转换成 CPU 密集型，觉得总结的很有道理，在这里记录一下，以上的测试均为笔者个人观点，如有错误欢迎指出。

# 参考资料 #

*  [Go 语言中 goroutine 的分析][Go _ goroutine]

[Go _ goroutine]: https://www.jianshu.com/p/be04990895c7