Go 并发管道：构建无死锁的闭包数据处理流程

p, e, d := NewPipeline() // 创建管道实例，e为输入通道，d为输出通道
p.Add(step1)
p.Add(step2)
p.Add(step3)

go emit(e) // 启动数据发射器
p.Execute() // 执行管道
drain(d) // 消耗输出数据

package main

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

// Widget 示例结构体
type Widget struct {
    ID        int
    Whiz      bool
    Pop       bool
    Bang      bool
    Processed bool
}

// StageMangler 定义了每个处理阶段的业务逻辑
type StageMangler func(*Widget)

// stage 函数是管道中的一个通用阶段
// f: 具体的处理逻辑
// chi: 输入通道 (只读)
// cho: 输出通道 (只写)
func stage(f StageMangler, chi <-chan *Widget, cho chan<- *Widget, wg *sync.WaitGroup) {
    defer wg.Done() // 确保goroutine完成时通知WaitGroup
    defer close(cho) // 确保在函数退出时关闭输出通道

    for widget := range chi {
        // 执行业务逻辑
        f(widget)
        // 将处理后的widget发送到下一个阶段
        cho <- widget
    }
    fmt.Printf("Stage finished processing and closed its output channel.\n")
}

// 示例处理函数
func whizWidgets(w *Widget) {
    time.Sleep(50 * time.Millisecond) // 模拟耗时操作
    w.Whiz = true
    fmt.Printf("Whizzed Widget ID: %d\n", w.ID)
}

func popWidgets(w *Widget) {
    time.Sleep(50 * time.Millisecond)
    w.Pop = true
    fmt.Printf("Popped Widget ID: %d\n", w.ID)
}

func bangWidgets(w *Widget) {
    time.Sleep(50 * time.Millisecond)
    w.Bang = true
    fmt.Printf("Banged Widget ID: %d\n", w.ID)
}

func finalDrain(chi <-chan *Widget, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Println("Starting final drain...")
    for widget := range chi {
        widget.Processed = true
        fmt.Printf("Final Drained Widget: %+v\n", widget)
    }
    fmt.Println("Final drain finished.")
}

func main() {
    var wg sync.WaitGroup

    // 定义管道的通道
    inputChan := make(chan *Widget, 10) // 缓冲通道，防止发送端阻塞
    whizPopChan := make(chan *Widget, 10)
    popBangChan := make(chan *Widget, 10)
    outputChan := make(chan *Widget, 10) // 最终输出通道

    // 启动管道的各个阶段
    wg.Add(1)
    go stage(whizWidgets, inputChan, whizPopChan, &wg)
    wg.Add(1)
    go stage(popWidgets, whizPopChan, popBangChan, &wg)
    wg.Add(1)
    go stage(bangWidgets, popBangChan, outputChan, &wg)

    // 启动数据发射器
    wg.Add(1)
    go func() {
        defer wg.Done()
        defer close(inputChan) // 发射器完成发送后关闭输入通道
        for i := 0; i < 5; i++ {
            widget := &Widget{ID: i}
            fmt.Printf("Emitting Widget ID: %d\n", widget.ID)
            inputChan <- widget
            time.Sleep(20 * time.Millisecond)
        }
        fmt.Println("Input emitter finished and closed input channel.")
    }()

    // 启动最终数据消费者（或称为“排干”阶段）
    wg.Add(1)
    go finalDrain(outputChan, &wg) // finalDrain也需要等待outputChan关闭

    // 等待所有goroutine完成
    wg.Wait()
    fmt.Println("All pipeline stages completed.")
}