sync.Map 并發(fā)安全的Map

反例如下，兩個Goroutine分別讀寫。

func unsafeMap(){
	var wg sync.WaitGroup
	m := make(map[int]int)
	wg.Add(2)
	go func() {
		defer wg.Done()
		for i := 0; i < 10000; i++ {
			m[i] = i
		}
	}()

	go func() {
		defer wg.Done()
		for i := 0; i < 10000; i++ {
			fmt.Println(m[i])
		}
	}()
	wg.Wait()
}

執(zhí)行報錯：

0
fatal error: concurrent map read and map write

goroutine 7 [running]:
runtime.throw({0x10a76fa, 0x0})
......

使用并發(fā)安全的Map

func safeMap() {
	var wg sync.WaitGroup
	var m sync.Map
	wg.Add(2)
	go func() {
		defer wg.Done()
		for i := 0; i < 10000; i++ {
			m.Store(i, i)
		}
	}()

	go func() {
		defer wg.Done()
		for i := 0; i < 10000; i++ {
			fmt.Println(m.Load(i))
		}
	}()
	wg.Wait()
}

• 不需要make就能使用
• 還內(nèi)置了Store、Load、LoadOrStore、Delete、Range等操作方法，自行體驗。

sync.Once 只執(zhí)行一次

很多場景下我們需要確保某些操作在高并發(fā)的場景下只執(zhí)行一次，例如只加載一次配置文件、只關(guān)閉一次通道等。

init 函數(shù)是當(dāng)所在的 package 首次被加載時執(zhí)行，若遲遲未被使用，則既浪費了內(nèi)存，又延長了程序加載時間。

sync.Once 可以在代碼的任意位置初始化和調(diào)用，因此可以延遲到使用時再執(zhí)行，并發(fā)場景下是線程安全的。

在多數(shù)情況下，sync.Once 被用于控制變量的初始化，這個變量的讀寫滿足如下三個條件：

• 當(dāng)且僅當(dāng)?shù)谝淮卧L問某個變量時，進行初始化（寫）；
• 變量初始化過程中，所有讀都被阻塞，直到初始化完成；
• 變量僅初始化一次，初始化完成后駐留在內(nèi)存里。

var loadOnce sync.Once
var x int
for i:=0;i<10;i++{
    loadOnce.Do(func() {
        x++
    })
}
fmt.Println(x)

輸出

1

sync.Cond 條件變量控制

sync.Cond 基于互斥鎖/讀寫鎖，它和互斥鎖的區(qū)別是什么呢？

互斥鎖 sync.Mutex 通常用來保護臨界區(qū)和共享資源，條件變量 sync.Cond 用來協(xié)調(diào)想要訪問共享資源的 goroutine。

也就是在存在共享變量時，可以直接使用sync.Cond來協(xié)調(diào)共享變量，比如最常見的共享隊列，多消費多生產(chǎn)的模式。

我一開始也很疑惑為什么不使用channel和select的模式來做生產(chǎn)者消費者模型（實際上也可以），這一節(jié)不是重點就不展開討論了。

創(chuàng)建實例

func NewCond(l Locker) *Cond

NewCond 創(chuàng)建 Cond 實例時，需要關(guān)聯(lián)一個鎖。

廣播喚醒所有

func (c *Cond) Broadcast()

Broadcast 喚醒所有等待條件變量 c 的 goroutine，無需鎖保護。

喚醒一個協(xié)程

func (c *Cond) Signal()

Signal 只喚醒任意 1 個等待條件變量 c 的 goroutine，無需鎖保護。

等待

func (c *Cond) Wait()

每個 Cond 實例都會關(guān)聯(lián)一個鎖 L（互斥鎖 *Mutex，或讀寫鎖 *RWMutex），當(dāng)修改條件或者調(diào)用 Wait 方法時，必須加鎖。

舉個不恰當(dāng)?shù)睦樱瑢崿F(xiàn)一個經(jīng)典的生產(chǎn)者和消費者模式，但有先決條件：

• 邊生產(chǎn)邊消費，可以多生產(chǎn)多消費。
• 生產(chǎn)后通知消費。
• 隊列為空時，暫停等待。
• 支持關(guān)閉，關(guān)閉后等待消費結(jié)束。
• 關(guān)閉后依然可以生產(chǎn)，但無法消費了。

var (
	cnt          int
	shuttingDown = false
	cond         = sync.NewCond(&sync.Mutex{})
)

• cnt 為隊列，這里直接用變量代替了，變量就是隊列長度。
• shuttingDown 消費關(guān)閉狀態(tài)。
• cond 現(xiàn)成的隊列控制。

生產(chǎn)者

func Add(entry int) {
	cond.L.Lock()
	defer cond.L.Unlock()
	cnt += entry
	fmt.Println("生產(chǎn)咯，來消費吧")
	cond.Signal()
}

消費者

func Get() (int, bool) {
	cond.L.Lock()
	defer cond.L.Unlock()
	for cnt == 0 && !shuttingDown {
		fmt.Println("未關(guān)閉但空了，等待生產(chǎn)")
		cond.Wait()
	}
	if cnt == 0 {
		fmt.Println("關(guān)閉咯，也消費完咯")
		return 0, true
	}
	cnt--
	return 1, false
}

關(guān)閉程序

func Shutdown() {
	cond.L.Lock()
	defer cond.L.Unlock()
	shuttingDown = true
	fmt.Println("要關(guān)閉咯，大家快消費")
	cond.Broadcast()
}

主程序

var wg sync.WaitGroup
	wg.Add(2)
	time.Sleep(time.Second)
	go func() {
		defer wg.Done()
		for i := 0; i < 10; i++ {
			go Add(1)
			if i%5 == 0 {
				time.Sleep(time.Second)
			}
		}
	}()
	go func() {
		defer wg.Done()
		shuttingDown := false
		for !shuttingDown {
			var cur int
			cur, shuttingDown = Get()
			fmt.Printf("當(dāng)前消費 %d, 隊列剩余 %d \n", cur, cnt)
		}
	}()
	time.Sleep(time.Second * 5)
	Shutdown()
	wg.Wait()

• 分別創(chuàng)建生產(chǎn)者與消費者。
• 生產(chǎn)10個，每5個休息1秒。
• 持續(xù)消費。
• 主程序關(guān)閉隊列。

輸出

生產(chǎn)咯，來消費吧
當(dāng)前消費 1, 隊列剩余 0?
未關(guān)閉但空了，等待生產(chǎn)
生產(chǎn)咯，來消費吧
生產(chǎn)咯，來消費吧
當(dāng)前消費 1, 隊列剩余 1?
當(dāng)前消費 1, 隊列剩余 0?
未關(guān)閉但空了，等待生產(chǎn)
生產(chǎn)咯，來消費吧
生產(chǎn)咯，來消費吧
生產(chǎn)咯，來消費吧
當(dāng)前消費 1, 隊列剩余 2?
當(dāng)前消費 1, 隊列剩余 1?
當(dāng)前消費 1, 隊列剩余 0?
未關(guān)閉但空了，等待生產(chǎn)
生產(chǎn)咯，來消費吧
生產(chǎn)咯，來消費吧
生產(chǎn)咯，來消費吧
生產(chǎn)咯，來消費吧
當(dāng)前消費 1, 隊列剩余 1?
當(dāng)前消費 1, 隊列剩余 2?
當(dāng)前消費 1, 隊列剩余 1?
當(dāng)前消費 1, 隊列剩余 0?
未關(guān)閉但空了，等待生產(chǎn)
要關(guān)閉咯，大家快消費
關(guān)閉咯，也消費完咯
當(dāng)前消費 0, 隊列剩余 0

小結(jié)

1.sync.Map 并發(fā)安全的Map。

2.sync.Once 只執(zhí)行一次，適用于配置讀取、通道關(guān)閉。

3.sync.Cond 控制協(xié)調(diào)共享資源。