Go言語 errgroupライブラリ：強力な並行性制御ツール

errgroupは、複数のgoroutineを並行して実行し、エラーを処理するために使用される、公式Goライブラリxのユーティリティです。sync.WaitGroupに基づいてerrgroup.Groupを実装し、並行プログラミングのためのより強力な機能を提供します。

errgroupの利点

sync.WaitGroupと比較して、errgroup.Groupには次の利点があります。

1. エラー処理：sync.WaitGroupは、goroutineの完了を待機するだけで、戻り値やエラーは処理しません。一方、errgroup.Groupは、戻り値を直接処理することはできませんが、goroutineがエラーを検出すると、他の実行中のgoroutineを直ちにキャンセルし、Waitメソッドで最初のnilでないエラーを返します。
2. コンテキストキャンセル：errgroupはcontext.Contextと組み合わせて使用​​できます。goroutineがエラーを検出すると、他のgoroutineを自動的にキャンセルし、リソースを効果的に制御し、不要な作業を回避します。
3. 並行プログラミングの簡素化：errgroupを使用すると、エラー処理のボイラープレートコードを削減できます。開発者は、エラー状態と同期ロジックを手動で管理する必要がないため、並行プログラミングがよりシンプルになり、保守しやすくなります。
4. 並行数の制限：errgroupは、過負荷を避けるために、同時実行goroutineの数を制限するためのインターフェースを提供します。これは、sync.WaitGroupにはない機能です。

sync.WaitGroupの使用例

errgroup.Groupを紹介する前に、まずsync.WaitGroupの使用法を確認しましょう。

package main

import (
    "fmt"
    "net/http"
    "sync"
)

func main() {
    var urls = []string{
        "http://www.golang.org/",
        "http://www.google.com/",
        "http://www.somestupidname.com/", 
    }
    var err error

    var wg sync.WaitGroup 

    for _, url := range urls {
        wg.Add(1) 

        go func() {
            defer wg.Done() 

            resp, e := http.Get(url)
            if e != nil { 
                err = e
                return
            }
            defer resp.Body.Close()
            fmt.Printf("fetch url %s status %s\n", url, resp.Status)
        }()
    }

    wg.Wait()
    if err != nil { 
        fmt.Printf("Error: %s\n", err)
    }
}

実行結果：

$ go run waitgroup/main.go
fetch url http://www.google.com/ status 200 OK
fetch url http://www.golang.org/ status 200 OK
Error: Get "http://www.somestupidname.com/": dial tcp: lookup www.somestupidname.com: no such host

sync.WaitGroupの典型的なイディオム：

var wg sync.WaitGroup

for ... {
    wg.Add(1)

    go func() {
        defer wg.Done()
        // do something
    }()
}

wg.Wait()

errgroup.Groupの使用例

基本的な使い方

errgroup.Groupの使用パターンはsync.WaitGroupと同様です。

package main

import (
    "fmt"
    "net/http"
    "golang.org/x/sync/errgroup"
)

func main() {
    var urls = []string{
        "http://www.golang.org/",
        "http://www.google.com/",
        "http://www.somestupidname.com/", 
    }

    var g errgroup.Group 

    for _, url := range urls {
        g.Go(func() error {
            resp, err := http.Get(url)
            if err != nil {
                return err 
            }
            defer resp.Body.Close()
            fmt.Printf("fetch url %s status %s\n", url, resp.Status)
            return nil 
        })
    }

    if err := g.Wait(); err != nil {
        fmt.Printf("Error: %s\n", err)
    }
}

実行結果：

$ go run examples/main.go
fetch url http://www.google.com/ status 200 OK
fetch url http://www.golang.org/ status 200 OK
Error: Get "http://www.somestupidname.com/": dial tcp: lookup www.somestupidname.com: no such host

コンテキストキャンセル

errgroupは、キャンセル機能を追加するためにerrgroup.WithContextを提供します。

package main

import (
    "context"
    "fmt"
    "net/http"
    "sync"
    "golang.org/x/sync/errgroup"
)

func main() {
    var urls = []string{
        "http://www.golang.org/",
        "http://www.google.com/",
        "http://www.somestupidname.com/", 
    }

    g, ctx := errgroup.WithContext(context.Background())

    var result sync.Map

    for _, url := range urls {
        g.Go(func() error {
            req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
            if err != nil {
                return err 
            }

            resp, err := http.DefaultClient.Do(req)
            if err != nil {
                return err 
            }
            defer resp.Body.Close()

            result.Store(url, resp.Status)
            return nil 
        })
    }

    if err := g.Wait(); err != nil {
        fmt.Println("Error: ", err)
    }

    result.Range(func(key, value any) bool {
        fmt.Printf("fetch url %s status %s\n", key, value)
        return true
    })
}

実行結果：

$ go run examples/withcontext/main.go
Error:  Get "http://www.somestupidname.com/": dial tcp: lookup www.somestupidname.com: no such host
fetch url http://www.google.com/ status 200 OK

http://www.somestupidname.com/へのリクエストがエラーを報告したため、プログラムはhttp://www.golang.org/へのリクエストをキャンセルしました。

並行数の制限

errgroupは、同時実行goroutineの数を制限するためにerrgroup.SetLimitを提供します。

package main

import (
    "fmt"
    "time"
    "golang.org/x/sync/errgroup"
)

func main() {
    var g errgroup.Group
    g.SetLimit(3)

    for i := 1; i <= 10; i++ {
        g.Go(func() error {
            fmt.Printf("Goroutine %d is starting\n", i)
            time.Sleep(2 * time.Second) 
            fmt.Printf("Goroutine %d is done\n", i)
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        fmt.Printf("Encountered an error: %v\n", err)
    }

    fmt.Println("All goroutines complete.")
}

実行結果：

$  go run examples/setlimit/main.go
Goroutine 3 is starting
Goroutine 1 is starting
Goroutine 2 is starting
Goroutine 2 is done
Goroutine 1 is done
Goroutine 5 is starting
Goroutine 3 is done
Goroutine 6 is starting
Goroutine 4 is starting
Goroutine 6 is done
Goroutine 5 is done
Goroutine 8 is starting
Goroutine 4 is done
Goroutine 7 is starting
Goroutine 9 is starting
Goroutine 9 is done
Goroutine 8 is done
Goroutine 10 is starting
Goroutine 7 is done
Goroutine 10 is done
All goroutines complete.

開始を試みる

errgroupは、タスクの開始を試みるためにerrgroup.TryGoを提供します。これは、errgroup.SetLimitと組み合わせて使用​​する必要があります。

package main

import (
    "fmt"
    "time"
    "golang.org/x/sync/errgroup"
)

func main() {
    var g errgroup.Group
    g.SetLimit(3)

    for i := 1; i <= 10; i++ {
        if g.TryGo(func() error {
            fmt.Printf("Goroutine %d is starting\n", i)
            time.Sleep(2 * time.Second) 
            fmt.Printf("Goroutine %d is done\n", i)
            return nil
        }) {
            fmt.Printf("Goroutine %d started successfully\n", i)
        } else {
            fmt.Printf("Goroutine %d could not start (limit reached)\n", i)
        }
    }

    if err := g.Wait(); err != nil {
        fmt.Printf("Encountered an error: %v\n", err)
    }

    fmt.Println("All goroutines complete.")
}

実行結果：

$ go run examples/trygo/main.go
Goroutine 1 started successfully
Goroutine 1 is starting
Goroutine 2 is starting
Goroutine 2 started successfully
Goroutine 3 started successfully
Goroutine 4 could not start (limit reached)
Goroutine 5 could not start (limit reached)
Goroutine 6 could not start (limit reached)
Goroutine 7 could not start (limit reached)
Goroutine 8 could not start (limit reached)
Goroutine 9 could not start (limit reached)
Goroutine 10 could not start (limit reached)
Goroutine 3 is starting
Goroutine 2 is done
Goroutine 3 is done
Goroutine 1 is done
All goroutines complete.

ソースコードの解釈

errgroupのソースコードは、主に3つのファイルで構成されています。

• メインロジックコード
• Go 1.24以降のバージョンのwithCancelCauseの実装
• Go 1.24より前のバージョンのwithCancelCauseの実装

コア構造

type token struct{}

type Group struct {
    cancel func(error)
    wg sync.WaitGroup
    sem chan token
    errOnce sync.Once
    err     error
}

• token：並行数を制御するためにシグナルを渡すために使用される空の構造体。
• Group：
  • cancel：コンテキストがキャンセルされたときに呼び出される関数。
  • wg：内部で使用されるsync.WaitGroup。
  • sem：同時実行コルーチンの数を制御するシグナルチャネル。
  • errOnce：エラーが一度だけ処理されるようにします。
  • err：最初のエラーを記録します。

主なメソッド

• SetLimit：並行数を制限します。

func (g *Group) SetLimit(n int) {
    if n < 0 {
        g.sem = nil
        return
    }
    if len(g.sem) != 0 {
        panic(fmt.Errorf("errgroup: modify limit while %v goroutines in the group are still active", len(g.sem)))
    }
    g.sem = make(chan token, n)
}

• Go：タスクを実行するために新しいコルーチンを開始します。

func (g *Group) Go(f func() error) {
    if g.sem != nil {
        g.sem <- token{}
    }

    g.wg.Add(1)
    go func() {
        defer g.done()

        if err := f(); err != nil {
            g.errOnce.Do(func() {
                g.err = err
                if g.cancel != nil {
                    g.cancel(g.err)
                }
            })
        }
    }()
}

• Wait：すべてのタスクが完了するのを待機し、最初のエラーを返します。

func (g *Group) Wait() error {
    g.wg.Wait()
    if g.cancel != nil {
        g.cancel(g.err)
    }
    return g.err
}

• TryGo：タスクの開始を試みます。

func (g *Group) TryGo(f func() error) bool {
    if g.sem != nil {
        select {
        case g.sem <- token{}:
        default:
            return false
        }
    }

    g.wg.Add(1)
    go func() {
        defer g.done()

        if err := f(); err != nil {
            g.errOnce.Do(func() {
                g.err = err
                if g.cancel != nil {
                    g.cancel(g.err)
                }
            })
        }
    }()
    return true
}

結論

errgroupは、sync.WaitGroupに基づいてエラー処理機能を追加する公式拡張ライブラリであり、同期、エラー伝播、コンテキストキャンセルなどの機能を提供します。そのWithContextメソッドはキャンセル機能を追加でき、SetLimitは同時実行数を制限でき、TryGoはタスクの開始を試みることができます。ソースコードは巧妙に設計されており、参照する価値があります。

Leapcell：Webホスティング、非同期タスク、およびRedis用の次世代サーバーレスプラットフォーム

最後に、golangのデプロイに最適なプラットフォームをお勧めします：Leapcell

1. 多言語サポート

• JavaScript、Python、Go、またはRustで開発。

2. 無料で無制限のプロジェクトをデプロイ

• 使用量に対してのみ支払い—リクエストも請求もありません。

3. 比類のないコスト効率

• アイドル料金なしの従量課金制。
• 例：25ドルで平均応答時間60msで694万リクエストをサポート。

4. 合理化された開発者エクスペリエンス

• 簡単なセットアップのための直感的なUI。
• 完全に自動化されたCI / CDパイプラインとGitOps統合。
• 実用的な洞察を得るためのリアルタイムのメトリックとロギング。

5. 容易なスケーラビリティと高パフォーマンス

• 高い並行性を容易に処理するための自動スケーリング。
• 運用上のオーバーヘッドゼロ—構築に集中するだけです。

ドキュメントで詳細をご覧ください！

Leapcell Twitter：https://x.com/LeapcellHQ