当前位置：首页 > 文章列表 > Golang > Go教程 > Golangchannel流水线案例详解

Golangchannel流水线案例详解

2025-08-25 13:31:00 0浏览收藏

推广推荐

支持 PC / 移动端，安全直达

在Golang实战开发的过程中，我们经常会遇到一些这样那样的问题，然后要卡好半天，等问题解决了才发现原来一些细节知识点还是没有掌握好。今天golang学习网就整理分享《Golang channel流水线实现案例解析》，聊聊，希望可以帮助到正在努力赚钱的你。

使用Go的channel和goroutine实现整数处理流水线：生成整数→平方→过滤大于100→输出结果，各阶段通过channel串联，每个阶段函数接收输入channel并返回输出channel，形成可组合、高效的数据处理链。

Golang管道模式实现 channel流水线案例

在Go语言中，channel和goroutine的组合非常适合实现“流水线（pipeline）”模式。流水线将数据处理过程拆分为多个阶段，每个阶段由一个或多个goroutine通过channel连接，形成一条高效、可扩展的数据处理链。下面通过一个具体的案例来展示如何用Golang实现channel流水线。

1. 场景说明：整数处理流水线

假设我们需要实现一个数据处理流程：生成一组整数 → 平方每个数 → 过滤出大于100的结果 → 输出最终结果。我们可以将每个步骤封装为一个独立的函数，通过channel串联起来。

2. 阶段函数实现

每个阶段接收一个输入channel，返回一个输出channel，形成可组合的流水线结构。

阶段1：生成数据

gen函数生成一组整数并通过channel输出。

func gen(nums ...int)   out := make(chan int, len(nums))
  go func() {
    for _, n := range nums {
      out <- n
    }
    close(out)
  }()
  return out
}

阶段2：平方处理

square函数从输入channel读取数据，计算平方后发送到输出channel。

func square(in <-chan int)   out := make(chan int)
  go func() {
    for n := range in {
      out <- n * n
    }
    close(out)
  }()
  return out
}

阶段3：过滤大于100的数据

filterOver100函数只传递大于100的值。

func filterOver100(in <-chan int)   out := make(chan int)
  go func() {
    for n := range in {
      if n > 100 {
        out <- n
      }
    }
    close(out)
  }()
  return out
}

3. 组合流水线并运行

将各个阶段通过channel串联起来，形成完整的处理流程。

func main() {
  // 构建流水线
  source := gen(1, 2, 3, 4, 5, 6, 7, 8, 9, 10) // 1~10
  squared := square(source) // 平方
  filtered := filterOver100(squared) // 过滤 >100

  // 消费结果
  for result := range filtered {
    fmt.Println(result)
  }
}

输出结果：
121 （11²）
144 （12²）
169 （13²）
196 （14²）
225 （15²）
... 以此类推，实际输入是1~10，平方后最大为100，因此实际无输出。若想看到输出，可将gen改为 gen(11,12,13) 或调整输入。

4. 优化：支持多阶段并行与扇出/扇入

在高并发场景中，可以对某个阶段启动多个worker，提高处理能力。

func squareParallel(in <-chan int, workers int)   out := make(chan int, workers)

  var wg sync.WaitGroup
  for i := 0; i     wg.Add(1)
    go func() {
      for n := range in {
        out <- n * n
      }
      wg.Done()
    }()
  }

  go func() {
    wg.Wait()
    close(out)
  }()
  return out
}