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。

相关文章

Web服务器-Nginx-高并发问题

Web服务器-Nginx-高并发问题

《Web服务器-Nginx-高并发问题》Nginx通过事件驱动、I/O多路复用和异步非阻塞技术高效处理高并发,结合动静分离和限流策略,提升性能与稳定性... 目录前言一、架构1. 原生多进程架构2. 事件驱动模型3. IO多路复用4. 异步非阻塞 I/O5. Nginx高并发配置实战二、动静分离1. 职责2
阅读更多...
GO语言短变量声明的实现示例

GO语言短变量声明的实现示例

《GO语言短变量声明的实现示例》在Go语言中,短变量声明是一种简洁的变量声明方式,使用:=运算符,可以自动推断变量类型,下面就来具体介绍一下如何使用,感兴趣的可以了解一下... 目录基本语法功能特点与var的区别适用场景注意事项基本语法variableName := value功能特点1、自动类型推
阅读更多...
GO语言中函数命名返回值的使用

GO语言中函数命名返回值的使用

《GO语言中函数命名返回值的使用》在Go语言中,函数可以为其返回值指定名称,这被称为命名返回值或命名返回参数,这种特性可以使代码更清晰,特别是在返回多个值时,感兴趣的可以了解一下... 目录基本语法函数命名返回特点代码示例命名特点基本语法func functionName(parameters) (nam
阅读更多...
Go之errors.New和fmt.Errorf 的区别小结

Go之errors.New和fmt.Errorf 的区别小结

《Go之errors.New和fmt.Errorf的区别小结》本文主要介绍了Go之errors.New和fmt.Errorf的区别,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考... 目录error的基本用法1. 获取错误信息2. 在条件判断中使用基本区别1.函数签名2.使用场景详细对
阅读更多...
Go语言连接MySQL数据库执行基本的增删改查

Go语言连接MySQL数据库执行基本的增删改查

《Go语言连接MySQL数据库执行基本的增删改查》在后端开发中,MySQL是最常用的关系型数据库之一,本文主要为大家详细介绍了如何使用Go连接MySQL数据库并执行基本的增删改查吧... 目录Go语言连接mysql数据库准备工作安装 MySQL 驱动代码实现运行结果注意事项Go语言执行基本的增删改查准备工作
阅读更多...
Spring Security 前后端分离场景下的会话并发管理

Spring Security 前后端分离场景下的会话并发管理

《SpringSecurity前后端分离场景下的会话并发管理》本文介绍了在前后端分离架构下实现SpringSecurity会话并发管理的问题,传统Web开发中只需简单配置sessionManage... 目录背景分析传统 web 开发中的 sessionManagement 入口ConcurrentSess
阅读更多...
Go中select多路复用的实现示例

Go中select多路复用的实现示例

《Go中select多路复用的实现示例》Go的select用于多通道通信,实现多路复用,支持随机选择、超时控制及非阻塞操作,建议合理使用以避免协程泄漏和死循环,感兴趣的可以了解一下... 目录一、什么是select基本语法：二、select 使用示例示例1：监听多个通道输入三、select的特性四、使用se
阅读更多...
Go语言使用Gin处理路由参数和查询参数

Go语言使用Gin处理路由参数和查询参数

《Go语言使用Gin处理路由参数和查询参数》在WebAPI开发中,处理路由参数（PathParameter）和查询参数（QueryParameter）是非常常见的需求,下面我们就来看看Go语言... 目录一、路由参数 vs 查询参数二、Gin 获取路由参数和查询参数三、示例代码四、运行与测试1. 测试编程路
阅读更多...
Go语言使用net/http构建一个RESTful API的示例代码

Go语言使用net/http构建一个RESTful API的示例代码

《Go语言使用net/http构建一个RESTfulAPI的示例代码》Go的标准库net/http提供了构建Web服务所需的强大功能,虽然众多第三方框架（如Gin、Echo）已经封装了很多功能,但... 目录引言一、什么是 RESTful API？二、实战目标：用户信息管理 API三、代码实现1. 用户数据
阅读更多...
MySQL中处理数据的并发一致性的实现示例

MySQL中处理数据的并发一致性的实现示例

《MySQL中处理数据的并发一致性的实现示例》在MySQL中处理数据的并发一致性是确保多个用户或应用程序同时访问和修改数据库时,不会导致数据冲突、数据丢失或数据不一致,MySQL通过事务和锁机制来管理... 目录一、事务（Transactions）1. 事务控制语句二、锁（Locks）1. 锁类型2. 锁粒
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2