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。

相关文章

详解如何使用Python从零开始构建文本统计模型

详解如何使用Python从零开始构建文本统计模型

《详解如何使用Python从零开始构建文本统计模型》在自然语言处理领域,词汇表构建是文本预处理的关键环节,本文通过Python代码实践,演示如何从原始文本中提取多尺度特征,并通过动态调整机制构建更精确... 目录一、项目背景与核心思想二、核心代码解析1. 数据加载与预处理2. 多尺度字符统计3. 统计结果可
阅读更多...
SpringBoot整合Sa-Token实现RBAC权限模型的过程解析

SpringBoot整合Sa-Token实现RBAC权限模型的过程解析

《SpringBoot整合Sa-Token实现RBAC权限模型的过程解析》：本文主要介绍SpringBoot整合Sa-Token实现RBAC权限模型的过程解析,本文给大家介绍的非常详细,对大家的学... 目录前言一、基础概念1.1 RBAC模型核心概念1.2 Sa-Token核心功能1.3 环境准备二、表结
阅读更多...
python多线程并发测试过程

python多线程并发测试过程

《python多线程并发测试过程》：本文主要介绍python多线程并发测试过程,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、并发与并行？二、同步与异步的概念？三、线程与进程的区别？需求1：多线程执行不同任务需求2：多线程执行相同任务总结一、并发与并行？1、
阅读更多...
Go语言中使用JWT进行身份验证的几种方式

Go语言中使用JWT进行身份验证的几种方式

《Go语言中使用JWT进行身份验证的几种方式》本文主要介绍了Go语言中使用JWT进行身份验证的几种方式,包括dgrijalva/jwt-go、golang-jwt/jwt、lestrrat-go/jw... 目录简介1. github.com/dgrijalva/jwt-go安装：使用示例：解释：2. gi
阅读更多...
go rate 原生标准限速库的使用

go rate 原生标准限速库的使用

《gorate原生标准限速库的使用》本文主要介绍了Go标准库golang.org/x/time/rate实现限流,采用令牌桶算法控制请求速率,提供Allow/Reserve/Wait方法,具有一定... 目录介绍安装API介绍rate.NewLimiter：创建限流器limiter.Allow()：请求是否
阅读更多...
Go 语言中的 Struct Tag 的用法详解

Go 语言中的 Struct Tag 的用法详解

《Go语言中的StructTag的用法详解》在Go语言中,结构体字段标签（StructTag）是一种用于给字段添加元信息（metadata）的机制,常用于序列化（如JSON、XML）、ORM映... 目录一、结构体标签的基本语法二、json:"token"的具体含义三、常见的标签格式变体四、使用示例五、使用
阅读更多...
Ubuntu上手动安装Go环境并解决“可执行文件格式错误”问题

Ubuntu上手动安装Go环境并解决“可执行文件格式错误”问题

《Ubuntu上手动安装Go环境并解决“可执行文件格式错误”问题》：本文主要介绍Ubuntu上手动安装Go环境并解决“可执行文件格式错误”问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未... 目录一、前言二、系统架构检测三、卸载旧版 Go四、下载并安装正确版本五、配置环境变量六、验证安装七、常见
阅读更多...
Go语言使用slices包轻松实现排序功能

Go语言使用slices包轻松实现排序功能

《Go语言使用slices包轻松实现排序功能》在Go语言开发中,对数据进行排序是常见的需求,Go1.18版本引入的slices包提供了简洁高效的排序解决方案,支持内置类型和用户自定义类型的排序操作,本... 目录一、内置类型排序：字符串与整数的应用1. 字符串切片排序2. 整数切片排序二、检查切片排序状态：
阅读更多...
基于Go语言实现Base62编码的三种方式以及对比分析

基于Go语言实现Base62编码的三种方式以及对比分析

《基于Go语言实现Base62编码的三种方式以及对比分析》Base62编码是一种在字符编码中使用62个字符的编码方式,在计算机科学中,,Go语言是一种静态类型、编译型语言,它由Google开发并开源,... 目录一、标准库现状与解决方案1. 标准库对比表2. 解决方案完整实现代码（含边界处理）二、关键实现细
阅读更多...
Linux高并发场景下的网络参数调优实战指南

Linux高并发场景下的网络参数调优实战指南

《Linux高并发场景下的网络参数调优实战指南》在高并发网络服务场景中,Linux内核的默认网络参数往往无法满足需求,导致性能瓶颈、连接超时甚至服务崩溃,本文基于真实案例分析,从参数解读、问题诊断到优... 目录一、问题背景：当并发连接遇上性能瓶颈1.1 案例环境1.2 初始参数分析二、深度诊断：连接状态与
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2