如何在Stage无Context参数时实现带Context终止的Go管道？

核心思路：由于Stage的签名和实现无法修改，我们只能在Pipeline的入口处为输入通道添加Context监听机制。当Context触发取消（如超时）时，关闭包装后的输入通道，触发后续所有Stage的goroutine依次退出并关闭输出通道，最终终止整个Pipeline的执行。

实现代码 #

首先实现一个辅助函数，用于为输入通道添加Context监听：

func withCancel(ctx context.Context, in In) Out {
    out := make(chan any)
    go func() {
        defer close(out)
        for {
            select {
            case <-ctx.Done():
                // Context取消，立即停止转发并关闭输出通道
                return
            case val, ok := <-in:
                if !ok {
                    // 原始输入通道关闭，终止转发
                    return
                }
                // 发送数据前再次检查Context，避免向已无接收者的通道发送数据导致阻塞
                select {
                case <-ctx.Done():
                    return
                case out <- val:
                }
            }
        }
    }()
    return out
}

然后完成ExecutePipeline函数的实现：

func ExecutePipeline(ctx context.Context, in In, stages ...Stage) Out {
    // 为初始输入通道添加Context监听
    current := withCancel(ctx, in)
    // 依次将通道传入各个Stage处理
    for _, stage := range stages {
        current = stage(current)
    }
    return current
}

工作原理 #

1. withCancel函数会启动一个goroutine，负责将原始输入通道的内容转发到新的输出通道，同时持续监听ctx.Done()信号。
2. 当Context触发取消（如测试中的100ms超时），该goroutine会立即退出并关闭输出通道。
3. 第一个Stage的输入通道即为这个已关闭的通道，Stage内部的goroutine在读取到通道关闭信号后，会执行defer close(out)关闭自身的输出通道。
4. 关闭信号会沿着所有Stage依次传递，最终导致Pipeline的最终输出通道关闭，for range循环自然终止，不会出现无限执行的情况。

注意事项 #

• 所有Stage的实现必须遵循给定模式：在输入通道关闭时，关闭自身的输出通道（这是保证整个Pipeline能正常终止的前提）。
• 如果Stage内部存在长时间阻塞的计算逻辑，仅靠通道关闭无法立即终止该逻辑，但基于给出的Stage模式，这类情况不会影响Pipeline的整体终止流程。

内容的提问来源于stack exchange，提问作者Demps