Go言語の並行処理設計パターン集！パイプライン・ファンアウト徹底解説

Go言語の並行処理設計パターン集（パイプライン・ファンアウトなど）

先生と生徒の会話形式で理解しよう

生徒

「Goで複数の処理を同時に動かすとき、どうやって設計すればいいですか？」

先生

「Goでは、goroutineとchannelを組み合わせて効率的に処理を設計する方法があります。代表的なのはパイプラインとファンアウトのパターンです。」

生徒

「パイプラインとファンアウトって何ですか？」

先生

「簡単に言うと、パイプラインは処理を順番につなげる流れ、ファンアウトは同じ処理を複数のgoroutineで分担するやり方です。」

1. パイプラインパターンとは？

パイプラインは、処理を段階ごとに分けて、それぞれを独立したgoroutineで実行し、channelでデータを渡す設計パターンです。これにより、大きな処理を小さく分けて効率よく処理できます。

例えば、データを読み込む、加工する、出力するという三段階の処理を考えます。各段階を別のgoroutineで実行し、channelでデータを渡すことで、処理全体の速度を向上させられます。

2. パイプラインのサンプルコード


package main

import "fmt"

func main() {
    nums := make(chan int)
    squares := make(chan int)

    // 数字を生成するgoroutine
    go func() {
        for i := 1; i <= 5; i++ {
            nums <- i
        }
        close(nums)
    }()

    // 数字を二乗するgoroutine
    go func() {
        for n := range nums {
            squares <- n * n
        }
        close(squares)
    }()

    // 結果を表示
    for sq := range squares {
        fmt.Println(sq)
    }
}

この例では、数字を生成するgoroutineと二乗するgoroutineがchannelでデータを渡しています。これがパイプラインの基本的な考え方です。

3. ファンアウトパターンとは？

ファンアウトは、同じ入力を複数のgoroutineで処理して負荷を分散する設計パターンです。大量の処理を効率よくさばくときに有効です。

例えば、ウェブから複数のデータを取得して処理する場合、goroutineを複数立ち上げて同時に処理することで、全体の処理時間を短縮できます。

4. ファンアウトのサンプルコード


package main

import (
    "fmt"
    "sync"
)

func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for j := range jobs {
        results <- j * 2
    }
}

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

    // ワーカーを3つ立ち上げる
    for w := 1; w <= 3; w++ {
        wg.Add(1)
        go worker(w, jobs, results, &wg)
    }

    // ジョブを送る
    for j := 1; j <= 5; j++ {
        jobs <- j
    }
    close(jobs)

    // ワーカーの終了を待つ
    go func() {
        wg.Wait()
        close(results)
    }()

    for res := range results {
        fmt.Println(res)
    }
}

この例では、3つのワーカーgoroutineがjobsチャネルからデータを受け取り、処理結果をresultsチャネルに送っています。これがファンアウトの基本形です。

5. パイプラインとファンアウトを組み合わせる

実際のシステムでは、パイプラインとファンアウトを組み合わせることで、高速かつ効率的な並行処理が可能です。パイプラインで段階的に処理しつつ、各段階でファンアウトを使い処理を分散させることで、大量データの処理もスムーズになります。

ポイントは、goroutineを無制限に増やさず、channelで適切にデータを制御することです。バッファ付きchannelを利用すると、処理が詰まるのを防ぎ、安定した並行処理ができます。

6. パイプラインにおけるエラーハンドリングの設計

パイプラインパターンを実装する際には、エラーハンドリングの設計も非常に重要です。並行処理では複数のgoroutineが同時に動いているため、どこでエラーが発生したのか分かりにくくなることがあります。そのため、エラー専用のchannelを用意して、各処理からエラーを集約する設計がよく使われます。

例えば、データ処理の途中で不正な値が見つかった場合、そのエラーを別のchannelに送ることで、メイン処理側でまとめて対応できます。このように設計することで、処理の見通しが良くなり、デバッグもしやすくなります。


package main

import "fmt"

func main() {
    data := make(chan int)
    errs := make(chan string)

    go func() {
        for i := 1; i <= 5; i++ {
            if i == 3 {
                errs <- "エラーが発生しました"
                continue
            }
            data <- i
        }
        close(data)
        close(errs)
    }()

    for {
        select {
        case d, ok := <-data:
            if ok {
                fmt.Println("データ:", d)
            }
        case e, ok := <-errs:
            if ok {
                fmt.Println("エラー:", e)
            } else {
                return
            }
        }
    }
}

このようにエラーを分離して扱うことで、安全で保守性の高い並行処理が実現できます。

7. ファンインパターンで結果をまとめる

ファンアウトで複数のgoroutineに処理を分散した場合、その結果を一つにまとめる必要があります。このときに使われるのがファンインパターンです。複数のchannelからのデータを一つのchannelに集約することで、後続の処理をシンプルにできます。

ファンインは、複数の入力を一つの出口に集めるイメージで、ログ収集や並列処理の結果集約などでよく使われます。


package main

import "fmt"

func main() {
    ch1 := make(chan int)
    ch2 := make(chan int)
    merged := make(chan int)

    go func() {
        for i := 1; i <= 3; i++ {
            ch1 <- i
        }
        close(ch1)
    }()

    go func() {
        for i := 4; i <= 6; i++ {
            ch2 <- i
        }
        close(ch2)
    }()

    go func() {
        for ch1 != nil || ch2 != nil {
            select {
            case v, ok := <-ch1:
                if ok {
                    merged <- v
                } else {
                    ch1 = nil
                }
            case v, ok := <-ch2:
                if ok {
                    merged <- v
                } else {
                    ch2 = nil
                }
            }
        }
        close(merged)
    }()

    for v := range merged {
        fmt.Println(v)
    }
}

このようにファンインを使うことで、複数の処理結果を一つの流れとして扱えるようになります。

Go言語を基礎からスッキリ学びたい人や、文法だけでなく「実用的な使い方」まで押さえたい人には、定番の入門書がこちらです。

基礎からわかるGo言語をAmazonで見る

※ Amazon広告リンク

8. 並行処理設計で重要なベストプラクティス

Go言語の並行処理を安全かつ効率的に行うためには、いくつかのベストプラクティスを理解しておくことが重要です。goroutineは軽量ですが、無制限に増やすとメモリやCPUに負荷がかかります。そのため、適切な数で制御することが大切です。

また、channelのクローズ処理を忘れると、処理が終了しない原因になります。特にパイプラインでは、すべての段階で適切にcloseを行う必要があります。これにより、後続の処理が正しく終了します。

さらに、select文を使うことで複数のchannelを効率よく扱うことができます。これにより、柔軟で拡張性の高い並行処理設計が可能になります。


package main

import "fmt"

func main() {
    ch := make(chan int, 2)

    go func() {
        ch <- 1
        ch <- 2
        close(ch)
    }()

    for v := range ch {
        fmt.Println(v)
    }
}

このような基本を押さえることで、Go言語の並行処理をより安全に扱うことができるようになります。パイプライン、ファンアウト、ファンインを組み合わせて設計することで、大規模なシステムでも効率的に処理を行うことが可能になります。