Go の同時実行パターン。ワーカープールとファンアウト/ファンイン-Golang-php.cn

Go の同時実行パターン。ワーカープールとファンアウト/ファンイン

DDD

リリース： 2024-10-07 22:11:01

オリジナル

656 人が閲覧しました

Concurrency patterns in Go; worker pools and fan-out/fan-in

Go はその優れた同時実行モデルで知られていますが、多くの開発者はゴルーチンとチャネルのみに焦点を当てています。ただし、ワーカープールやファンアウト/ファンインなどの同時実行パターンにより、実際の効率が得られます。

この記事では、これらの高度な概念について説明し、Go アプリケーションのスループットを最大化するのに役立ちます。

同時実行性が重要な理由

同時実行性により、特に I/O 操作、Web リクエスト、バックグラウンド処理などのタスクを処理する場合、プログラムはタスクを効率的に実行できます。 Go では、ゴルーチンは何千もの同時タスクを管理するための軽量な方法を提供しますが、構造がなければボトルネックに遭遇する可能性があります。そこでワーカープールとファンアウト/ファンインパターンが登場します。

ワーカープール

ワーカープールを使用すると、タスクを固定の「ワーカー」に割り当てることでゴルーチンの数を制限できます。これにより、オーバーサブスクリプションが防止され、リソース消費が削減され、タスクの実行が管理可能になります。


package main

import (
    "fmt"
    "sync"
    "time"
)

func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for j := range jobs {
        fmt.Printf("Worker %d started job %d\n", id, j)
        time.Sleep(time.Second) // Simulate work
        fmt.Printf("Worker %d finished job %d\n", id, j)
        results <- j * 2
    }
}

func main() {
    jobs := make(chan int, 100)
    results := make(chan int, 100)
    var wg sync.WaitGroup

    // Start 3 workers
    for w := 1; w <= 3; w++ {
        wg.Add(1)
        go worker(w, jobs, results, &wg)
    }

    // Send jobs
    for j := 1; j <= 5; j++ {
        jobs <- j
    }
    close(jobs)

    // Wait for workers to finish
    wg.Wait()
    close(results)

    for result := range results {
        fmt.Println("Result:", result)
    }
}

ログイン後にコピー

この例では:

ジョブを同時に処理する 3 人のワーカーがいます。
各ジョブはチャネル経由でワーカーに渡され、結果は処理のために収集されます。

ファンアウト/ファンインパターン

ファンアウト/ファンインパターンにより、複数のゴルーチンが同じタスクを処理できるようになり、ファンインでは結果が 1 つの出力に収集されます。これは、タスクを分割して結果を集計する場合に便利です。


package main

import (
    "fmt"
    "sync"
    "time"
)

func workerFanOut(id int, tasks <-chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    for task := range tasks {
        fmt.Printf("Worker %d processing task %d\n", id, task)
        time.Sleep(time.Second) // Simulate work
    }
}

func main() {
    var wg sync.WaitGroup
    tasks := make(chan int, 10)

    // Fan-out: Launch multiple workers
    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go workerFanOut(i, tasks, &wg)
    }

    // Send tasks
    for i := 1; i <= 9; i++ {
        tasks <- i
    }
    close(tasks)

    // Wait for workers to finish
    wg.Wait()

    fmt.Println("All tasks are processed.")
}

ログイン後にコピー

上記のコード内: