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

时间  2019-11-07
标签 golang 并发 模型 轻松 入门 流水线 fan 模式 栏目 Go 繁體版
原文   原文链接

前一篇文章《Golang并发模型：轻松入门流水线模型》，介绍了流水线模型的概念，这篇文章是流水线模型进阶，介绍FAN-IN和FAN-OUT，FAN模式可让咱们的流水线模型更好的利用Golang并发，提升软件性能。但FAN模式不必定是万能，不见得能提升程序的性能，甚至还不如普通的流水线。咱们先介绍下FAN模式，再看看它怎么提高性能的，它是否是万能的。git

这篇文章内容略多，原本打算分几回写的，但不如一次读完爽，因此干脆仍是放一篇文章了，要是时间不充足，利用好碎片时间，能够每次看1个标题的内容。github

FAN-IN和FAN-OUT模式

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

  • FAN-OUT模式：多个goroutine从同一个通道读取数据，直到该通道关闭。OUT是一种张开的模式，因此又被称为扇出，能够用来分发任务。
  • FAN-IN模式：1个goroutine从多个通道读取数据，直到这些通道关闭。IN是一种收敛的模式，因此又被称为扇入，用来收集处理的结果。

fan-in和fan-out.png

FAN-IN和FAN-OUT实践

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

  • 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协程并发运行，结果顺序是没法肯定的，因此你获得的结果，不必定与下面的相同。bash

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

FAN模式真能提高性能吗？

相信你内心已经有了答案，能够的。咱们仍是使用老问题，对比一下简单的流水线和FAN模式的流水线，修改下代码，增长程序的执行时间：并发

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

普通流水线：less

// 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，快去实践一把。

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

完整示例代码

本文全部代码都在仓库，可查看完整示例代码：https://github.com/Shitaibin/...

并发系列文章推荐

  1. 若是这篇文章对你有帮助，请点个赞/喜欢，鼓励我持续分享，感谢。
  2. 个人文章列表，点此可查看
  3. 若是喜欢本文，随意转载，但请保留此原文连接。

一块儿学Golang-分享有料的Go语言技术

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程