怎样用Golang实现管道过滤模式 构建高效数据处理流水线的技巧

package main

import (
    "fmt"
    "strings"
    "sync"
)

// genLines 模拟一个数据源，生成多行字符串
func genLines(lines ...string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for _, line := range lines {
            out <- line
        }
    }()
    return out
}

// filterEmptyLines 过滤空行
func filterEmptyLines(in <-chan string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for line := range in {
            if strings.TrimSpace(line) != "" {
                out <- line
            }
        }
    }()
    return out
}

// toUpperCase 将每行内容转为大写
func toUpperCase(in <-chan string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for line := range in {
            out <- strings.ToUpper(line)
        }
    }()
    return out
}

// countLines 统计行数并打印
func countLines(in <-chan string) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        count := 0
        for line := range in {
            fmt.Println("Processed:", line) // 模拟处理
            count++
        }
        out <- count
    }()
    return out
}

func main() {
    // 构建管道
    lines := genLines("hello world", "", "  go ", "pipeline", " ", "end")
    filtered := filterEmptyLines(lines)
    uppercased := toUpperCase(filtered)
    totalCount := countLines(uppercased)

    // 等待最终结果
    finalCount := <-totalCount
    fmt.Printf("Total processed lines: %d\n", finalCount)
}

// 错误处理示例：使用context.Context
func processWithError(ctx context.Context, in <-chan string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for {
            select {
            case <-ctx.Done(): // 收到取消信号，退出
                fmt.Println("processWithError: Context cancelled, exiting.")
                return
            case line, ok := <-in:
                if !ok { // 输入channel关闭
                    return
                }
                if strings.Contains(line, "error") {
                    fmt.Println("processWithError: Detected error in line, cancelling context.")
                    // 实际应用中，这里会通过context.CancelFunc来取消
                    // 但为了示例简单，我们模拟直接返回，并假定外部会处理取消
                    return // 模拟错误发生，此阶段停止
                }
                out <- strings.ToUpper(line)
            }
        }
    }()
    return out
}

// 主函数中，会有一个类似 errgroup.Group 的结构来管理goroutine和context

// 扇出示例：并行处理数据
func worker(id int, in <-chan string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for line := range in {
            // 模拟耗时操作
            // fmt.Printf("Worker %d processing: %s\n", id, line)
            out <- fmt.Sprintf("Worker %d processed: %s", id, strings.ToUpper(line))
        }
    }()
    return out
}

// 扇入示例：合并多个channel的输出
func merge(cs ...<-chan string) <-chan string {
    var wg sync.WaitGroup
    out := make(chan string)

    // 为每个输入channel启动一个goroutine来复制数据到输出channel
    output := func(c <-chan string) {
        defer wg.Done()
        for n := range c {
            out <- n
        }
    }

    wg.Add(len(cs))
    for _, c := range cs {
        go output(c)
    }

    // 启动一个goroutine在所有输入channel都关闭后关闭输出channel
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

// 在main函数中调用：
// lines := genLines("line1", "line2", "line3", "line4", "line5")
//
// // 扇出到3个worker
// worker1Out := worker(1, lines)
// worker2Out := worker(2, lines)
// worker3Out := worker(3, lines)
//
// // 扇入合并结果
// mergedOut := merge(worker1Out, worker2Out, worker3Out)
//
// for res := range mergedOut {
//  fmt.Println(res)
// }