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

待兔 等级 313 0 0

前一篇文章《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 中神奇的 slice
声明:本文仅限于简书发布,其他第三方网站均为盗版,原文地址: golang 中神奇的 slice(https://links.jianshu.com/go?tohttps%3A%2F%2Fliqiang.io%2Fpost%2Fimagesliceingolang) 在 golang 中,似乎人们都不太喜欢使用 Linked List,甚至于原
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】GoWeb框架之Gin-简明教程
Gin 简介 Gin is a HTTP web framework written in Go (Golang). It features a
Golang中常用的字符串操作
Golang中常用的字符串操作 一、标准库相关的Package go import( "strings" ) 二、常用字符串操作 1. 判断是否为空字符串 1.1 使用“”进行判断 思路:直接判断是否等于""空字符串,由于Golang中字符串不能为 nil,且为值类型,所以直接与空字符串比较即可。 举例: go
Golang精编100题-搞定golang面试
Golang精编100题 能力模型 | 级别 | 模型 | | | | | 初级 primary | 熟悉基本语法,能够看懂代码的意图; 在他人指导下能够完成用户故事的开发,编写的代码符合CleanCode规范; | | 中级 intermediate | 能够独立完成用户故事的开发和
golang 分析调试高阶技巧
layout: post title: “golang 调试高阶技巧” date: 2020603 1:44:09 0800 categories: golang GC 垃圾回收 golang 高阶调试 Golang tools nm compile
深入理解 Go Slice
(https://imghelloworld.osscnbeijing.aliyuncs.com/0ce8a8773a658d4b843e5796a0dbf001.png) image 原文地址:深入理解 Go Slice(https://github.com/EDDYCJY/blog/blob/master/golang/pkg/20
golang包循环引用的几种解决方案
golang包循环引用的几种解决方案 发表于2020年11月2日2020年11月3日(https://libuba.com/2020/11/02/golang%e5%8c%85%e5%be%aa%e7%8e%af%e5%bc%95%e7%94%a8%e7%9a%84%e5%87%a0%e7%a7%8d%e8%a7%
GO开发[一]:golang语言初探
一.Golang的安装 1.https://dl.gocn.io/ (国内下载地址) (https://imghelloworld.osscnbeijing.aliyuncs.com/658c5d13c377
golang - DES加密ECB(模式)
Java默认DES算法使用DES/ECB/PKCS5Padding,而golang认为这种方式是不安全的,所以故意没有提供这种加密方式,那如果我们还是要用到怎么办?下面贴上golang版的DES ECB加密解密代码(默认对密文做了base64处理)。
聊聊golang的DDD项目结构
序本文主要研究一下golang的DDD项目结构interfacesfoodappserver/interfacesinterfaces git:(master) tree.|____fileupload| |____fileformat.go| |____fileupload.go|____food_handler.go|__
go get下载包失败问题
关于我由于某些不可抗力的原因,国内使用go get命令安装包时会经常会出现timeout的问题。本文介绍几个常用的解决办法。 从github克隆golang在github上建立了一个镜像库,如https://github.com/golang/net就对应是 https://golang.org/x/net的镜像库。 要下载golang.org/x/net包