Go的標準庫中有一個類型叫條件變量：sync.Cond。這種類型與互斥鎖和讀寫鎖不同，它不是開箱即用的，它需要與互斥鎖組合使用：
// NewCond returns a new Cond with Locker l.
func NewCond(l Locker) *Cond {
return Cond{L: l}
}
// A Locker represents an object that can be locked and unlocked.
type Locker interface {
Lock()
Unlock()
}
通過使用 NewCond 函數(shù)可以返回 *sync.Cond 類型的結果， *sync.Cond 我們主要操作其三個方法，分別是：
wait()：等待通知
Signal()：單播通知
Broadcast()：廣播通知
具體的函數(shù)說明如下：
// Wait atomically unlocks c.L and suspends execution
// of the calling goroutine. After later resuming execution,
// Wait locks c.L before returning. Unlike in other systems,
// Wait cannot return unless awoken by Broadcast or Signal.
//
// Because c.L is not locked when Wait first resumes, the caller
// typically cannot assume that the condition is true when
// Wait returns. Instead, the caller should Wait in a loop:
//
// c.L.Lock()
// for !condition() {
// c.Wait()
// }
// ... make use of condition ...
// c.L.Unlock()
//
func (c *Cond) Wait() {
c.checker.check()
t := runtime_notifyListAdd(c.notify)
c.L.Unlock()
runtime_notifyListWait(c.notify, t)
c.L.Lock()
}
// Signal wakes one goroutine waiting on c, if there is any.
//
// It is allowed but not required for the caller to hold c.L
// during the call.
func (c *Cond) Signal() {
c.checker.check()
runtime_notifyListNotifyOne(c.notify)
}
// Broadcast wakes all goroutines waiting on c.
//
// It is allowed but not required for the caller to hold c.L
// during the call.
func (c *Cond) Broadcast() {
c.checker.check()
runtime_notifyListNotifyAll(c.notify)
}
條件變量sync.Cond本質(zhì)上是一些正在等待某個條件的線程的同步機制。
sync.Cond 主要實現(xiàn)一個條件變量，假如 goroutine A 執(zhí)行前需要等待另外的goroutine B 的通知，那邊處于等待的goroutine A 會保存在一個通知列表，也就是說需要某種變量狀態(tài)的goroutine A 將會等待/Wait在那里，當某個時刻狀態(tài)改變時負責通知的goroutine B 通過對條件變量通知的方式（Broadcast，Signal）來通知處于等待條件變量的goroutine A, 這樣便可首先一種“消息通知”的同步機制。
以go的http處理為例，在Go的源碼中http模塊server部分源碼中所示，當需要處理一個新的連接的時候，若連接conn是實現(xiàn)自*tls.Conn的情況下，會進行相關的客戶端與服務端的“握手”處理Handshake()， 入口代碼如下：
if tlsConn, ok := c.rwc.(*tls.Conn); ok {
if d := c.server.ReadTimeout; d != 0 {
c.rwc.SetReadDeadline(time.Now().Add(d))
}
if d := c.server.WriteTimeout; d != 0 {
c.rwc.SetWriteDeadline(time.Now().Add(d))
}
if err := tlsConn.Handshake(); err != nil {
c.server.logf("http: TLS handshake error from %s: %v", c.rwc.RemoteAddr(), err)
return
}
c.tlsState = new(tls.ConnectionState)
*c.tlsState = tlsConn.ConnectionState()
if proto := c.tlsState.NegotiatedProtocol; validNPN(proto) {
if fn := c.server.TLSNextProto[proto]; fn != nil {
h := initNPNRequest{tlsConn, serverHandler{c.server}}
fn(c.server, tlsConn, h)
}
return
}
}
其中的Handshake函數(shù)代碼通過使用條件變量的方式來處理新連接握手調(diào)用的同步問題：
func (c *Conn) Handshake() error {
c.handshakeMutex.Lock()
defer c.handshakeMutex.Unlock()
for {
if err := c.handshakeErr; err != nil {
return err
}
if c.handshakeComplete {
return nil
}
if c.handshakeCond == nil {
break
}
c.handshakeCond.Wait()
}
c.handshakeCond = sync.NewCond(c.handshakeMutex)
c.handshakeMutex.Unlock()
c.in.Lock()
defer c.in.Unlock()
c.handshakeMutex.Lock()
if c.handshakeErr != nil || c.handshakeComplete {
panic("handshake should not have been able to complete after handshakeCond was set")
}
if c.isClient {
c.handshakeErr = c.clientHandshake()
} else {
c.handshakeErr = c.serverHandshake()
}
if c.handshakeErr == nil {
c.handshakes++
} else {
c.flush()
}
if c.handshakeErr == nil !c.handshakeComplete {
panic("handshake should have had a result.")
}
c.handshakeCond.Broadcast()
c.handshakeCond = nil
return c.hand
我們也可以再通過一個例子熟悉sync.Cond的使用：
我們嘗試實現(xiàn)一個讀寫同步的例子，需求是：我們有數(shù)個讀取器和數(shù)個寫入器，讀取器必須依賴寫入器對緩存區(qū)進行數(shù)據(jù)寫入后，才可從緩存區(qū)中對數(shù)據(jù)進行讀出。我們思考下，要實現(xiàn)類似的功能，除了使用channel，還能如何做？
寫入器每次完成寫入數(shù)據(jù)后，它都需要某種通知機制廣播給處于阻塞狀態(tài)的讀取器，告訴它們可以對數(shù)據(jù)進行訪問，這其實跟sync.Cond 的 廣播機制是不是很像？ 有了這個廣播機制，我們可以通過sync.Cond來實現(xiàn)這個例子了：
package main
import (
"bytes"
"fmt"
"io"
"sync"
"time"
)
type MyDataBucket struct {
br *bytes.Buffer
gmutex *sync.RWMutex
rcond *sync.Cond //讀操作需要用到的條件變量
}
func NewDataBucket() *MyDataBucket {
buf := make([]byte, 0)
db := MyDataBucket{
br: bytes.NewBuffer(buf),
gmutex: new(sync.RWMutex),
}
db.rcond = sync.NewCond(db.gmutex.RLocker())
return db
}
func (db *MyDataBucket) Read(i int) {
db.gmutex.RLock()
defer db.gmutex.RUnlock()
var data []byte
var d byte
var err error
for {
//讀取一個字節(jié)
if d, err = db.br.ReadByte(); err != nil {
if err == io.EOF {
if string(data) != "" {
fmt.Printf("reader-%d: %s\n", i, data)
}
db.rcond.Wait()
data = data[:0]
continue
}
}
data = append(data, d)
}
}
func (db *MyDataBucket) Put(d []byte) (int, error) {
db.gmutex.Lock()
defer db.gmutex.Unlock()
//寫入一個數(shù)據(jù)塊
n, err := db.br.Write(d)
db.rcond.Broadcast()
return n, err
}
func main() {
db := NewDataBucket()
go db.Read(1)
go db.Read(2)
for i := 0; i 10; i++ {
go func(i int) {
d := fmt.Sprintf("data-%d", i)
db.Put([]byte(d))
}(i)
time.Sleep(100 * time.Millisecond)
}
}
當使用sync.Cond的時候有兩點移動要注意的：
- 一定要在調(diào)用cond.Wait方法前，鎖定與之關聯(lián)的讀寫鎖
- 一定不要忘記在cond.Wait后，若數(shù)據(jù)已經(jīng)處理完畢，在返回前要對與之關聯(lián)的讀寫鎖進行解鎖。
如下面 Wait() 的源碼所示，Cond.Wait會自動釋放鎖等待信號的到來，當信號到來后，第一個獲取到信號的Wait將繼續(xù)往下執(zhí)行并從新上鎖
func (c *Cond) Wait() {
c.checker.check()
t := runtime_notifyListAdd(c.notify)
c.L.Unlock()
runtime_notifyListWait(c.notify, t)
c.L.Lock()
}
如果不釋放鎖， 其它收到信號的gouroutine將阻塞無法繼續(xù)執(zhí)行。
總結
以上就是這篇文章的全部內(nèi)容了，希望本文的內(nèi)容對大家的學習或者工作具有一定的參考學習價值，如果有疑問大家可以留言交流，謝謝大家對腳本之家的支持。
您可能感興趣的文章:- 詳解Go語言變量作用域
- go語言 全局變量和局部變量實例
- go語言的初始化順序,包,變量,init詳解
- go語言的工作空間和GOPATH環(huán)境變量介紹
- Go語言變量創(chuàng)建的五種方法
- Go語言基本的語法和內(nèi)置數(shù)據(jù)類型初探
- Go語言變量與基礎數(shù)據(jù)類型詳情
