Go并发编程部分知识

整理了一下Go并发编程的一些知识点，有点乱，勿喷

锁

Go的sync提供了互斥锁sync.Mutex和读写锁sync.RWMutex

互斥锁 (sync.Mutex)

互斥顾名思义。使用互斥锁的两段程序不能同时运行。

• Lock 加锁
• Unlock 释放锁

我们可以通过在代码前调用Lock方法，在代码后面使用Unlock方法保证代码的互斥操作。类似C语言的PV操作。一个Go协程调用Lock方法获得锁之后，其他协程都会堵塞在Lock方法，直到释放锁。

读写锁(sync.RWMutex)

在实际应用中，很大一部分场景是，写操作 $\lt$ 读操作， 所以如果对于一个读操作我们加锁，很有可能堵塞其他的读操作，而常识也知道，读操作不会影响数据。

所以如果想要保证读操作的安全性，只需要保证并发读操作的时候没有写操作即可。

Go语言的读写锁，分为读锁和写锁。读锁是允许同时进行的，但是写锁是互斥的。

• 读锁不互斥，没有写锁的情况下，读锁不堵塞，允许多个协程获取读锁。
• 写锁是互斥的，存在写锁，其他协程读锁会堵塞。
• 读锁和写锁是互斥的，如果存在读锁、写锁堵塞，存在写锁、读锁堵塞。

sync.RWMutex提供了四个方法

• Lock 加写锁
• Unlock 释放写锁
• RLock 加读锁
• RUnlock 释放读锁

这个读写锁是为了解决读多写少时的问题。

性能比较

为了方便测试，都是实现了接口。

假设一个读写操作都是1μs

• Lock

  package LockTest
  
  import (
  	"sync"
  	"time"
  )
  
  type MUT interface {
  	Read()
  	Write()
  }
  
  type Lock struct {
  	value int
  	mu    sync.Mutex
  }
  
  func (L *Lock) Read() {
  	L.mu.Lock()
  	_ = L.value
  	time.Sleep(time.Microsecond)
  	L.mu.Unlock()
  }
  
  func (L *Lock) Write() {
  	L.mu.Lock()
  	L.value++
  	time.Sleep(time.Microsecond)
  	L.mu.Unlock()
  }
  
  

• RWLock

  package LockTest
  
  import (
  	"sync"
  	"time"
  )
  
  type RWLock struct {
  	value int
  	mu    sync.RWMutex
  }
  
  func (L *RWLock) Write() {
  	L.mu.Lock()
  	L.value++
  	time.Sleep(time.Microsecond)
  	L.mu.Unlock()
  }
  
  func (L *RWLock) Read() {
  	L.mu.RLock()
  	_ = L.value
  	time.Sleep(time.Microsecond)
  	L.mu.RUnlock()
  }
  

• test

  package Test
  
  import (
  	"Go-Concurrency-Test_/LockTest"
  	"sync"
  	"testing"
  )
  
  func test(b *testing.B, mut LockTest.MUT, readNum, writeNum int) {
  	for i := 0; i < b.N; i++ {
  		var wg sync.WaitGroup
  		for j := 0; j <= readNum; j++ {
  			wg.Add(1)
  			go func() {
  				mut.Read()
  				wg.Done()
  			}()
  		}
  
  		for j := 0; j <= writeNum; j++ {
  			wg.Add(1)
  			go func() {
  				mut.Write()
  				wg.Done()
  			}()
  		}
  		wg.Wait()
  	}
  }
  func BenchmarkReadMore(b *testing.B)    { test(b, &LockTest.Lock{}, 8000, 1000) }
  func BenchmarkReadRWMore(b *testing.B)  { test(b, &LockTest.RWLock{}, 8000, 1000) }
  func BenchmarkWriteMore(b *testing.B)   { test(b, &LockTest.Lock{}, 8000, 1000) }
  func BenchmarkWriteRWMore(b *testing.B) { test(b, &LockTest.RWLock{}, 8000, 1000) }
  func BenchmarkLock(b *testing.B)        { test(b, &LockTest.Lock{}, 5000, 5000) }
  func BenchmarkRWLock(b *testing.B)      { test(b, &LockTest.RWLock{}, 5000, 5000) }
  
  

运行测试

goos: linux
goarch: amd64    
BenchmarkReadMore
BenchmarkReadMore-12       	       6	 262894692 ns/op
BenchmarkReadRWMore
BenchmarkReadRWMore-12     	      45	  24374298 ns/op
BenchmarkWriteMore
BenchmarkWriteMore-12      	       4	 258559273 ns/op
BenchmarkWriteRWMore
BenchmarkWriteRWMore-12    	      66	  22369855 ns/op
BenchmarkLock
BenchmarkLock-12           	       4	 293963420 ns/op
BenchmarkRWLock
BenchmarkRWLock-12         	      10	 123536948 ns/op

sync.mutex分析

原文章

互斥锁有两种状态：正常状态和饥饿状态。

在正常状态下，所有等待锁的 goroutine 按照FIFO顺序等待。唤醒的 goroutine 不会直接拥有锁，而是会和新请求锁的 goroutine 竞争锁的拥有。新请求锁的 goroutine 具有优势：它正在 CPU 上执行，而且可能有好几个，所以刚刚唤醒的 goroutine 有很大可能在锁竞争中失败。在这种情况下，这个被唤醒的 goroutine 会加入到等待队列的前面。 如果一个等待的 goroutine 超过 1ms 没有获取锁，那么它将会把锁转变为饥饿模式。

在饥饿模式下，锁的所有权将从 unlock 的 goroutine 直接交给交给等待队列中的第一个。新来的 goroutine 将不会尝试去获得锁，即使锁看起来是 unlock 状态, 也不会去尝试自旋操作，而是放在等待队列的尾部。

如果一个等待的 goroutine 获取了锁，并且满足一以下其中的任何一个条件：(1)它是队列中的最后一个；(2)它等待的时候小于1ms。它会将锁的状态转换为正常状态。

正常状态有很好的性能表现，饥饿模式也是非常重要的，因为它能阻止尾部延迟的现象。

空结构体

package main

import (
	"fmt"
	"unsafe"
)

func main() {
	fmt.Println(unsafe.Sizeof(struct{}{}))
}

$ go run main.go
0

空结构体实例struct{} 不占据任何内存空间。

作用

实现Set

Go语言标准库没有提供set的实现，通常用map来实现。

实际上在使用set的时候，我们只关注key而不需要value，因此可以直接把value设置成struct{}

type Set map[string]struct{}

func (s Set) Exists(key string) bool {
	_, ok := s[key]
	return ok
}

func (s Set) Add(key string) {
	s[key] = struct{}{}
}

func (s Set) Delete(key string) {
	delete(s, key)
}

不发送数据的channel

func work(ch chan struct{}){
    <- ch
    fmt.Println("1")
    close(ch)
}

func main(){
    ch := make(chan struct{})
    go work(ch)
    ch <- struct{}{}
}

goroutine

控制协程数量

利用channel缓冲区

func MyRoutine() {
	var wg sync.WaitGroup

	ch := make(chan struct{}, 5)
	for i := 0; i <= 100; i++ {
		ch <- struct{}{}
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			log.Println(i)
			time.Sleep(time.Second)
			<-ch
		}(i)
	}
	wg.Wait()
}

• make(chan sturct{}, 3) 创建缓冲区大小为3的channel，在没有被接收的情况下，最多发送三个消息就会被堵塞。