Golang并发模型:轻松入门流水线FAN模式

待兔 等级 593 0 0
标签: golangGolang

前一篇文章《Golang并发模型:轻松入门流水线模型》,介绍了流水线模型的概念,这篇文章是流水线模型进阶,介绍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.png

FAN-IN和FAN-OUT实践

我们这次试用FAN-OUT和FAN-IN,解决《Golang并发模型:轻松入门流水线模型》中提到的问题:计算一个整数切片中元素的平方值并把它打印出来。

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

FAN模式流水线示例

package main

import (
    "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.WaitGroup

    collect := func(in <-chan int) {
        defer wg.Done()
        for n := range in {
            out <- n
        }
    }

    wg.Add(len(cs))
    // FAN-IN
    for _, 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-OUT
    c1 := square(in)
    c2 := square(in)
    c3 := square(in)

    // consumer
    for ret := range merge(c1, c2, c3) {
        fmt.Printf("%3d ", ret)
    }
    fmt.Println()
} 

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

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

FAN模式真能提升性能吗?

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

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

普通流水线

// hi_simple.go

package main

import (
    "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
            // simulate
            time.Sleep(time.Second)
        }
    }()

    return out
}

func main() {
    in := producer(10)
    ch := square(in)

    // consumer
    for _ = range ch {
    }
} 

使用FAN模式的流水线

// hi_fan.go
package main

import (
    "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
            // simulate
            time.Sleep(time.Second)
        }
    }()

    return out
}

func merge(cs ...<-chan int) <-chan int {
    out := make(chan int)

    var wg sync.WaitGroup

    collect := func(in <-chan int) {
        defer wg.Done()
        for n := range in {
            out <- n
        }
    }

    wg.Add(len(cs))
    // FAN-IN
    for _, 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-OUT
    c1 := square(in)
    c2 := square(in)
    c3 := square(in)

    // consumer
    for _ = 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.go

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)
    ch := square(in)

    // consumer
    for _ = range ch {
    }
} 

FAN模式流水线修改代码

// hi_fan.go
package main

import (
    "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-OUT
    c1 := square(in)
    c2 := square(in)
    c3 := square(in)

    // consumer
    for _ = 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,快去实践一把。

并发系列文章推荐

下一篇,写流水线中协程的“优雅退出”,欢迎关注。

如果这篇文章对你有帮助,请点个赞/喜欢,让我知道我的写作是有价值的,感谢。

收藏
评论区

相关推荐

关于Golang的那些事(一) -- Node.js和Golang对比
之前一直用Node.js作为开发语言,用了差不多4年的Node.js,涉及前端和后端,最近看到Golang这个新兴之秀挺火的,于是想探究探究一下这门语言,对比了一下他们的Github repo,截止现在Node.js的repo有72.5K星, issue数量是859个,Golang的repo有75.7K星,issue数量是5K个。从趋势来看,Golang来势
Golang并发模型:轻松入门流水线FAN模式
前一篇文章《Golang并发模型:轻松入门流水线模型》(https://segmentfault.com/a/1190000017142506),介绍了流水线模型的概念,这篇文章是流水线模型进阶,介绍FANIN和FANOUT,FAN模式可以让我们的流水线模型更好的利用Golang并发,提高软件性能。但FAN模式不一定是万能,不见得能提高程序的性能,甚
godoc 命令和 golang 代码文档管理
介绍 godoc 是 golang 自带的文档查看器,更多的提供部署服务 go doc 和 godoc 在 golang 1.13 被移除了,可以自行安装 golang.org go1.13 godoc(https://links.jianshu.com/go?tohttps%3A%2F%2Fgolang.org%2Fdoc%2Fg
Mac安装Golang和vscode
Mac第一次安装golang和vscode一起使用,遇到了不少的坑,下面介绍一下正确的安装方式。 1、使用brew安装Golang 如果不知道brew是什么,或怎么安装请看这里 brew官网(https://brew.sh/index_zhcn) brew install golang 安装完成后可以使用
golang 分析调试高阶技巧
layout: post title: “golang 调试高阶技巧” date: 2020603 1:44:09 0800 categories: golang GC 垃圾回收 golang 高阶调试 Golang tools nm compile
Go语言入门系列(一)之Go的安装和使用
1.安装环境 ====== 1. 进入[Golang官网](https://www.oschina.net/action/GoToLink?url=https%3A%2F%2Fgolang.org),进入下载页面。 (如果打不开可访问[Golang中国](https://www.oschina.net/action/GoToLink?u
Archlinux下Visual Studio Code配置Golang开发环境
一、Golang的安装 ----------- GoLang安装并验证一下: [cox@localhost ~]$ sudo pacman -S go [cox@localhost ~]$ go version go version go1.8.3 linux/amd64s 要注意,Golang的安装要确保两个环境变量,一个是G
Golang Gin实践 番外 请入门 Makefile
<h1>Golang Gin实践 番外 请入门 Makefile</h1> <p>原文地址:<a href="https://github.com/EDDYCJY/blog/blob/master/golang/gin/2018-08-26-Gin%E5%AE%9E%E8%B7%B5-%E7%95%AA%E5%A4%96-%E8%AF%B7%E5%85%A5
Golang In PingCAP
随着 Golang 在后端领域越来越流行,有越来越多的公司选择 Golang 作为主力开发语言。本次 GopherChina Beijing 2016 大会上,看到 Golang 在各家公司从人工智能到自动运维,从 Web 应用到基础架构都发挥着越来越多的作用。可以说 Golang 在这几年间,获得了长足的进步。 PingCAP 是一家由几名 Go
Golang modules 初探
今天天色刚刚亮起,起床看到golang 1.11正式发版了,有着两个重要的特性:modules和WebAssembly。 本博文只要说的是modules,从Java转golang的同学肯定是对golang的包管理充满了无奈之情,我也曾在博客中介绍过[glide](https://my.oschina.net/u/553243/blog/1475626),也
Golang 内存管理源码剖析
Golang 的内存管理基于 tcmalloc,可以说起点挺高的。但是 Golang 在实现的时候还做了很多优化,我们下面通过源码来看一下 Golang 的内存管理实现。下面的源码分析基于 go1.8rc3。 1.tcmalloc 介绍 ------------- 关于 tcmalloc 可以参考这篇文章 [tcmalloc 介绍](https://ww
Golang 开发环境搭建
Golang 是 Google 发布的开发语言,Go 编译的程序速度可以媲美 C/C++。 安装 -- sudo apt-get install golang sudo apt-get install golang-go.tools 使用 -- * 编译运行程序 go run main.go * 查看命令文
Golang 文章正文抽取(readability)
readability for golang Golang版本是根据[readabiliity for node.js](https://www.oschina.net/action/GoToLink?url=https%3A%2F%2Fgithub.com%2Fluin%2Freadability)以及[readability for python](h
Golang依赖管理工具:glide从入门到精通使用
介绍 -- 不论是开发Java还是你正在学习的Golang,都会遇到**依赖管理**问题。Java有牛逼轰轰的Maven和Gradle。 Golang亦有godep、govendor、glide、gvt、gopack等等,本文主要给大家介绍[gilde](https://www.oschina.net/action/GoToLink?url=https%3
Sentinel
![9.28头图.png](https://ucc.alicdn.com/pic/developer-ecology/af7ab6c27c3c4c3aa5dc2cce3c9e8ab9.png) > \*\*导读:\*\*2020年,Sentinel 推出 Go 原生版本[Sentinel-Golang](https://www.oschina.net/ac