golang における Context の使用シナリオはどのようなものかご存知ですか?

次のコラム golang チュートリアル では、golang での Context の使用シナリオを紹介します。

golang でのコンテキスト使用シナリオ

context は、Go1.7 以降の標準ライブラリに含まれています。その主な用途を一言で言えば、 goroutine のライフサイクルを制御することです。計算タスクが goroutine によって引き継がれ、何らかの理由 (タイムアウトまたは強制終了) で goroutine の計算タスクを中止したい場合、この Context が使用されます。

この記事では、主に golang でのコンテキストの使用シナリオについて説明します。

シナリオ 1: RPC 呼び出し

メインの goroutine RPC には 4 つあり、 RPC2/3/4 は並行してリクエストされますが、ここでは、RPC2 リクエストが失敗した後、直接エラーが返され、RPC3/4 が計算の継続を停止することを期待しています。このときContextを利用します。

これの具体的な実装は次のとおりです。

package main

import (
	"context"
	"sync"
	"github.com/pkg/errors"
)

func Rpc(ctx context.Context, url string) error {
	result := make(chan int)
	err := make(chan error)

	go func() {
		// 进行RPC调用，并且返回是否成功，成功通过result传递成功信息，错误通过error传递错误信息
		isSuccess := true
		if isSuccess {
			result <- 1
		} else {
			err <- errors.New("some error happen")
		}
	}()

	select {
		case <- ctx.Done():
			// 其他RPC调用调用失败
			return ctx.Err()
		case e := <- err:
			// 本RPC调用失败，返回错误信息
			return e
		case <- result:
			// 本RPC调用成功，不返回错误信息
			return nil
	}
}


func main() {
	ctx, cancel := context.WithCancel(context.Background())

	// RPC1调用
	err := Rpc(ctx, "http://rpc_1_url")
	if err != nil {
		return
	}

	wg := sync.WaitGroup{}

	// RPC2调用
	wg.Add(1)
	go func(){
		defer wg.Done()
		err := Rpc(ctx, "http://rpc_2_url")
		if err != nil {
			cancel()
		}
	}()

	// RPC3调用
	wg.Add(1)
	go func(){
		defer wg.Done()
		err := Rpc(ctx, "http://rpc_3_url")
		if err != nil {
			cancel()
		}
	}()

	// RPC4调用
	wg.Add(1)
	go func(){
		defer wg.Done()
		err := Rpc(ctx, "http://rpc_4_url")
		if err != nil {
			cancel()
		}
	}()

	wg.Wait()
}

もちろん、ここでは waitGroup を使用して、すべての RPC 呼び出しが完了するまで main 関数が終了しないようにしています。

Rpc 関数の最初のパラメータは CancelContext です。この Context はマイクです。CancelContext を作成すると、リスナー (ctx) とマイク (キャンセル関数) が返されます。)すべてのゴルーチンはこのリスナー (ctx) を保持しており、メインのゴルーチンがすべてのゴルーチンに終了することを通知したい場合、cancel 関数を通じてすべてのゴルーチンに終了情報を通知します。もちろん、すべてのゴルーチンには、リスナー終了信号 (ctx->Done()) を処理するための組み込みロジックが必要です。 Rpc 関数の内部を調べ、select を使用して ctx のどれが完了し、現在の rpc 呼び出しが最初に終了するかを決定できます。

この waitGroup と RPC 呼び出しの 1 つは、すべての RPC ロジックに通知します。実際、パッケージがすでにそれを行っています。エラーグループ。この errorGroup パッケージの具体的な使用方法については、このパッケージのテスト例を参照してください。

ここでの cancel() が複数回呼び出されるのではないかと心配する人もいるかもしれませんが、コンテキスト パッケージ内の cancel 呼び出しは冪等です。安心して何度でも呼び出せます。

ここを見てみるのもいいかもしれません。ここでの Rpc 関数は、この例では実際には「ブロッキング」リクエストです。このリクエストが http.Get または http.Post を使用して実装されている場合、実際には、 Rpc 関数は終了しましたが、内部の実際の http.Get は終了していません。したがって、ここでの関数は http.Do などの「ノンブロッキング」であることが最適であり、その後、何らかの方法で中断される可能性があることを理解する必要があります。たとえば、この記事のこの例のように、Context を使用して http.Request をキャンセルします。

func httpRequest(
  ctx context.Context,
  client *http.Client,
  req *http.Request,
  respChan chan []byte,
  errChan chan error
) {
  req = req.WithContext(ctx)
  tr := &http.Transport{}
  client.Transport = tr
  go func() {
    resp, err := client.Do(req)
    if err != nil {
      errChan <- err
    }
    if resp != nil {
      defer resp.Body.Close()
      respData, err := ioutil.ReadAll(resp.Body)
      if err != nil {
        errChan <- err
      }
      respChan <- respData
    } else {
      errChan <- errors.New("HTTP request failed")
    }
  }()
  for {
    select {
    case <-ctx.Done():
      tr.CancelRequest(req)
      errChan <- errors.New("HTTP request cancelled")
      return
    case <-errChan:
      tr.CancelRequest(req)
      return
    }
  }
}

http.Client.Do を使用し、ctx.Done を受信すると、transport.CancelRequest を呼び出して終了します。
net/dail/DialContext を参照することもできます。
つまり、実装するパッケージを「停止可能/制御可能」にしたい場合は、実装する関数内で Context を受信できるようにするのが最善です。関数とハンドル Context.Done。

シナリオ 2: PipeLine

パイプライン モデルは組立ライン モデルです。組立ラインの数人の作業者が n 製品を持ち、それらを 1 つずつ組み立てます。実際、パイプライン モデルの実装はコンテキストとは何の関係もありませんが、chan を使用してコンテキストなしでパイプライン モデルを実装することもできます。ただし、パイプライン全体を制御するには、Context を使用する必要があります。この記事の Go のパイプライン パターンの例は、非常に良い例です。このコードの大まかな説明は次のとおりです。

runSimplePipeline には 3 つのパイプライン ワーカーがあり、lineListSource はパラメータを 1 つずつ分割して送信し、lineParser は文字列を int64 に処理し、sink は特定の値に基づいてデータが利用可能かどうかを判断します。 。すべての戻り値には基本的に 2 つのチャネルがあり、1 つはデータの受け渡し用、もう 1 つはエラーの受け渡し用です。 (<-chan 文字列、<-chan エラー) input には基本的に 2 つの値があり、1 つは音声伝送制御に使用される Context、もう 1 つは (<-chan 内の) input product です。

これら 3 つのワーカーの特定の関数では、ケース <-ctx.Done() を処理するために switch が使用されていることがわかります。生産ラインにおける指令制御です。

func lineParser(ctx context.Context, base int, in <-chan string) (
	<-chan int64, <-chan error, error) {
	...
	go func() {
		defer close(out)
		defer close(errc)

		for line := range in {

			n, err := strconv.ParseInt(line, base, 64)
			if err != nil {
				errc <- err
				return
			}

			select {
			case out <- n:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out, errc, nil
}

シナリオ 3: タイムアウト リクエスト

RPC リクエストを送信するとき、このリクエストにタイムアウト制限を課すことがよくあります。 RPC リクエストが 10 秒を超えると、自動的に切断されます。もちろん、CancelContext を使用してこの機能を実現することもできます (新しいゴルーチンを開始し、このゴルーチンはキャンセル関数を保持し、時間が経過するとキャンセル関数が呼び出されます)。

この要件は非常に一般的であるため、コンテキスト パッケージもこの要件 timerCtx を実装します。特定のインスタンス化メソッドは、WithDeadline および WithTimeout です。

timerCtx の特定のロジックは、time.AfterFunc を通じて ctx.cancel を呼び出すことです。

公式例:

package main

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

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
    defer cancel()

    select {
    case <-time.After(1 * time.Second):
        fmt.Println("overslept")
    case <-ctx.Done():
        fmt.Println(ctx.Err()) // prints "context deadline exceeded"
    }
}

http クライアントにタイムアウトを追加する一般的な方法です

uri := "https://httpbin.org/delay/3"
req, err := http.NewRequest("GET", uri, nil)
if err != nil {
	log.Fatalf("http.NewRequest() failed with '%s'\n", err)
}

ctx, _ := context.WithTimeout(context.Background(), time.Millisecond*100)
req = req.WithContext(ctx)

resp, err := http.DefaultClient.Do(req)
if err != nil {
	log.Fatalf("http.DefaultClient.Do() failed with:\n'%s'\n", err)
}
defer resp.Body.Close()

http サーバーでタイムアウトを設定するにはどうすればよいですか?

package main

import (
	"net/http"
	"time"
)

func test(w http.ResponseWriter, r *http.Request) {
	time.Sleep(20 * time.Second)
	w.Write([]byte("test"))
}


func main() {
	http.HandleFunc("/", test)
	timeoutHandler := http.TimeoutHandler(http.DefaultServeMux, 5 * time.Second, "timeout")
	http.ListenAndServe(":8080", timeoutHandler)
}

我们看看TimeoutHandler的内部，本质上也是通过context.WithTimeout来做处理。

func (h *timeoutHandler) ServeHTTP(w ResponseWriter, r *Request) {
  ...
		ctx, cancelCtx = context.WithTimeout(r.Context(), h.dt)
		defer cancelCtx()
	...
	go func() {
    ...
		h.handler.ServeHTTP(tw, r)
	}()
	select {
    ...
	case <-ctx.Done():
		...
	}
}

场景四：HTTP服务器的request互相传递数据

context还提供了valueCtx的数据结构。

这个valueCtx最经常使用的场景就是在一个http服务器中，在request中传递一个特定值，比如有一个中间件，做cookie验证，然后把验证后的用户名存放在request中。

我们可以看到，官方的request里面是包含了Context的，并且提供了WithContext的方法进行context的替换。

package main

import (
	"net/http"
	"context"
)

type FooKey string

var UserName = FooKey("user-name")
var UserId = FooKey("user-id")

func foo(next http.HandlerFunc) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		ctx := context.WithValue(r.Context(), UserId, "1")
		ctx2 := context.WithValue(ctx, UserName, "yejianfeng")
		next(w, r.WithContext(ctx2))
	}
}

func GetUserName(context context.Context) string {
	if ret, ok := context.Value(UserName).(string); ok {
		return ret
	}
	return ""
}

func GetUserId(context context.Context) string {
	if ret, ok := context.Value(UserId).(string); ok {
		return ret
	}
	return ""
}

func test(w http.ResponseWriter, r *http.Request) {
	w.Write([]byte("welcome: "))
	w.Write([]byte(GetUserId(r.Context())))
	w.Write([]byte(" "))
	w.Write([]byte(GetUserName(r.Context())))
}

func main() {
	http.Handle("/", foo(test))
	http.ListenAndServe(":8080", nil)
}

在使用ValueCtx的时候需要注意一点，这里的key不应该设置成为普通的String或者Int类型，为了防止不同的中间件对这个key的覆盖。最好的情况是每个中间件使用一个自定义的key类型，比如这里的FooKey，而且获取Value的逻辑尽量也抽取出来作为一个函数，放在这个middleware的同包中。这样，就会有效避免不同包设置相同的key的冲突问题了。

以上がgolang における Context の使用シナリオはどのようなものかご存知ですか?の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。