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 函數可以返回 *sync.Cond 類型的結果， *sync.Cond 我們主要操作其三個方法，分別是：
wait()：等待通知
Signal()：單播通知
Broadcast()：廣播通知
具體的函數說明如下：
// 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本質上是一些正在等待某個條件的線程的同步機制。
sync.Cond 主要實現一個條件變量，假如 goroutine A 執行前需要等待另外的goroutine B 的通知，那邊處于等待的goroutine A 會保存在一個通知列表，也就是說需要某種變量狀態的goroutine A 將會等待/Wait在那里，當某個時刻狀態改變時負責通知的goroutine B 通過對條件變量通知的方式（Broadcast，Signal）來通知處于等待條件變量的goroutine A, 這樣便可首先一種“消息通知”的同步機制。
以go的http處理為例，在Go的源碼中http模塊server部分源碼中所示，當需要處理一個新的連接的時候，若連接conn是實現自*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函數代碼通過使用條件變量的方式來處理新連接握手調用的同步問題：
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的使用：
我們嘗試實現一個讀寫同步的例子，需求是：我們有數個讀取器和數個寫入器，讀取器必須依賴寫入器對緩存區進行數據寫入后，才可從緩存區中對數據進行讀出。我們思考下，要實現類似的功能，除了使用channel，還能如何做？
寫入器每次完成寫入數據后，它都需要某種通知機制廣播給處于阻塞狀態的讀取器，告訴它們可以對數據進行訪問，這其實跟sync.Cond 的 廣播機制是不是很像？ 有了這個廣播機制，我們可以通過sync.Cond來實現這個例子了：
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 {
//讀取一個字節
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()
//寫入一個數據塊
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的時候有兩點移動要注意的：
- 一定要在調用cond.Wait方法前，鎖定與之關聯的讀寫鎖
- 一定不要忘記在cond.Wait后，若數據已經處理完畢，在返回前要對與之關聯的讀寫鎖進行解鎖。
如下面 Wait() 的源碼所示，Cond.Wait會自動釋放鎖等待信號的到來，當信號到來后，第一個獲取到信號的Wait將繼續往下執行并從新上鎖
func (c *Cond) Wait() {
c.checker.check()
t := runtime_notifyListAdd(c.notify)
c.L.Unlock()
runtime_notifyListWait(c.notify, t)
c.L.Lock()
}
如果不釋放鎖， 其它收到信號的gouroutine將阻塞無法繼續執行。
總結
以上就是這篇文章的全部內容了，希望本文的內容對大家的學習或者工作具有一定的參考學習價值，如果有疑問大家可以留言交流，謝謝大家對腳本之家的支持。
您可能感興趣的文章:- 詳解Go語言變量作用域
- go語言 全局變量和局部變量實例
- go語言的初始化順序,包,變量,init詳解
- go語言的工作空間和GOPATH環境變量介紹
- Go語言變量創建的五種方法
- Go語言基本的語法和內置數據類型初探
- Go語言變量與基礎數據類型詳情
