2. 程式人生 > >Golang併發模型：輕鬆入門流水線FAN模式

Golang併發模型：輕鬆入門流水線FAN模式

阿新 • • 發佈：2019-01-05

前一篇文章《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，快去實踐一把。

完整示例程式碼

本文所有程式碼都在倉庫，可檢視完整示例程式碼：https://github.com/Shitaibin/golang_pipeline_step_by_step

併發系列文章推薦

下一篇，寫流水線中協程的“優雅退出”，歡迎關注。

  1. 如果這篇文章對你有幫助，不妨關注下我的Github，有文章會收到通知。
  2. 本文作者：大彬
  3. 如果喜歡本文，隨意轉載，但請保留此原文連結：http://lessisbetter.site/2018/11/28/golang-pipeline-fan-model/
關注公眾號，獲取最新Golang文章。 一起學Golang-分享有料的Go語言技術

相關推薦

Golang併發模型：輕鬆入門流水線FAN模式

前一篇文章《Golang併發模型：輕鬆入門流水線模型》，介紹了流水線模型的概念，這篇文章是流水線模型進階，介紹FAN-IN和FAN-OUT，FAN模式可以讓我們的流水線模型更好的利用Golang併發，提高軟體效能。但FAN模式不一定是萬能，不見得能提高程式的效能，甚至還不如普通的流水線。

Golang併發模型：輕鬆入門流水線模型

Golang作為一個實用主義的程式語言，非常注重效能，在語言特性上天然支援併發，它有多種併發模型，通過流水線模型系列文章，你會更好的使用Golang併發特性，提高你的程式效能。 這篇文章主要介紹流水線模型的流水線概念，後面文章介紹流水線模型的FAN-IN和FAN-OUT，最後介紹下如

Golang併發模型：併發協程的優雅退出

goroutine作為Golang併發的核心，我們不僅要關注它們的建立和管理，當然還要關注如何合理的退出這些協程，不（合理）退出不然可能會造成阻塞、panic、程式行為異常、資料結果不正確等問題。這篇文章介紹，如何合理的退出goroutine，減少軟體bug。 goroutine在退

《跟老齊學Python：輕鬆入門》學習筆記：章一：基礎物件型別

第1章 基本物件型別 浮點數的所有運算都要注意 1.1 數和四則運算 1.python可以自動處理大整數問題，不必擔心溢位 2.單精度浮點數（4個位元組）在計算機內部儲存方式： 1位為符號位（0為正，1為負），8位用作整數，最後的23位表示有效數字 3.每個物件在記憶

併發模型（二）——Master-Worker模式

    Master-Worker模式是常用的並行模式之一，它的核心思想是，系統有兩個程序協作工作：Master程序，負責接收和分配任務；Worker程序，負責處理子任務。當Worker程序將子任務處理完成後，結果返回給Master程序，由Master程序做歸納彙總，最後得到最

java併發程式設計：如何實現生產者消費者模式？

package concurrent; import java.util.LinkedList; import java.util.Queue; public class ProducerConsumer { public static class Storage{

Golang從入門到精通（十八）：Golang併發程式設計之Goroutine

程序，執行緒，並行和併發 一個應用程式是執行在機器上的一個程序；程序是一個執行在自己記憶體地址空間裡的獨立執行體。一個程序由一個或多個作業系統執行緒組成，這些執行緒其實是共享同一個記憶體地址空間的一起工作的執行體。幾乎所有’正式’的程式都是多執行緒的，以便讓使

Golang從入門到精通（十九）：Golang併發程式設計之Channel

Go語言併發模型 Go 語言中使用了CSP模型來進行執行緒通訊，準確說，是輕量級執行緒goroutine之間的通訊。CSP模型和Actor模型類似，也是由獨立的，併發執行的實體所構成，實體之間也是通過傳送訊息進行通訊的。 Actor模型和CSP模型區別 A

Golang 入門系列（十七）幾個常見的併發模型——生產者消費者模型

前面已經講過很多Golang系列知識，包括併發，鎖等內容，感興趣的可以看看以前的文章，https://www.cnblogs.com/zhangweizhong/category/1275863.html， 接下來講幾個golang常見的併發模型，今天先說說生產者消費者模型。   生產者消費者模型

（2.1.27.2）Java併發程式設計：JAVA的記憶體模型

Java定義了自身的記憶體模型是為了遮蔽掉不同硬體和作業系統的記憶體模型差異 Java為了處理記憶體的不可見性與重排序的問題，定義了Happens-Before 原則 Happens-Before 原則的理解：對於兩個操作A和B（這兩個操作可以在不同的執行

Java併發程式設計裡的volatile。Java記憶體模型核CPU記憶體架構的對應關係 Java併發程式設計：volatile關鍵字解析

  CPU記憶體架構：https://www.jianshu.com/p/3d1eb589b48e Java記憶體模型：https://www.jianshu.com/p/27a9003c33f4 多執行緒下的快取一致性問題：https://www.jianshu.com/p/97dc5242

Golang非CSP併發模型外的其他並行方法總結

Golang最為讓人熟知的併發模型當屬CSP併發模型，也就是由goroutine和channel構成的GMP併發模型，具體內容不在贅述了，可以翻回之前的文章檢視。在這裡，要講講Golang的其他併發方式。 Golang不僅可以使用CSP併發模式，還可以使用傳統的共享資料的併發模式。 臨界區(critica

Spark：Spark 程式設計模型及快速入門

Spark程式設計模型SparkContext類和SparkConf類程式碼中初始化我們可通過如下方式呼叫 SparkContext 的簡單建構函式,以預設的引數值來建立相應的物件。val sc = new SparkContext("local[4]", "Test Spa

漫談併發程式設計：Actor模型

0x00 前言 一般來說有兩種策略用來在併發執行緒中進行通訊：共享資料和訊息傳遞。熟悉c和java併發程式設計的都會比較熟悉共享資料的策略，比如java程式設計師就會常用到java.util.concurrent包中同步、鎖相關的資料結構。 使用共享資料方式

Java併發程式設計：JMM (Java記憶體模型) 以及與volatile關鍵字詳解

計算機系統的一致性 在現代計算機作業系統中，多工處理幾乎是一項必備的功能，因為嵌入了多核處理器，計算機系統真正做到了同一時間執行若干個任務，是名副其實的多核系統。在多核系統中，為了提升CPU與記憶體的互動效率，一般都設定了一層 “快取記憶體區” 作為記憶體與處理器之間的緩衝，使得CPU在運算的過程中直接從快

Python之路(第三十八篇) 併發程式設計：程序同步鎖/互斥鎖、訊號量、事件、佇列、生產者消費者模型

一、程序鎖（同步鎖/互斥鎖） 程序之間資料不共享,但是共享同一套檔案系統,所以訪問同一個檔案,或同一個列印終端,是沒有問題的, 而共享帶來的是競爭，競爭帶來的結果就是錯亂，如何控制，就是加鎖處理。 例子 #併發執行,效率高,但競爭同一列印終端,帶來了列印錯亂 from multiproc

輕鬆入門Golang pprof實用不忽悠

網上已搜尋golang pprof，資料不少，簡明高效的一個沒看到，這篇文章5步教你用會pprof獲取cpu和記憶體prof。 第1步：安裝易用的pprof golang自帶的prof包是runtime/pprof，這個是低級別的，需要你手動做一些設定等等周邊工作，不利於我們快速

【無私分享：從入門到精通ASP.NET MVC】從0開始，一起搭框架、做專案（2）建立資料庫和資料模型

索引 簡述 我們建立一個數據庫然後生成資料模型，新增ADO.NET實體資料模型，這裡呢常用的有兩種方式，一種是DbFirst，一種是CodeFirst，為了簡便，我們使用DbFirst。有基礎的朋友可以使用CodeFirst，對專案沒有影響的。 專案準備 我們用的工具是：VS 2013 + Sql

Golang 學習一：開發環境搭建-入門必備

一、下載適合本機的Golang版本注意對應好自己PC 的system. Windows 直接下載對應的msi安裝即可，Linux注意發行版本。 本文Linux選擇go1.8.1.linux-amd64.tar.gz版本。二、Golang安裝1.windows安裝的過程比較簡單

GOLANG的context和併發模型

GOLANG1.7新增了context，最初這個package是在golang.org/x/net/context中的，後來證實對很多程式都是必須的，就納入了標準庫。 對context的介紹是在context，讀這個blog之前，要先讀pipelines（pi