15

Golang协程并发的流水线模型

 3 years ago
source link: https://studygolang.com/articles/31647
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.

背景

最近由于性能问题,后端服务一直在做python到golang的迁移和重构。go语言精简优雅,既有编译型语言的严谨和高性能,又有解释型语言的开发效率,出色的并发性能也是go区别于其他语言的一大特色。go的并发编程代码虽然简单,但重在其并发模型和流程的设计。所以这里总结下golang协程并发常用的流水线模型。

简单的流水线思维

流水线模式并不是什么新奇的概念,但是它能极大地提高生产效率。比如实际生活中的汽车生产流水线,流水线上的每一个流程负责不同的工作,比如第一个流程是拼装车身,第二个流程是安装发动机,第三个流程是装轮胎...,这些步骤我们可以类比成go并发流程中的协程,每一个协程就是一个任务。流水线上面传递的车身、发动机、轮胎,这些我们可以类比成协程间需要传递的数据,而在这些流程(协程)间传递这些配件(数据),自然就要通过传送带(channel)。在流水线上,我们装四个轮胎肯定不是一个一个来装的,肯定是有四个机械臂同时来装。因此装轮胎这个步骤我们有4个协程在并发工作来提高效率。这么一来,流水线模型的基本要素就构成了。

Golang的并发模型灵感其实都来自我们生活,对程序而言,高的生产效率就是高的性能。 在Golang中,流水线由多个流程节点组成,流程之间通过channel连接,每个流程节点可以由多个同时运行的goroutine组成。

vUVNRfU.png!mobile

如何构造流水线

有了流水线模式的思维,接下来就是如何构造流水线了。简单来说,其实就是通过channel将任务流程连接起来,两个相邻的流程互为生产者和消费者,通过channel进行通信。耗时的流程可以将任务分散到多个协程来执行。

我们先来看一个最简单的流水线,如下图,A是生产者流程,B是它的消费流程,同时又是C的生产者流程。A,B,C三个协程直接,通过读写channel进行通信。

ZjeQJbY.png!mobile

那如果此时B流程可以将a channel中的任务并发执行呢,很简单,我们只需要起多个B协程就可以了。如下图。

YbqeAb.png!mobile

总之,我们构造流水线并发的思路是关注数据的流动,数据流动的过程交给channel,channel两端数据处理的每个环节都交给goroutine,这个流程连起来,就构成了流水线模型。

关于channel

为什么我们可以选择channel来进行协程间的通信呢,协程之间又是怎么保持同步顺序呢,当然这都要归功于channel。channel是go提供的进程内协程间的通信方式,它是协程/线程安全的,channe的读写阻塞会导致协程的切换。

channel的操作和状态组合可以有以下几种情况:

VfUZziF.png!mobile

**有1个特殊场景**:当`nil`的通道在`select`的某个`case`中时,这个case会阻塞,但不会造成死锁。

channel不仅可以保证协程安全的数据流动,还可以保证协程的同步。当有并发问题时,channel也是我们首先应该想到的数据结构。不过,当使用有缓冲区的channel时,才能达到协程并发的效果,并且生产者和消费者的协程间是相对同步的。使用无缓冲区的channel时,是没有并发效果的,协程间是绝对同步的,生产者和消费者必须同时写和读协程才能运行。

channel关注的是数据的流动,这种场景下都可以考虑使用channel。比如:消息传递、信号广播、任务分发、结果汇总、同步与异步、并发控制... 更多的不在这里赘述了,总之, Share memory by communicating, don't communicate by sharing memory.

流水线模型实例

举个简单栗子,计算80000以内的质数并输出。

这个例子如果我们采用非并发的方式,就是for循环80000,挨个判断是不是素数再输出。不过如果我们采用流水线的并发模型会更高效。

从数据流动的角度来分析,需要遍历生成1-80000的数字到一个channel中,数字判断是否为素数,输出结果到一个channel中。因此我们需要两个channel,channel的两端就设计成协程即可。

1、遍历生成原始80000个数据(生产者)

2、计算这80000个数据中的素数(生产者+消费者)

3、取结果输出(消费者)

package gen_channel
import "fmt"
import "time"
func generate_source(data_source_chan chan int) {
   for i := 1; i <= 80000; i++ {
      data_source_chan <- i
   }
   fmt.Println("写入协程结束")
   close(data_source_chan)
}
func generate_sushu(data_source_chan chan int, data_result_chan chan int, gen_chan chan bool) {
   for num:= range data_source_chan {
      falg := true
 for i := 2; i < num; i++ {
         if num%i == 0 {
            falg = false
 break }
      }
      if falg == true {
         data_result_chan <- num
      }
   }
   fmt.Println("该协程结束")
   gen_chan <- true
}
func workpool(data_source_chan chan int, data_result_chan chan int, gen_chan chan bool, gen_num int){
   // 开启8个协程
 for i := 0; i < gen_num; i++ {
      go generate_sushu(data_source_chan, data_result_chan, gen_chan)
   }
}
func Channel_main() {
   data_source_chan := make(chan int, 2000)
   data_result_chan := make(chan int, 2000)
   gen_chan := make(chan bool, 8)
   time1 := time.Now().Unix()
   go generate_source(data_source_chan)
   // 协程池,任务分发
   workpool(data_source_chan, data_result_chan, gen_chan, 8)
   go func() {
      for i := 0; i < 8; i++ {
         <-gen_chan
      }
      close(data_result_chan)
      fmt.Println("spend timeis ", time.Now().Unix()-time1)
   }()
   for date_result := range data_result_chan {
      fmt.Println(date_result)
   }
}

有疑问加站长微信联系(非本文作者)

eUjI7rn.png!mobile

About Joyk


Aggregate valuable and interesting links.
Joyk means Joy of geeK