併發控制--context篇

時間  2021-08-13
標籤 golang shell 數據結構 併發 ide 函數 線程 翻譯 code 協程 欄目 Go 简体版
原文   原文鏈接

併發控制--context篇

 

 

1. 前言

context翻譯成中文是」上下文」，即它能夠控制一組呈樹狀結構的goroutine，每一個goroutine擁有相同的上下文。golang

典型的使用場景以下圖所示：shell

圖中因爲goroutine派生出子goroutine，而子goroutine又繼續派生新的goroutine，這種狀況下使用WaitGroup就不太容易，由於子goroutine個數不容易肯定。而使用context就能夠很容易實現。數據結構


2 Context 實現原理

context實際上只定義了接口，凡是實現該接口的類均可稱爲是一種context，官方包中實現了幾個經常使用的context，分別可用於不一樣的場景。併發

2.1 接口定義

源碼包中src/context/context.go:Context 定義了該接口：ide

type Context interface {
    Deadline() (deadline time.Time, ok bool)

    Done() <-chan struct{}

    Err() error

    Value(key interface{}) interface{}
}

基礎的context接口只定義了4個方法，下面分別簡要說明一下：函數

Deadline()線程

該方法返回一個deadline和標識是否已設置deadline的bool值，若是沒有設置deadline，則ok == false，此時deadline爲一個初始值的time.Time值翻譯


Done()code

該方法返回一個channel，須要在select-case語句中使用，如case <-context.Done():。協程

當context關閉後，Done()返回一個被關閉的管道，關閉的管道仍然是可讀的，據此goroutine能夠收到關閉請求；
當context還未關閉時，Done()返回nil。


Err()

context包中定義了一個空的context， 名爲emptyCtx，用於context的根節點，空的context只是簡單的實現了Context，自己不包含任何值，僅用於其餘context的父節點。

emptyCtx類型定義以下代碼所示：

type emptyCtx int   // 定義一個int類型，經過實現context的四個方法來實現context接口

func (*emptyCtx) Deadline() (deadline time.Time, ok bool) {
    return
}

func (*emptyCtx) Done() <-chan struct{} {
    return nil
}

func (*emptyCtx) Err() error {
    return nil
}

func (*emptyCtx) Value(key interface{}) interface{} {
    return nil
}

context包中定義了一個公用的emptCtx全局變量，名爲background，可使用context.Background()獲取它，實現代碼以下所示：

var background = new(emptyCtx)
func Background() Context {
    return background
}

context包提供了4個方法建立不一樣類型的context，使用這四個方法時若是沒有父context，都須要傳入backgroud，即backgroud做爲其父節點：

  • WithCancle()
  • WithDeadline()
  • WithTimeout()
  • WithValue()

context包中實現Context接口的struct，除了emptyCtx外，還有cancelCtx、timerCtx和valueCtx三種，正是基於這三種context實例，實現了上述4種類型的context。


context包中各context類型之間的關係，以下圖所示：


struct cancelCtx、timerCtx、valueCtx都繼承於Context，下面分別介紹這三個struct。


2.1 cancelCtx

源碼包中src/context/context.go:cancelCtx 定義了該類型context：

type cancelCtx struct {
    Context

    mu       sync.Mutex            // protects following fields
    done     chan struct{}         // created lazily, closed by first cancel call
    children map[canceler]struct{} // set to nil by the first cancel call
    err      error                 // set to non-nil by the first cancel call
}

children中記錄了由此context派生的全部child，此context被cancel時會把其中的全部child都cancel掉。

cancelCtx與deadline和value無關，因此只須要實現Done()和Err()外露接口便可。


2.1.1 Done()接口實現

按照Context定義，Done()接口只須要返回一個channel便可，對於cancelCtx來講只須要返回成員變量done便可。

這裏直接看下源碼，很是簡單：

func (c *cancelCtx) Done() <-chan struct{} {
    c.mu.Lock()
    if c.done == nil {
        c.done = make(chan struct{})
    }
    d := c.done
    c.mu.Unlock()
    return d
}

因爲cancelCtx沒有指定初始化函數，因此cancelCtx.done可能還未分配，因此須要考慮初始化。
cancelCtx.done會在context被cancel時關閉，因此cancelCtx.done的值通常經歷以下三個階段：
nil –> chan struct{} –> closed chan。


2.1.2 Err()接口實現

按照Context定義，Err()只須要返回一個error告知context被關閉的緣由。對於cancelCtx來講只須要返回成員變量err便可。

源碼以下：

func (c *cancelCtx) Err() error {
    c.mu.Lock()
    err := c.err
    c.mu.Unlock()
    return err
}

2.1.3 cancel()接口實現

cancel()內部方法是理解cancelCtx的最關鍵的方法，其做用是關閉本身和其後代，其後代存儲在cancelCtx.children的map中，其中key值即後代對象，value值並無意義，這裏使用map只是爲了方便查詢而已。

cancel方法實現僞代碼以下所示：

func (c *cancelCtx) cancel(removeFromParent bool, err error) {
    c.mu.Lock()

    c.err = err                          //設置一個error，說明關閉緣由
    close(c.done)                     //將channel關閉，以此通知派生的context

    for child := range c.children {   //遍歷全部children，逐個調用cancel方法
        child.cancel(false, err)
    }
    c.children = nil
    c.mu.Unlock()

    if removeFromParent {            //正常狀況下，須要將本身從parent刪除
        removeChild(c.Context, c)
    }
}

實際上，WithCancel()返回的第二個用於cancel context的方法正是此cancel()。


2.1.4 WithCancel()方法實現

WithCancel()方法做了三件事：

1.初始化一個cancelCtx實例
2.將cancelCtx實例添加到其父節點的children中(若是父節點也能夠被cancel的話)
3.返回cancelCtx實例和cancel()方法

其實現源碼以下所示：

func WithCancel(parent Context) (ctx Context, cancel CancelFunc) {
    c := newCancelCtx(parent)
    propagateCancel(parent, &c)   //將自身添加到父節點
    return &c, func() { c.cancel(true, Canceled) }
}

這裏將自身添加到父節點的過程有必要簡單說明一下：

  1. 若是父節點也支持cancel，也就是說其父節點確定有children成員，那麼把新context添加到children裏便可；
  2. 若是父節點不支持cancel，就繼續向上查詢，直到找到一個支持cancel的節點，把新context添加到children裏；
  3. 若是全部的父節點均不支持cancel，則啓動一個協程等待父節點結束，而後再把當前context結束。

2.1.5 使用案例

一個典型的使用cancel context的例子以下所示：

package main

import (
    "fmt"
    "time"
    "context"
)

func HandelRequest(ctx context.Context) {
    go WriteRedis(ctx)                      // 子協程A建立子協程B
    go WriteDatabase(ctx)                   // 子協程A建立子協程C
    
    for {
        select {
        case <-ctx.Done():
            fmt.Println("HandelRequest Done.")
            return
        default:
            fmt.Println("HandelRequest running")
            time.Sleep(2 * time.Second)
        }
    }
}

func WriteRedis(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            fmt.Println("WriteRedis Done.")
            return
        default:
            fmt.Println("WriteRedis running")
            time.Sleep(2 * time.Second)
        }
    }
}

func WriteDatabase(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            fmt.Println("WriteDatabase Done.")
            return
        default:
            fmt.Println("WriteDatabase running")
            time.Sleep(2 * time.Second)
        }
    }
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    go HandelRequest(ctx)      // 主線程下建立子協程A

    time.Sleep(5 * time.Second)
    fmt.Println("It's time to stop all sub goroutines!")
    cancel()

    //Just for test whether sub goroutines exit or not
    time.Sleep(5 * time.Second)
}

上面代碼中協程HandelRequest()用於處理某個請求，其又會建立兩個協程：WriteRedis()、WriteDatabase()，main協程建立context，並把context在各子協程間傳遞，main協程在適當的時機能夠cancel掉全部子協程。

程序輸出以下所示：

WriteDatabase running
WriteRedis running
HandelRequest running
WriteRedis running
WriteDatabase running
HandelRequest running
HandelRequest running
WriteDatabase running
WriteRedis running
It's time to stop all sub goroutines!
WriteDatabase Done.
WriteRedis Done.
HandelRequest Done.

2.2 timerCtx

源碼包中src/context/context.go:timerCtx 定義了該類型context：

type timerCtx struct {
    cancelCtx
    timer *time.Timer // Under cancelCtx.mu.

    deadline time.Time
}

timerCtx在cancelCtx基礎上增長了deadline用於標示自動cancel的最終時間，而timer就是一個觸發自動cancel的定時器。

由此，衍生出WithDeadline()和WithTimeout()。實現上這兩種類型實現原理同樣，只不過使用語境不同：

  • deadline: 指定最後期限，好比context將2018.10.20 00:00:00之時自動結束
  • timeout: 指定最長存活時間，好比context將在30s後結束。

對於接口來講，timerCtx在cancelCtx基礎上還須要實現Deadline()和cancel()方法，其中cancel()方法是重寫的。


2.2.1 Deadline()接口實現

Deadline()方法僅僅是返回timerCtx.deadline而矣。而timerCtx.deadline是WithDeadline()或WithTimeout()方法設置的。


2.2.2 cancel()接口實現

cancel()方法基本繼承cancelCtx，只須要額外把timer關閉。

timerCtx被關閉後，timerCtx.cancelCtx.err將會存儲關閉緣由：

  • 若是deadline到來以前手動關閉，則關閉緣由與cancelCtx顯示一致；
  • 若是deadline到來時自動關閉，則緣由爲：」context deadline exceeded」

2.2.3 WithDeadline()方法實現

WithDeadline()方法實現步驟以下：

  • 初始化一個timerCtx實例
  • 將timerCtx實例添加到其父節點的children中(若是父節點也能夠被cancel的話)
  • 啓動定時器，定時器到期後會自動cancel本context
  • 返回timerCtx實例和cancel()方法

也就是說，timerCtx類型的context不只支持手動cancel，也會在定時器到來後自動cancel。


2.2.4 WithTimeout()方法實現

WithTimeout()實際調用了WithDeadline，兩者實現原理一致。

看代碼會很是清晰：

func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc) {
    return WithDeadline(parent, time.Now().Add(timeout))
}

2.2.5 典型使用案例

下面例子中使用WithTimeout()得到一個context並在其子協程中傳遞：

package main

import (
    "fmt"
    "time"
    "context"
)

func HandelRequest(ctx context.Context) {
    go WriteRedis(ctx)
    go WriteDatabase(ctx)
    for {
        select {
        case <-ctx.Done():
            fmt.Println("HandelRequest Done.")
            return
        default:
            fmt.Println("HandelRequest running")
            time.Sleep(2 * time.Second)
        }
    }
}

func WriteRedis(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            fmt.Println("WriteRedis Done.")
            return
        default:
            fmt.Println("WriteRedis running")
            time.Sleep(2 * time.Second)
        }
    }
}

func WriteDatabase(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            fmt.Println("WriteDatabase Done.")
            return
        default:
            fmt.Println("WriteDatabase running")
            time.Sleep(2 * time.Second)
        }
    }
}

func main() {
    ctx, _ := context.WithTimeout(context.Background(), 5 * time.Second) // 超時5S自動cancel
    go HandelRequest(ctx)

    time.Sleep(10 * time.Second)
}

主協程中建立一個10s超時的context，並將其傳遞給子協程，10s自動關閉context。程序輸出以下：

HandelRequest running
WriteRedis running
WriteDatabase running
HandelRequest running
WriteRedis running
WriteDatabase running
HandelRequest running
WriteRedis running
WriteDatabase running
HandelRequest Done.
WriteDatabase Done.
WriteRedis Done.

2.3 valueCtx

源碼包中src/context/context.go:valueCtx 定義了該類型context：

type valueCtx struct {
    Context
    key, val interface{}
}

valueCtx只是在Context基礎上增長了一個key-value對，用於在各級協程間傳遞一些數據。

因爲valueCtx既不須要cancel，也不須要deadline，那麼只須要實現Value()接口便可。


2.3.1 Value（）接口實現

由valueCtx數據結構定義可見，valueCtx.key和valueCtx.val分別表明其key和value值。 實現也很簡單：

func (c *valueCtx) Value(key interface{}) interface{} {
    if c.key == key {
        return c.val
    }
    return c.Context.Value(key)
}

這裏有個細節須要關注一下，即當前context查找不到key時，會向父節點查找，若是查詢不到則最終返回interface{}。也就是說，能夠經過子context查詢到父的value值。


2.3.2 WithValue（）方法實現

WithValue()實現也是很是的簡單, 僞代碼以下：

func WithValue(parent Context, key, val interface{}) Context {
    if key == nil {
        panic("nil key")
    }
    return &valueCtx{parent, key, val}
}

2.3.3 使用案例

下面示例程序展現valueCtx的用法：

package main

import (
    "fmt"
    "time"
    "context"
)

func HandelRequest(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            fmt.Println("HandelRequest Done.")
            return
        default:
            fmt.Println("HandelRequest running, parameter: ", ctx.Value("parameter"))
            time.Sleep(2 * time.Second)
        }
    }
}

func main() {
    ctx := context.WithValue(context.Background(), "parameter", "1")  // 傳遞值
    go HandelRequest(ctx)

    time.Sleep(10 * time.Second)
}

上例main()中經過WithValue()方法得到一個context，須要指定一個父context、key和value。而後通將該context傳遞給子協程HandelRequest，子協程能夠讀取到context的key-value。

注意：本例中子協程沒法自動結束，由於context是不支持cancle的，也就是說<-ctx.Done()永遠沒法返回。若是須要返回，須要在建立context時指定一個能夠cancel的context做爲父節點，使用父節點的cancel()在適當的時機結束整個context。


3. 總結

  • Context僅僅是一個接口定義，根據實現的不一樣，能夠衍生出不一樣的context類型；
  • cancelCtx實現了Context接口，經過WithCancel()建立cancelCtx實例；
  • timerCtx實現了Context接口，經過WithDeadline()和WithTimeout()建立timerCtx實例；
  • valueCtx實現了Context接口，經過WithValue()建立valueCtx實例；
  • 三種context實例可互爲父節點，從而能夠組合成不一樣的應用形式；
♥永遠年輕，永遠熱淚盈眶♥
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程