概述
我一直在找一種好的方法來解釋 go 語言的并發(fā)模型：
不要通過共享內(nèi)存來通信，相反，應(yīng)該通過通信來共享內(nèi)存
但是沒有發(fā)現(xiàn)一個好的解釋來滿足我下面的需求：
1.通過一個例子來說明最初的問題
2.提供一個共享內(nèi)存的解決方案
3.提供一個通過通信的解決方案
這篇文章我就從這三個方面來做出解釋。
讀過這篇文章后你應(yīng)該會了解通過通信來共享內(nèi)存的模型，以及它和通過共享內(nèi)存來通信的區(qū)別，你還將看到如何分別通過這兩種模型來解決訪問和修改共享資源的問題。
前提
設(shè)想一下我們要訪問一個銀行賬號：
type Account interface {
Withdraw(uint)
Deposit(uint)
Balance() int
}
type Bank struct {
account Account
}
func NewBank(account Account) *Bank {
return &Bank{account: account}
}
func (bank *Bank) Withdraw(amount uint, actor_name string) {
fmt.Println("[-]", amount, actor_name)
bank.account.Withdraw(amount)
}
func (bank *Bank) Deposit(amount uint, actor_name string) {
fmt.Println("[+]", amount, actor_name)
bank.account.Deposit(amount)
}
func (bank *Bank) Balance() int {
return bank.account.Balance()
}
因為 Account 是一個接口，所以我們提供一個簡單的實現(xiàn)：
type SimpleAccount struct{
balance int
}
func NewSimpleAccount(balance int) *SimpleAccount {
return &SimpleAccount{balance: balance}
}
func (acc *SimpleAccount) Deposit(amount uint) {
acc.setBalance(acc.balance + int(amount))
}
func (acc *SimpleAccount) Withdraw(amount uint) {
if acc.balance >= int(mount) {
acc.setBalance(acc.balance - int(amount))
} else {
panic("杰克窮死")
}
}
func (acc *SimpleAccount) Balance() int {
return acc.balance
}
func (acc *SimpleAccount) setBalance(balance int) {
acc.add_some_latency() //增加一個延時函數(shù)，方便演示
acc.balance = balance
}
func (acc *SimpleAccount) add_some_latency() {
<-time.After(time.Duration(rand.Intn(100)) * time.Millisecond)
}
你可能注意到了 balance 沒有被直接修改，而是被放到了 setBalance 方法里進(jìn)行修改。這樣設(shè)計是為了更好的描述問題。稍后我會做出解釋。
把上面所有部分弄好以后我們就可以像下面這樣使用它啦：
func main() {
balance := 80
b := NewBank(bank.NewSimpleAccount(balance))
fmt.Println("初始化余額", b.Balance())
b.Withdraw(30, "馬伊琍")
fmt.Println("-----------------")
fmt.Println("剩余余額", b.Balance())
}
運(yùn)行上面的代碼會輸出：
初始化余額 80
[-] 30 馬伊琍
-----------------
剩余余額 50
沒錯！
不錯在現(xiàn)實生活中，一個銀行賬號可以有很多個附屬卡，不同的附屬卡都可以對同一個賬號進(jìn)行存取錢，所以我們來修改一下代碼：
func main() {
balance := 80
b := NewBank(bank.NewSimpleAccount(balance))
fmt.Println("初始化余額", b.Balance())
done := make(chan bool)
go func() { b.Withdraw(30, "馬伊琍"); done <- true }()
go func() { b.Withdraw(10, "姚笛"); done <- true }()
//等待 goroutine 執(zhí)行完成
<-done
<-done
fmt.Println("-----------------")
fmt.Println("剩余余額", b.Balance())
}
這兒兩個附屬卡并發(fā)的從賬號里取錢，來看看輸出結(jié)果：
初始化余額 80
[-] 30 馬伊琍
[-] 10 姚笛
-----------------
剩余余額 70
這下把文章高興壞了:)
結(jié)果當(dāng)然是錯誤的，剩余余額應(yīng)該是40而不是70，那么讓我們看看到底哪兒出問題了。
問題
當(dāng)并發(fā)訪問共享資源時，無效狀態(tài)有很大可能會發(fā)生。
在我們的例子中，當(dāng)兩個附屬卡同一時刻從同一個賬號取錢后，我們最后得到銀行賬號(即共享資源)錯誤的剩余余額(即無效狀態(tài))。
我們來看一下執(zhí)行時候的情況：
處理情況
--------------
_馬伊琍_|_姚笛_
1. 獲取余額 80 | 80
2. 取錢 -30 | -10
3. 當(dāng)前剩余 50 | 70
... | ...
4. 設(shè)置余額 50 ? 70 //該先設(shè)置哪個好呢？
5. 后設(shè)置的生效了
--------------
6. 剩余余額 70
上面 ... 的地方描述了我們 add_some_latency 實現(xiàn)的延時狀況，現(xiàn)實世界經(jīng)常發(fā)生延遲情況。所以最后的剩余余額就由最后設(shè)置余額的那個附屬卡決定。
解決辦法
我們通過兩種方法來解決這個問題：
1.共享內(nèi)存的解決方案
2.通過通信的解決方案
所有的解決方案都是簡單的封裝了一下 SimpleAccount 來實現(xiàn)保護(hù)機(jī)制。
共享內(nèi)存的解決方案
又叫 “通過共享內(nèi)存來通信”。
這種方案暗示了使用鎖機(jī)制來預(yù)防同時訪問和修改共享資源。鎖告訴其它處理程序這個資源已經(jīng)被一個處理程序占用了，因此別的處理程序需要排隊直到當(dāng)前處理程序處理完畢。
讓我們來看看 LockingAccount 是怎么實現(xiàn)的：
type LockingAccount struct {
lock sync.Mutex
account *SimpleAccount
}
//封裝一下 SimpleAccount
func NewLockingAccount(balance int) *LockingAccount {
return &LockingAccount{account: NewSimpleAccount(balance)}
}
func (acc *LockingAccount) Deposit(amount uint) {
acc.lock.Lock()
defer acc.lock.Unlock()
acc.account.Deposit(amount)
}
func (acc *LockingAccount) Withdraw(amount uint) {
acc.lock.Lock()
defer acc.lock.Unlock()
acc.account.Withdraw(amount)
}
func (acc *LockingAccount) Balance() int {
acc.lock.Lock()
defer acc.lock.Unlock()
return acc.account.Balance()
}
直接明了！注意 lock sync.Lock，lock.Lock()，lock.Unlock()。
這樣每次一個附屬卡訪問銀行賬號(即共享資源)，這個附屬卡會自動獲得鎖直到最后操作完畢。
我們的 LockingAccount 像下面這樣使用：
func main() {
balance := 80
b := NewBank(bank.NewLockingAccount(balance))
fmt.Println("初始化余額", b.Balance())
done := make(chan bool)
go func() { b.Withdraw(30, "馬伊琍"); done <- true }()
go func() { b.Withdraw(10, "姚笛"); done <- true }()
//等待 goroutine 執(zhí)行完成
<-done
<-done
fmt.Println("-----------------")
fmt.Println("剩余余額", b.Balance())
}
輸出的結(jié)果是：
初始化余額 80
[-] 30 馬伊琍
[-] 10 姚笛
-----------------
剩余余額 40
現(xiàn)在結(jié)果正確了！
在這個例子中第一個處理程序加鎖后獨(dú)享共享資源，其它處理程序只能等待它執(zhí)行完成。
我們接著看一下執(zhí)行時的情況，假設(shè)馬伊琍先拿到了鎖：
處理過程
________________
_馬伊琍_|__姚笛__
加鎖 ><
得到余額 80 |
取錢 -30 |
當(dāng)前余額 50 |
... |
設(shè)置余額 50 |
解除鎖 <>
|
當(dāng)前余額 50
|
加鎖 ><
得到余額 | 50
取錢 | -10
當(dāng)前余額 | 40
| ...
設(shè)置余額 | 40
解除鎖 <>
________________
剩余余額 40
現(xiàn)在我們的處理程序在訪問共享資源時相繼的產(chǎn)生了正確的結(jié)果。
通過通信的解決方案
又叫 “通過通信來共享內(nèi)存”。
現(xiàn)在賬號被命名為 ConcurrentAccount，像下面這樣來實現(xiàn)：
type ConcurrentAccount struct {
account *SimpleAccount
deposits chan uint
withdrawals chan uint
balances chan chan int
}
func NewConcurrentAccount(amount int) *ConcurrentAccount{
acc := &ConcurrentAccount{
account : &SimpleAccount{balance: amount},
deposits: make(chan uint),
withdrawals: make(chan uint),
balances: make(chan chan int),
}
acc.listen()
return acc
}
func (acc *ConcurrentAccount) Balance() int {
ch := make(chan int)
acc.balances <- ch
return <-ch
}
func (acc *ConcurrentAccount) Deposit(amount uint) {
acc.deposits <- amount
}
func (acc *ConcurrentAccount) Withdraw(amount uint) {
acc.withdrawals <- amount
}
func (acc *ConcurrentAccount) listen() {
go func() {
for {
select {
case amnt := <-acc.deposits:
acc.account.Deposit(amnt)
case amnt := <-acc.withdrawals:
acc.account.Withdraw(amnt)
case ch := <-acc.balances:
ch <- acc.account.Balance()
}
}
}()
}
ConcurrentAccount 同樣封裝了 SimpleAccount ，然后增加了通信通道
調(diào)用代碼和加鎖版本的一樣，這里就不寫了，唯一不一樣的就是初始化銀行賬號的時候：
b := NewBank(bank.NewConcurrentAccount(balance))
運(yùn)行產(chǎn)生的結(jié)果和加鎖版本一樣：
初始化余額 80
[-] 30 馬伊琍
[-] 10 姚笛
-----------------
剩余余額 40
讓我們來深入了解一下細(xì)節(jié)。
通過通信來共享內(nèi)存是如何工作的
一些基本注意點(diǎn)：
共享資源被封裝在一個控制流程中。
結(jié)果就是資源成為了非共享狀態(tài)。沒有處理程序能夠直接訪問或者修改資源。你可以看到訪問和修改資源的方法實際上并沒有執(zhí)行任何改變。
func (acc *ConcurrentAccount) Balance() int {
ch := make(chan int)
acc.balances <- ch
balance := <-ch
return balance
}
func (acc *ConcurrentAccount) Deposit(amount uint) {
acc.deposits <- amount
}
func (acc *ConcurrentAccount) Withdraw(amount uint) {
acc.withdrawals <- amount
}
訪問和修改是通過消息和控制流程通信。
在控制流程中任何訪問和修改的動作都是相繼發(fā)生的。
當(dāng)控制流程接收到訪問或者修改的請求后會立即執(zhí)行相關(guān)動作。讓我們仔細(xì)看看這個流程：
func (acc *ConcurrentAccount) listen() {
// 執(zhí)行控制流程
go func() {
for {
select {
case amnt := <-acc.deposits:
acc.account.Deposit(amnt)
case amnt := <-acc.withdrawals:
acc.account.Withdraw(amnt)
case ch := <-acc.balances:
ch <- acc.account.Balance()
}
}
}()
}
select 不斷地從各個通道中取出消息，每個通道都跟它們所要執(zhí)行的操作相一致。
重要的一點(diǎn)是：在 select 聲明內(nèi)部的一切都是相繼執(zhí)行的(在同一個處理程序中排隊執(zhí)行)。一次只有一個事件(在通道中接受或者發(fā)送)發(fā)生，這樣就保證了同步訪問共享資源。
領(lǐng)會這個有一點(diǎn)繞。
讓我們用例子來看看 Balance() 的執(zhí)行情況：
一張附屬卡的流程 | 控制流程
----------------------------------------------
1. b.Balance() |
2. ch -> [acc.balances]-> ch
3. <-ch | balance = acc.account.Balance()
4. return balance <-[ch]<- balance
5 |
這兩個流程都干了點(diǎn)什么呢？
附屬卡的流程
1.調(diào)用 b.Balance()
2.新建通道 ch，將 ch 通道塞入通道 acc.balances 中與控制流程通信，這樣控制流程也可以通過 ch 來返回余額
3.等待 <-ch 來取得要接受的余額
4.接受余額
5.繼續(xù)
控制流程
1.空閑或者處理
2.通過 acc.balances 通道里面的 ch 通道來接受余額請求
3.取得真正的余額值
4.將余額值發(fā)送到 ch 通道
5.準(zhǔn)備處理下一個請求
控制流程每次只處理一個 事件。這也就是為什么除了描述出來的這些以外，第2-4步?jīng)]有別的操作執(zhí)行。
總結(jié)
這篇博客描述了問題以及問題的解決辦法，但那時沒有深入去探究不同解決辦法的優(yōu)缺點(diǎn)。
其實這篇文章的例子更適合用 mutex，因為這樣代碼更加清晰。
最后，請毫無顧忌的指出我的錯誤！
