溫馨提示×

溫馨提示×

您好，登錄后才能下訂單哦！

密碼登錄×
登錄注冊×
其他方式登錄
點擊 登錄注冊 即表示同意《億速云用戶服務(wù)條款》

go語言同步教程之條件變量

發(fā)布時間：2020-10-25 15:23:06 來源：腳本之家 閱讀：158 作者：domac的菜園子 欄目：編程語言

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 類型的結(jié)果， *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)改變時負責(zé)通知的goroutine B 通過對條件變量通知的方式（Broadcast，Signal）來通知處于等待條件變量的goroutine A, 這樣便可首先一種“消息通知”的同步機制。

以go的http處理為例，在Go的源碼中http模塊server部分源碼中所示，當需要處理一個新的連接的時候，若連接conn是實現(xiàn)自*tls.Conn的情況下，會進行相關(guān)的客戶端與服務(wù)端的“握手”處理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方法前，鎖定與之關(guān)聯(lián)的讀寫鎖
  • 一定不要忘記在cond.Wait后，若數(shù)據(jù)已經(jīng)處理完畢，在返回前要對與之關(guān)聯(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í)行。

總結(jié)

以上就是這篇文章的全部內(nèi)容了，希望本文的內(nèi)容對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，如果有疑問大家可以留言交流，謝謝大家對億速云的支持。

向AI問一下細節(jié)

免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點不代表本網(wǎng)站立場，如果涉及侵權(quán)請聯(lián)系站長郵箱：is@yisu.com進行舉報，并提供相關(guān)證據(jù)，一經(jīng)查實，將立刻刪除涉嫌侵權(quán)內(nèi)容。

AI
助
手