Go語言非同步API設計的扇入扇出模式詳解

package main
import (
	"context"
	"log"
	"sync"
	"time"
)
// Task 包含任務編號及任務所需時長
type Task struct {
	Number int
	Cost   time.Duration
}
// task channel 生成器
func taskChannelGerenator(ctx context.Context, taskList []Task) <-chan Task {
	taskCh := make(chan Task)
	go func() {
		defer close(taskCh)
		for _, task := range taskList {
			select {
			case <-ctx.Done():
				return
			case taskCh <- task:
			}
		}
	}()
	return taskCh
}
// doTask 處理並返回已處理的任務編號作為通道的函數
func doTask(ctx context.Context, taskCh <-chan Task) <-chan int {
	doneTaskCh := make(chan int)
	go func() {
		defer close(doneTaskCh)
		for task := range taskCh {
			select {
			case <-ctx.Done():
				return
			default:
				log.Printf("do task number: %dn", task.Number)
				// task 任務處理
				// 根據任務耗時休眠
				time.Sleep(task.Cost)
				doneTaskCh <- task.Number // 已處理任務的編號放入通道
			}
		}
	}()
	return doneTaskCh
}
// `fan-in` 意味著將多個資料流複用或合併成一個流。
// merge 函數接收引數傳遞的多個通道 「taskChs」,並返回單個通道 「<-chan int」
func merge(ctx context.Context, taskChs []<-chan int) <-chan int {
	var wg sync.WaitGroup
	mergedTaskCh := make(chan int)
	mergeTask := func(taskCh <-chan int) {
		defer wg.Done()
		for t := range taskCh {
			select {
			case <-ctx.Done():
				return
			case mergedTaskCh <- t:
			}
		}
	}
	wg.Add(len(taskChs))
	for _, taskCh := range taskChs {
		go mergeTask(taskCh)
	}
	// 等待所有任務處理完畢
	go func() {
		wg.Wait()
		close(mergedTaskCh)
	}()
	return mergedTaskCh
}
func main() {
	start := time.Now()
	// 使用 context 來防止 goroutine 洩漏，即使在處理過程中被中斷
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	// taskList 定義每個任務及其成本
	taskList := []Task{
		Task{1, 1 * time.Second},
		Task{2, 7 * time.Second},
		Task{3, 2 * time.Second},
		Task{4, 3 * time.Second},
		Task{5, 5 * time.Second},
		Task{6, 3 * time.Second},
	}
	// taskChannelGerenator 是一個函數，它接收一個 taskList 並將其轉換為 Task 型別的通道
	// 執行結果（int slice channel）儲存在 worker 中
	// 由於 doTask 的結果是一個通道，被分給了多個 worker，這就對應了 fan-out 處理
	taskCh := taskChannelGerenator(ctx, taskList)
	numWorkers := 4
	workers := make([]<-chan int, numWorkers)
	for i := 0; i < numWorkers; i++ {
		workers[i] = doTask(ctx, taskCh)  // doTask 處理並返回已處理的任務編號作為通道的函數
	}
	count := 0
	for d := range merge(ctx, workers) { // merge 從中讀取已處理的任務編號
		count++
		log.Printf("done task number: %dn", d)
	}
	log.Printf("Finished. Done %d tasks. Total time: %fs", count, time.Since(start).Seconds())
}