Go并发模型：流水线模型

2024-09-08 13:38
文章标签 go 模型 并发 流水线

本文主要是介绍Go并发模型：流水线模型，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

Go作为一个实用主义的编程语言，非常注重性能，在语言特性上天然支持并发，Go并发模型有多种模式，通过流水线模型系列文章，你会更好的使用Go的并发特性，提高的程序性能。

这篇文章主要介绍流水线模型的流水线概念，后面文章介绍流水线模型的FAN-IN和FAN-OUT，最后介绍下如何合理的关闭流水线的协程。

Golang的并发核心思路

Golang并发核心思路是关注数据流动。数据流动的过程交给channel，数据处理的每个环节都交给goroutine，把这些流程画起来，有始有终形成一条线，那就能构成流水线模型。

但我们先从简单的入手。

从一个简单的流水线入手

流水线并不是什么新奇的概念，它能极大的提高生产效率，在当代社会流水线非常普遍，我们用的几乎任何产品（手机、电脑、汽车、水杯），都是从流水线上生产出来的。以汽车为例，整个汽车流水线要经过几百个组装点，而在某个组装点只组装固定的零部件，然后传递给下一个组装点，最终一台完整的汽车从流水线上生产出来。

Golang的并发模型灵感其实都来自我们生活，对软件而言，高的生产效率就是高的性能。

在Golang中，流水线由多个阶段组成，每个阶段之间通过channel连接，每个节点可以由多个同时运行的goroutine组成。

从最简单的流水线入手。下图的流水线由3个阶段组成，分别是A、B、C，A和B之间是通道aCh，B和C之间是通道bCh，A生成数据传递给B，B生成数据传递给C。

流水线中，第一个阶段的协程是生产者，它们只生产数据。最后一个阶段的协程是消费者，它们只消费数据。下图中A是生成者，C是消费者，而B只是中间过程的处理者。

在这里插入图片描述

举个例子，设计一个程序：计算一个整数切片中元素的平方值并把它打印出来。非并发的方式是使用for遍历整个切片，然后计算平方，打印结果。

我们使用流水线模型实现这个简单的功能，从流水线的角度，可以分为3个阶段：

  1. 遍历切片，这是生产者。
  2. 计算平方值。
  3. 打印结果，这是消费者。

下面这段代码：

  • producer()负责生产数据，它会把数据写入通道，并把它写数据的通道返回。
  • square()负责从某个通道读数字，然后计算平方，将结果写入通道，并把它的输出通道返回。
  • main()负责启动producer和square，并且还是消费者，读取suqre的结果，并打印出来。
package mainimport ("fmt"
)func producer(nums ...int) <-chan int {out := make(chan int)go func() {defer close(out)for _, n := range nums {out <- n}}()return out
}func square(inCh <-chan int) <-chan int {out := make(chan int)go func() {defer close(out)for n := range inCh {out <- n * n}}()return out
}func main() {in := producer(1, 2, 3, 4)ch := square(in)// consumerfor ret := range ch {fmt.Printf("%3d", ret)}fmt.Println()
}

结果：

➜  awesome git:(master) ✗ go run hi.go1  4  9 16

这是一种原始的流水线模型，这种原始能让我们掌握流水线的思路。

流水线的特点

  • 每个阶段把数据通过channel传递给下一个阶段。
  • 每个阶段要创建1个goroutine和1个通道，这个goroutine向里面写数据，函数要返回这个通道。
  • 有1个函数来组织流水线，我们例子中是main函数。

如果你没了解过流水线，建议自己把以上的程序写一遍，如果遇到问题解决了，那才真正掌握了流水线模型的思路。

流水线FAN模式

流水线模型进阶，介绍FAN-IN和FAN-OUT，FAN模式可以让我们的流水线模型更好的利用Golang并发，提高软件性能。但FAN模式不一定是万能，不见得能提高程序的性能，甚至还不如普通的流水线。我们先介绍下FAN模式，再看看它怎么提升性能的，它是不是万能的。

FAN-IN和FAN-OUT模式

Golang的并发模式灵感来自现实世界，这些模式是通用的，毫无例外，FAN模式也是对当前世界的模仿。以汽车组装为例，汽车生产线上有个阶段是给小汽车装4个轮子，可以把这个阶段任务交给4个人同时去做，这4个人把轮子都装完后，再把汽车移动到生产线下一个阶段。这个过程中，就有任务的分发，和任务结果的收集。其中任务分发是FAN-OUT，任务收集是FAN-IN。

  • FAN-OUT模式：多个goroutine从同一个通道读取数据，直到该通道关闭。OUT是一种张开的模式，所以又被称为扇出，可以用来分发任务。
  • FAN-IN模式：1个goroutine从多个通道读取数据，直到这些通道关闭。IN是一种收敛的模式，所以又被称为扇入，用来收集处理的结果。
    在这里插入图片描述

FAN-IN和FAN-OUT实践

我们这次试用FAN-OUT和FAN-IN，解决上文中提到的问题：计算一个整数切片中元素的平方值并把它打印出来。

  • producer()保持不变，负责生产数据。
  • squre()也不变，负责计算平方值。
  • 修改main()，启动3个square，这3个squre从producer生成的通道读数据，这是FAN-OUT。
  • 增加merge()，入参是3个square各自写数据的通道，给这3个通道分别启动1个协程，把数据写入到自己创建的通道，并返回该通道，这是FAN-IN。

FAN模式流水线示例：

package mainimport ("fmt""sync"
)func producer(nums ...int) <-chan int {out := make(chan int)go func() {defer close(out)for _, n := range nums {out <- i}}()return out
}func square(inCh <-chan int) <-chan int {out := make(chan int)go func() {defer close(out)for n := range inCh {out <- n * n}}()return out
}func merge(cs ...<-chan int) <-chan int {out := make(chan int)var wg sync.WaitGroupcollect := func(in <-chan int) {defer wg.Done()for n := range in {out <- n}}wg.Add(len(cs))// FAN-INfor _, c := range cs {go collect(c)}// 错误方式：直接等待是bug，死锁，因为merge写了out，main却没有读// wg.Wait()// close(out)// 正确方式go func() {wg.Wait()close(out)}()return out
}func main() {in := producer(1, 2, 3, 4)// FAN-OUTc1 := square(in)c2 := square(in)c3 := square(in)// consumerfor ret := range merge(c1, c2, c3) {fmt.Printf("%3d ", ret)}fmt.Println()
}

3个squre协程并发运行，结果顺序是无法确定的，所以你得到的结果，不一定与下面的相同。

➜  awesome git:(master) ✗ go run hi.go1   4  16   9

FAN模式真能提升性能吗？

相信你心里已经有了答案，可以的。我们还是使用老问题，对比一下简单的流水线和FAN模式的流水线，修改下代码，增加程序的执行时间：

  • produer()使用参数生成指定数量的数据。
  • square()增加阻塞操作，睡眠1s，模拟阶段的运行时间。
  • main()关闭对结果数据的打印，降低结果处理时的IO对FAN模式的对比。

普通流水线：

// hi_simple.gopackage mainimport ("fmt"
)func producer(n int) <-chan int {out := make(chan int)go func() {defer close(out)for i := 0; i < n; i++ {out <- i}}()return out
}func square(inCh <-chan int) <-chan int {out := make(chan int)go func() {defer close(out)for n := range inCh {out <- n * n// simulatetime.Sleep(time.Second)}}()return out
}func main() {in := producer(10)ch := square(in)// consumerfor _ = range ch {}
}

使用FAN模式的流水线：

// hi_fan.go
package mainimport ("sync""time"
)func producer(n int) <-chan int {out := make(chan int)go func() {defer close(out)for i := 0; i < n; i++ {out <- i}}()return out
}func square(inCh <-chan int) <-chan int {out := make(chan int)go func() {defer close(out)for n := range inCh {out <- n * n// simulatetime.Sleep(time.Second)}}()return out
}func merge(cs ...<-chan int) <-chan int {out := make(chan int)var wg sync.WaitGroupcollect := func(in <-chan int) {defer wg.Done()for n := range in {out <- n}}wg.Add(len(cs))// FAN-INfor _, c := range cs {go collect(c)}// 错误方式：直接等待是bug，死锁，因为merge写了out，main却没有读// wg.Wait()// close(out)// 正确方式go func() {wg.Wait()close(out)}()return out
}func main() {in := producer(10)// FAN-OUTc1 := square(in)c2 := square(in)c3 := square(in)// consumerfor _ = range merge(c1, c2, c3) {}
}

多次测试，每次结果近似，结果如下：

  • FAN模式利用了7%的CPU，而普通流水线CPU只使用了3%，FAN模式能够更好的利用CPU，提供更好的并发，提高Golang程序的并发性能。
  • FAN模式耗时10s，普通流水线耗时4s。在协程比较费时时，FAN模式可以减少程序运行时间，同样的时间，可以处理更多的数据。
➜  awesome git:(master) ✗ time go run hi_simple.go
go run hi_simple.go  0.17s user 0.18s system 3% cpu 10.389 total
➜  awesome git:(master) ✗ 
➜  awesome git:(master) ✗ time go run hi_fan.go
go run hi_fan.go  0.17s user 0.16s system 7% cpu 4.288 total

也可以使用Benchmark进行测试，看2个类型的执行时间，结论相同。为了节约篇幅，这里不再介绍，方法和结果贴在Gist了，想看的朋友瞄一眼，或自己动手搞搞。

FAN模式一定能提升性能吗？

FAN模式可以提高并发的性能，那我们是不是可以都使用FAN模式？

不行的，因为FAN模式不一定能提升性能。

依然使用之前的问题，再次修改下代码，其他不变：

  • squre()去掉耗时。
  • main()增加producer()的入参，让producer生产10,000,000个数据。

简单版流水线修改代码：

// hi_simple.gofunc square(inCh <-chan int) <-chan int {out := make(chan int)go func() {defer close(out)for n := range inCh {out <- n * n}}()return out
}func main() {in := producer(10000000)ch := square(in)// consumerfor _ = range ch {}
}

FAN模式流水线修改代码：

// hi_fan.go
package mainimport ("sync"
)func square(inCh <-chan int) <-chan int {out := make(chan int)go func() {defer close(out)for n := range inCh {out <- n * n}}()return out
}func main() {in := producer(10000000)// FAN-OUTc1 := square(in)c2 := square(in)c3 := square(in)// consumerfor _ = range merge(c1, c2, c3) {}
}

结果，可以跑多次，结果近似：

➜  awesome git:(master) ✗ time go run hi_simple.go    
go run hi_simple.go  9.96s user 5.93s system 168% cpu 9.424 total
➜  awesome git:(master) ✗ time go run hi_fan.go        
go run hi_fan.go  23.35s user 11.51s system 297% cpu 11.737 total

从这个结果，我们能看到2点。

  • FAN模式可以提高CPU利用率。
  • FAN模式不一定能提升效率，降低程序运行时间。

优化FAN模式

既然FAN模式不一定能提高性能，如何优化？

不同的场景优化不同，要依具体的情况，解决程序的瓶颈。

我们当前程序的瓶颈在FAN-IN，squre函数很快就完成，merge函数它把3个数据写入到1个通道的时候出现了瓶颈，适当使用带缓冲通道可以提高程序性能，再修改下代码

merge()中的out修改为：

out := make(chan int, 100)

结果：

➜  awesome git:(master) ✗ time go run hi_fan_buffered.go 
go run hi_fan_buffered.go  19.85s user 8.19s system 323% cpu 8.658 total

使用带缓存通道后，程序的性能有了较大提升，CPU利用率提高到323%，提升了8%，运行时间从11.7降低到8.6，降低了26%。

FAN模式的特点很简单，相信你已经掌握了，如果记不清了看这里，本文所有代码在该Github仓库。
FAN模式很有意思，并且能提高Golang并发的性能，如果想以后运用自如，用到自己的项目中去，还是要写写自己的Demo，快去实践一把。

这篇关于Go并发模型：流水线模型的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1148260。

相关文章

0基础租个硬件玩deepseek,蓝耘元生代智算云|本地部署DeepSeek R1模型的操作流程

0基础租个硬件玩deepseek,蓝耘元生代智算云|本地部署DeepSeek R1模型的操作流程

《0基础租个硬件玩deepseek,蓝耘元生代智算云|本地部署DeepSeekR1模型的操作流程》DeepSeekR1模型凭借其强大的自然语言处理能力,在未来具有广阔的应用前景,有望在多个领域发... 目录0基础租个硬件玩deepseek,蓝耘元生代智算云|本地部署DeepSeek R1模型，3步搞定一个应
阅读更多...
Go路由注册方法详解

Go路由注册方法详解

《Go路由注册方法详解》Go语言中,http.NewServeMux()和http.HandleFunc()是两种不同的路由注册方式,前者创建独立的ServeMux实例,适合模块化和分层路由,灵活性高... 目录Go路由注册方法1. 路由注册的方式2. 路由器的独立性3. 灵活性4. 启动服务器的方式5.
阅读更多...
Deepseek R1模型本地化部署+API接口调用详细教程(释放AI生产力)

Deepseek R1模型本地化部署+API接口调用详细教程(释放AI生产力)

《DeepseekR1模型本地化部署+API接口调用详细教程(释放AI生产力)》本文介绍了本地部署DeepSeekR1模型和通过API调用将其集成到VSCode中的过程,作者详细步骤展示了如何下载和... 目录前言一、deepseek R1模型与chatGPT o1系列模型对比二、本地部署步骤1.安装oll
阅读更多...
Spring AI Alibaba接入大模型时的依赖问题小结

Spring AI Alibaba接入大模型时的依赖问题小结

《SpringAIAlibaba接入大模型时的依赖问题小结》文章介绍了如何在pom.xml文件中配置SpringAIAlibaba依赖,并提供了一个示例pom.xml文件,同时,建议将Maven仓... 目录（一）pom.XML文件：（二）application.yml配置文件（一）pom.xml文件：首
阅读更多...
Go语言中三种容器类型的数据结构详解

Go语言中三种容器类型的数据结构详解

《Go语言中三种容器类型的数据结构详解》在Go语言中,有三种主要的容器类型用于存储和操作集合数据：本文主要介绍三者的使用与区别,感兴趣的小伙伴可以跟随小编一起学习一下... 目录基本概念1. 数组（Array）2. 切片（Slice）3. 映射（Map）对比总结注意事项基本概念在 Go 语言中，有三种主要
阅读更多...
Go Mongox轻松实现MongoDB的时间字段自动填充

Go Mongox轻松实现MongoDB的时间字段自动填充

《GoMongox轻松实现MongoDB的时间字段自动填充》这篇文章主要为大家详细介绍了Go语言如何使用mongox库,在插入和更新数据时自动填充时间字段,从而提升开发效率并减少重复代码,需要的可以... 目录前言时间字段填充规则Mongox 的安装使用 Mongox 进行插入操作使用 Mongox 进行更
阅读更多...
如何在本地部署 DeepSeek Janus Pro 文生图大模型

如何在本地部署 DeepSeek Janus Pro 文生图大模型

《如何在本地部署DeepSeekJanusPro文生图大模型》DeepSeekJanusPro模型在本地成功部署,支持图片理解和文生图功能,通过Gradio界面进行交互,展示了其强大的多模态处... 目录什么是 Janus Pro1. 安装 conda2. 创建 python 虚拟环境3. 克隆 janus
阅读更多...
本地私有化部署DeepSeek模型的详细教程

本地私有化部署DeepSeek模型的详细教程

《本地私有化部署DeepSeek模型的详细教程》DeepSeek模型是一种强大的语言模型,本地私有化部署可以让用户在自己的环境中安全、高效地使用该模型,避免数据传输到外部带来的安全风险,同时也能根据自... 目录一、引言二、环境准备（一）硬件要求（二）软件要求（三）创建虚拟环境三、安装依赖库四、获取 Dee
阅读更多...
Go语言利用泛型封装常见的Map操作

Go语言利用泛型封装常见的Map操作

《Go语言利用泛型封装常见的Map操作》Go语言在1.18版本中引入了泛型,这是Go语言发展的一个重要里程碑,它极大地增强了语言的表达能力和灵活性,本文将通过泛型实现封装常见的Map操作,感... 目录什么是泛型泛型解决了什么问题Go泛型基于泛型的常见Map操作代码合集总结什么是泛型泛型是一种编程范式，允
阅读更多...
DeepSeek模型本地部署的详细教程

DeepSeek模型本地部署的详细教程

《DeepSeek模型本地部署的详细教程》DeepSeek作为一款开源且性能强大的大语言模型,提供了灵活的本地部署方案,让用户能够在本地环境中高效运行模型,同时保护数据隐私,在本地成功部署DeepSe... 目录一、环境准备（一）硬件需求（二）软件依赖二、安装Ollama三、下载并部署DeepSeek模型选
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2