go-concurrency

Go の並行処理スキル

このスキルでは、goroutine、チャネル、同期、並行システム構築のベストプラクティスなど、Go の並行処理プリミティブとパターンについて専門的なガイダンスを提供します。

使用するタイミング

次の場合にこのスキルを有効にしてください：

• 並行／並列処理を実装するとき
• goroutine とチャネルを扱うとき
• 同期プリミティブ（mutex、wait group など）を使用するとき
• データ競合をデバッグするとき
• 並行処理のパフォーマンスを最適化するとき
• ワーカープールやパイプラインを実装するとき
• コンテキストのキャンセルを処理するとき

Goroutine の基本

Goroutine の基本

// Simple goroutine
go func() {
    fmt.Println("Hello from goroutine")
}()

// Goroutine with parameters
go func(msg string) {
    fmt.Println(msg)
}("Hello")

// Goroutine with closure
message := "Hello"
go func() {
    fmt.Println(message) // Captures message
}()

よくある落とし穴

// ❌ BAD: Loop variable capture
for i := 0; i < 5; i++ {
    go func() {
        fmt.Println(i) // All goroutines may print 5
    }()
}

// ✅ GOOD: Pass as parameter
for i := 0; i < 5; i++ {
    go func(n int) {
        fmt.Println(n) // Each prints correct value
    }(i)
}

// ✅ GOOD: Create local copy
for i := 0; i < 5; i++ {
    i := i // Create new variable
    go func() {
        fmt.Println(i)
    }()
}

チャネルのパターン

チャネルの種類

// Unbuffered channel (synchronous)
ch := make(chan int)

// Buffered channel (asynchronous up to buffer size)
ch := make(chan int, 10)

// Send-only channel
func send(ch chan<- int) {
    ch <- 42
}

// Receive-only channel
func receive(ch <-chan int) {
    value := <-ch
}

// Bidirectional channel
ch := make(chan int)

チャネル操作

// Send
ch <- value

// Receive
value := <-ch

// Receive with ok check
value, ok := <-ch
if !ok {
    // Channel closed
}

// Close channel
close(ch)

// Range over channel
for value := range ch {
    fmt.Println(value)
}

Select 文

// Wait for first available operation
select {
case msg1 := <-ch1:
    fmt.Println("Received from ch1:", msg1)
case msg2 := <-ch2:
    fmt.Println("Received from ch2:", msg2)
case ch3 <- value:
    fmt.Println("Sent to ch3")
default:
    fmt.Println("No channels ready")
}

// Timeout pattern
select {
case result := <-ch:
    return result, nil
case <-time.After(5 * time.Second):
    return nil, errors.New("timeout")
}

// Context cancellation
select {
case result := <-ch:
    return result, nil
case <-ctx.Done():
    return nil, ctx.Err()
}

同期プリミティブ

Mutex

type SafeCounter struct {
    mu    sync.Mutex
    count int
}

func (c *SafeCounter) Increment() {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.count++
}

func (c *SafeCounter) Value() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.count
}

RWMutex

type Cache struct {
    mu    sync.RWMutex
    items map[string]interface{}
}

func (c *Cache) Get(key string) (interface{}, bool) {
    c.mu.RLock() // Multiple readers allowed
    defer c.mu.RUnlock()
    value, ok := c.items[key]
    return value, ok
}

func (c *Cache) Set(key string, value interface{}) {
    c.mu.Lock() // Exclusive write access
    defer c.mu.Unlock()
    c.items[key] = value
}

WaitGroup

func processItems(items []Item) {
    var wg sync.WaitGroup

    for _, item := range items {
        wg.Add(1)
        go func(item Item) {
            defer wg.Done()
            process(item)
        }(item)
    }

    wg.Wait() // Wait for all goroutines
}

Once

type Database struct {
    instance *sql.DB
    once     sync.Once
}

func (d *Database) GetConnection() *sql.DB {
    d.once.Do(func() {
        d.instance, _ = sql.Open("postgres", "connection-string")
    })
    return d.instance
}

並行処理のパターン

ワーカープール

type WorkerPool struct {
    workerCount int
    jobs        chan Job
    results     chan Result
    wg          sync.WaitGroup
}

type Job struct {
    ID   int
    Data interface{}
}

type Result struct {
    JobID int
    Value interface{}
    Error error
}

func NewWorkerPool(workerCount int) *WorkerPool {
    return &WorkerPool{
        workerCount: workerCount,
        jobs:        make(chan Job, 100),
        results:     make(chan Result, 100),
    }
}

func (p *WorkerPool) Start(ctx context.Context) {
    for i := 0; i < p.workerCount; i++ {
        p.wg.Add(1)
        go p.worker(ctx)
    }
}

func (p *WorkerPool) worker(ctx context.Context) {
    defer p.wg.Done()

    for {
        select {
        case job, ok := <-p.jobs:
            if !ok {
                return
            }
            result := processJob(job)
            select {
            case p.results <- result:
            case <-ctx.Done():
                return
            }
        case <-ctx.Done():
            return
        }
    }
}

func (p *WorkerPool) Submit(job Job) {
    p.jobs <- job
}

func (p *WorkerPool) Results() <-chan Result {
    return p.results
}

func (p *WorkerPool) Close() {
    close(p.jobs)
    p.wg.Wait()
    close(p.results)
}

// Usage
ctx := context.Background()
pool := NewWorkerPool(10)
pool.Start(ctx)

for i := 0; i < 100; i++ {
    pool.Submit(Job{ID: i, Data: fmt.Sprintf("job-%d", i)})
}

go func() {
    for result := range pool.Results() {
        if result.Error != nil {
            log.Printf("Job %d failed: %v", result.JobID, result.Error)
        } else {
            log.Printf("Job %d completed: %v", result.JobID, result.Value)
        }
    }
}()

pool.Close()

パイプラインパターン

// Generator stage
func generator(ctx context.Context, nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            select {
            case out <- n:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

// Processing stage
func square(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            select {
            case out <- n * n:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

// Another processing stage
func double(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            select {
            case out <- n * 2:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

// Usage - compose pipeline
ctx := context.Background()
numbers := generator(ctx, 1, 2, 3, 4, 5)
squared := square(ctx, numbers)
doubled := double(ctx, squared)

for result := range doubled {
    fmt.Println(result)
}

Fan-Out／Fan-In

// Fan-out: distribute work to multiple goroutines
func fanOut(ctx context.Context, input <-chan int, workers int) []<-chan int {
    channels := make([]<-chan int, workers)

    for i := 0; i < workers; i++ {
        channels[i] = worker(ctx, input)
    }

    return channels
}

func worker(ctx context.Context, input <-chan int) <-chan int {
    output := make(chan int)
    go func() {
        defer close(output)
        for n := range input {
            select {
            case output <- expensiveOperation(n):
            case <-ctx.Done():
                return
            }
        }
    }()
    return output
}

// Fan-in: merge multiple channels into one
func fanIn(ctx context.Context, channels ...<-chan int) <-chan int {
    var wg sync.WaitGroup
    output := make(chan int)

    multiplex := func(ch <-chan int) {
        defer wg.Done()
        for n := range ch {
            select {
            case output <- n:
            case <-ctx.Done():
                return
            }
        }
    }

    wg.Add(len(channels))
    for _, ch := range channels {
        go multiplex(ch)
    }

    go func() {
        wg.Wait()
        close(output)
    }()

    return output
}

// Usage
ctx := context.Background()
input := generator(ctx, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

// Fan-out to 3 workers
workers := fanOut(ctx, input, 3)

// Fan-in results
results := fanIn(ctx, workers...)

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

セマフォパターン

type Semaphore struct {
    sem chan struct{}
}

func NewSemaphore(maxConcurrency int) *Semaphore {
    return &Semaphore{
        sem: make(chan struct{}, maxConcurrency),
    }
}

func (s *Semaphore) Acquire() {
    s.sem <- struct{}{}
}

func (s *Semaphore) Release() {
    <-s.sem
}

// Usage
sem := NewSemaphore(5) // Max 5 concurrent operations

for _, item := range items {
    sem.Acquire()
    go func(item Item) {
        defer sem.Release()
        process(item)
    }(item)
}

レート制限

// Token bucket rate limiter
type RateLimiter struct {
    ticker   *time.Ticker
    tokens   chan struct{}
}

func NewRateLimiter(rate time.Duration, burst int) *RateLimiter {
    rl := &RateLimiter{
        ticker: time.NewTicker(rate),
        tokens: make(chan struct{}, burst),
    }

    // Fill bucket initially
    for i := 0; i < burst; i++ {
        rl.tokens <- struct{}{}
    }

    // Refill tokens
    go func() {
        for range rl.ticker.C {
            select {
            case rl.tokens <- struct{}{}:
            default:
            }
        }
    }()

    return rl
}

func (rl *RateLimiter) Wait(ctx context.Context) error {
    select {
    case <-rl.tokens:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

func (rl *RateLimiter) Stop() {
    rl.ticker.Stop()
}

// Usage
limiter := NewRateLimiter(time.Second/10, 5) // 10 requests per second, burst of 5
defer limiter.Stop()

for _, request := range requests {
    if err := limiter.Wait(ctx); err != nil {
        log.Printf("Rate limit error: %v", err)
        continue
    }
    processRequest(request)
}

並行コードでのエラー処理

errgroup パッケージ

import "golang.org/x/sync/errgroup"

func fetchURLs(ctx context.Context, urls []string) error {
    g, ctx := errgroup.WithContext(ctx)

    for _, url := range urls {
        url := url // Capture for goroutine
        g.Go(func() error {
            return fetchURL(ctx, url)
        })
    }

    // Wait for all goroutines, return first error
    return g.Wait()
}

// With limited concurrency
func fetchURLsLimited(ctx context.Context, urls []string) error {
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(10) // Max 10 concurrent

    for _, url := range urls {
        url := url
        g.Go(func() error {
            return fetchURL(ctx, url)
        })
    }

    return g.Wait()
}

ベストプラクティス

1. チャネルは必ず送信側で閉じてください
2. キャンセルとタイムアウトには context を使用してください
3. goroutine のリークを避け、終了できることを確認してください
4. ブロッキングを避けるため、バッファ付きチャネルを使用してください
5. 読み取りが多いワークロードでは sync.RWMutex を優先してください
6. 処理の負荷が高いループ内で defer を使用しないでください
7. レースディテクターを使ってテストしてください：go test -race
8. エラー伝播には errgroup を使用してください
9. ワーカープールで同時実行操作の数を制限してください
10. 最適化する前にプロファイリングしてください

データ競合の検出

# Run tests with race detector
go test -race ./...

# Run program with race detector
go run -race main.go

# Build with race detector
go build -race

避けるべき一般的なパターン

// ❌ BAD: Unbounded goroutine creation
for _, item := range millionItems {
    go process(item) // May create millions of goroutines
}

// ✅ GOOD: Use worker pool
pool := NewWorkerPool(100)
for _, item := range millionItems {
    pool.Submit(item)
}

// ❌ BAD: Goroutine leak
func leak() <-chan int {
    ch := make(chan int)
    go func() {
        ch <- expensiveComputation() // If receiver never reads, goroutine leaks
    }()
    return ch
}

// ✅ GOOD: Use context for cancellation
func noLeak(ctx context.Context) <-chan int {
    ch := make(chan int)
    go func() {
        defer close(ch)
        result := expensiveComputation()
        select {
        case ch <- result:
        case <-ctx.Done():
        }
    }()
    return ch
}

リソース

追加の例とパターンは以下にあります：

• assets/examples/ - 完全な並行処理の例
• assets/patterns/ - 一般的な並行処理パターン
• references/ - Go の並行処理に関するリソースや論文へのリンク