Goroutines と Channels を使ってジョブを並行化した時のメモ

var bufferSize = 20

func ProcessItems(items []item.Item) {
	failed := false
	processedCount := 0

	for _, targetItem := range items {
		result, err := item.ProcessItem(targetItem) // 重いブロッキング I/O が中で動く
		if err != nil {
			failed = true
			break // 1個でも失敗したら処理を抜ける
		}
        
		logger.Info(result) // 処理成功時は結果をロガーで出力する

		processedCount++

		if processedCount%bufferSize == 0 {
			// バッファをクリアする処理
		}
	}

	if !failed && processedCount%bufferSize > 0 {
		// バッファをクリアする処理
	}
}

var wg sync.WaitGroup
for _, targetItem := range items {
    wg.Add(1)
    go func(targetItem item.Item) {
        defer wg.Done()
        result, _ := item.ProcessItem(targetItem) // 重いブロッキング I/O が中で動く
        
        logger.Info(result) // 処理成功時は結果をロガーで出力する
    }(targetItem)
}

wg.Wait() // 全ての Goroutine で wg.Done() が呼ばれるまで処理がブロックされる

concurrency := 20 // 同時処理数は20個までとする

chItems := make(chan item.Item) // メインスレッドからワーカーに `item.Item` を受け渡す Channel

var wg sync.WaitGroup
for i := 0; i < concurrency; i++ {
    wg.Add(1)
    go func() {
        defer wg.Done()
        
        // close(chItems) が呼ばれるまで targetItem の取り出しが (Goroutine 間で均等に) 行われる
        for targetItem := range chItems {
            result, _ := item.ProcessItem(targetItem) // 重いブロッキング I/O が中で動く
        
            logger.Info(result) // 処理成功時は結果をロガーで出力する
        }
    }()
}

for _, targetItem := range items {
    chItems <- targetItem
}
close(chItems) // 全て送り出したら close する. これをしないと `for targetItem := range chItems` が正しく動作しない

wg.Wait() // 全ての Goroutine で wg.Done() が呼ばれるまで処理がブロックされる

result, err := item.ProcessItem(targetItem) // 重いブロッキング I/O が中で動く
if err != nil {
    //
}
logger.Info(result)

// 関数の戻り値が多値な為、1つの構造体にまとめる
type taskResult struct {
	result string
	err    error
}

// ...

concurrency := 20 // 同時処理数は20個までとする

chItems := make(chan item.Item)     // メインスレッドからワーカーに `item.Item` を受け渡す Channel
chResults := make(chan *taskResult) // ワーカーからメインスレッドに `item.ProcessItem` の結果を返す Channel

var wg sync.WaitGroup
for i := 0; i < concurrency; i++ {
    wg.Add(1)
    go func() {
        defer wg.Done()

        // close(chItems) が呼ばれるまで targetItem の取り出しが (Goroutine 間で均等に) 行われる
        for targetItem := range chItems {
            r, err := item.ProcessItem(targetItem)        // 重いブロッキング I/O が中で動く
            chResults <- &taskResult{result: r, err: err} // 結果を構造体に入れ、 `chResults` に送出
        }
    }()
}

for _, targetItem := range items {
    chItems <- targetItem
}
close(chItems) // 全て送り出したら close する. これをしないと `for targetItem := range chItems` が正しく動作しない

wg.Wait() // 全ての Goroutine で wg.Done() が呼ばれるまで処理がブロックされる

concurrency := 20 // 同時処理数は20個までとする

chItems := make(chan item.Item)     // メインスレッドからワーカーに `item.Item` を受け渡す Channel
chResults := make(chan *taskResult) // ワーカーからメインスレッドに `item.ProcessItem` の結果を返す Channel

numWorkers := int32(concurrency) // sync.WaitGroup の代わりにカウンタを使う
for i := 0; i < concurrency; i++ {
    go func() {
        defer atomic.AddInt32(&numWorkers, -1) // Goroutine が終わる度にカウンタをデクリメントする. 並行処理中に安全にに処理する為 `atomic.AddInt32` を使う.

        // close(chItems) が呼ばれるまで targetItem の取り出しが (Goroutine 間で均等に) 行われる
        for targetItem := range chItems {
            r, err := item.ProcessItem(targetItem)        // 重いブロッキング I/O が中で動く
            chResults <- &taskResult{result: r, err: err} // 結果を構造体に入れ、 `chResults` に送出
        }
    }()
}

// `targetItem` の送出は別の Goroutine 内で行う
go func() {
    for _, targetItem := range items {
        chItems <- targetItem
    }
    close(chItems)
}()

// ワーカーが全て終了するまで `chResults` から値を取り出し続ける
for numWorkers > 0 {
    result := <-chResults
    if result.err != nil {
        // TODO: エラー処理
    } else {
        logger.Info(result.result) // 処理成功時は結果をロガーで出力する
        
        // TODO: バッファリング処理
    }
}

concurrency := 20 // 同時処理数は20個までとする

chItems := make(chan item.Item)     // メインスレッドからワーカーに `item.Item` を受け渡す Channel
chResults := make(chan *taskResult) // ワーカーからメインスレッドに `item.ProcessItem` の結果を返す Channel

ctx, cancel := context.WithCancel(context.Background()) // 既に使っている `context.Context` があればそれを指定する。今回は無いので `context.Background()` を新たに作って指定
defer cancel()

numWorkers := int32(concurrency) // sync.WaitGroup の代わりにカウンタを使う
for i := 0; i < concurrency; i++ {
    go func() {
        defer atomic.AddInt32(&numWorkers, -1) // Goroutine が終わる度にカウンタをデクリメントする. 並行処理中に安全にに処理する為 `atomic.AddInt32` を使う.

        // close(chItems) が呼ばれるまで targetItem の取り出しが (Goroutine 間で均等に) 行われる
        for targetItem := range chItems {
            r, err := item.ProcessItem(targetItem)        // 重いブロッキング I/O が中で動く
            chResults <- &taskResult{result: r, err: err} // 結果を構造体に入れ、 `chResults` に送出
        }
    }()
}

// `targetItem` の送出は別の Goroutine 内で行う
go func() {
    defer close(chItems) // Goroutine 終了時に確実に close する

    for _, targetItem := range items {
        select {
        case <-ctx.Done():
            // cancel() が実行された後は余計な `item.ProcessItem` を実行したくないので、送出を中止して Goroutine を終了する
            return
        default:
            chItems <- targetItem
        }
    }
}()

failed := false
processedCount := 0

// ワーカーが全て終了するまで `chResults` から値を取り出し続ける
for numWorkers > 0 {
    select {
    case result := <-chResults:
        if failed {
            break
        }

        // エラーを検出したら `cancel()` を実行し、他の Goroutine にそれを通知し、終了する.
        // 但しメイン Goroutine のこの時点では終了せず全てのワーカーが閉じられるまで待つ.
        // (途中でループから抜けると、ワーカー内で `chResults <-` で送出している部分がブロックされてしまい、 Goroutine が終了せずリークしてしまう)
        if result.err != nil {
            logger.Error(fmt.Sprintf("Error detected: %e", result.err))
            cancel()
            failed = true
            break
        }

        logger.Info(result.result) // 処理成功時は結果をロガーで出力する

        processedCount++

        if processedCount%20 == 0 {
            logger.Info("Flush buffer!")
        }
    default:
        // `default` が無いと、 `chResults` deadlock になるので注意
    }
}

if !failed && processedCount%20 > 0 {
    logger.Info("Flush buffer!")
}

// ...

if processedCount%20 == 0 {
    // 20件毎にバッファをクリアする. エラーが発生した時は他と同様以降の処理を止める
    if err := バッファのクリア処理; err != nil {
        logger.Error(fmt.Sprintf("Error detected: %e", err))
        cancel()
        failed = true
    }
}

// ...