goroutine 和 channel

goroutine-看一個需求
需求：要求統計 1-9000000000 的數字中，哪些是素數？

分析思路：

傳統的方法，就是使用一個回圈，回圈的判斷各個數是不是素數，[很慢]
使用并發或者并行的方式，將統計素數的任務分配給多個 goroutine 去完成，這時就會使用到goroutine.【速度提高 4 倍】

goroutine-基本介紹

行程和執行緒介紹

程式、行程和執行緒的關系

并發和并行

多執行緒程式在單核上運行，就是并發
多執行緒程式在多核上運行，就是并行

Go 協程和 Go 主執行緒

Go 主執行緒(有程式員直接稱為執行緒/也可以理解成行程): 一個 Go 執行緒上，可以起多個協程，你可以這樣理解，協程是輕量級的執行緒[編譯器做優化]，

Go 協程的特點

有獨立的堆疊空間
共享程式堆空間
調度由用戶控制
協程是輕量級的執行緒

goroutine-快速入門

案例說明
請撰寫一個程式，完成如下功能:

在主執行緒(可以理解成行程)中，開啟一個 goroutine, 該協程每隔 1 秒輸出 "hello,world"
在主執行緒中也每隔一秒輸出"hello,golang", 輸出 10 次后，退出程式
要求主執行緒和 goroutine 同時執行.

package main
import (
	"fmt"
	"strconv"
	"time"
)

// 在主執行緒(可以理解成行程)中，開啟一個goroutine, 該協程每隔1秒輸出 "hello,world"
// 在主執行緒中也每隔一秒輸出"hello,golang", 輸出10次后，退出程式
// 要求主執行緒和goroutine同時執行

//撰寫一個函式，每隔1秒輸出 "hello,world"
func test() {
	for i := 1; i <= 10; i++ {
		fmt.Println("tesst () hello,world " + strconv.Itoa(i))
		time.Sleep(time.Second)
	}
}

func main() {

	go test() // 開啟了一個協程

	for i := 1; i <= 10; i++ {
		fmt.Println(" main() hello,golang" + strconv.Itoa(i))
		time.Sleep(time.Second)
	}
}

快速入門小結

主執行緒是一個物理執行緒，直接作用在 cpu 上的，是重量級的，非常耗費 cpu 資源，
協程從主執行緒開啟的，是輕量級的執行緒，是邏輯態，對資源消耗相對小，
Golang 的協程機制是重要的特點，可以輕松的開啟上萬個協程，其它編程語言的并發機制是一般基于執行緒的，開啟過多的執行緒，資源耗費大，這里就突顯 Golang 在并發上的優勢了

goroutine 的調度模型

MPG 模式基本介紹

解釋一下MPG含義：
M（Machine）：作業系統的主執行緒
P（Processor）：協程執行需要的資源（背景關系context），可以看作一個區域的調度器，使go代碼在一個執行緒上跑，他是實作從N：1到N：M映射的關鍵
G（Gorountine）：協程，有自己的堆疊，包含指令指標（instruction pointer）和其它資訊（正在等待的channel等等），用于調度，一個P下面可以有多個G

MPG 模式運行的狀態一

P的數量可以通過GOMAXPROCS()來設定，他其實代表了真正的并發度，即有多少個goroutine可以同時運行，P同時也維護著G（協程）的佇列（稱之為runqueue行程佇列），Go代碼中的M每有一個陳述句被執行，P就在末尾加入一個G（從runqueue佇列中取出來的），在下一個調度點（P），就從runqueue佇列中取出G，
P可以在OS執行緒（主執行緒，或者是M）被阻塞時，轉到另一個OS執行緒（M）！Go中的調度器保證有足夠的執行緒來運行所有的P，當啟用一個M0中的G0被sysCall（系統呼叫）的時候，M0下面的P轉給另一個執行緒M1（可以是創建的，也可以是原本就存在的），M1接受了P（包括P所帶的runqueue的佇列里面所有狀態的G，但不包括已經被syscall的G0），繼續運行，而M0會等待執行syscall的G0的回傳值，當G0的syscall結束后，他的主執行緒M0會嘗試取得一個P來運行G0，一般情況下，他會從其他的M里面偷一個P過來，如果沒有偷到的話就會把G0放到一個Global runqueue（全域行程佇列）中，然后把自己（M0）放進執行緒池或者轉為休眠狀態，

設定 Golang 運行的 cpu 數

介紹：為了充分了利用多 cpu 的優勢，在 Golang 程式中，設定運行的 cpu 數目

package main
import (
	"runtime"
	"fmt"
)

func main() {
	cpuNum := runtime.NumCPU()
	fmt.Println("cpuNum=", cpuNum)

	//可以自己設定使用多個cpu
	runtime.GOMAXPROCS(cpuNum - 1)
	fmt.Println("ok")
}

channel(管道)-看個需求

需求：現在要計算 1-200 的各個數的階乘，并且把各個數的階乘放入到 map 中，最后顯示出來，
要求使用 goroutine 完成

分析思路：

使用 goroutine 來完成，效率高，但是會出現并發/并行安全問題.
這里就提出了不同 goroutine 如何通信的問題

代碼實作

使用 goroutine 來完成(看看使用 gorotine 并發完成會出現什么問題? 然后去解決)
在運行某個程式時，如何知道是否存在資源競爭問題，方法很簡單，在編譯該程式時，增加一個引數 -race 即可

不同 goroutine 之間如何通訊

全域變數的互斥鎖
使用管道 channel 來解決

使用全域變數加鎖同步改行程式

因為沒有對全域變數 m 加鎖，因此會出現資源爭奪問題，代碼會出現錯誤，提示 concurrent map
writes
解決方案：加入互斥鎖
我們的數的階乘很大，結果會越界,可以將求階乘改成 sum += uint64(i)

package main
import (
	"fmt"
	_ "time"
	"sync"
)

// 需求：現在要計算 1-200 的各個數的階乘，并且把各個數的階乘放入到map中，
// 最后顯示出來，要求使用goroutine完成 

// 思路
// 1. 撰寫一個函式，來計算各個數的階乘，并放入到 map中.
// 2. 我們啟動的協程多個，統計的將結果放入到 map中
// 3. map 應該做出一個全域的.

var (
	myMap = make(map[int]int, 10)  
	//宣告一個全域的互斥鎖
	//lock 是一個全域的互斥鎖， 
	//sync 是包: synchornized 同步
	//Mutex : 是互斥
	lock sync.Mutex
)

// test 函式就是計算 n!, 讓將這個結果放入到 myMap
func test(n int) {
	
	res := 1
	for i := 1; i <= n; i++ {
		res *= i
	}

	//這里我們將 res 放入到myMap
	//加鎖
	lock.Lock()
	myMap[n] = res //concurrent map writes?
	//解鎖
	lock.Unlock()
}

func main() {

	// 我們這里開啟多個協程完成這個任務[200個]
	for i := 1; i <= 20; i++ {
		go test(i)
	}


	//休眠10秒鐘【第二個問題 】
	//time.Sleep(time.Second * 5)

	//這里我們輸出結果,變數這個結果
	lock.Lock()
	for i, v := range myMap {
		fmt.Printf("map[%d]=%d\n", i, v)
	}
	lock.Unlock()

}

為什么需要 channel

前面使用全域變數加鎖同步來解決 goroutine 的通訊，但不完美
主執行緒在等待所有 goroutine 全部完成的時間很難確定，我們這里設定 10 秒，僅僅是估算，
如果主執行緒休眠時間長了，會加長等待時間，如果等待時間短了，可能還有 goroutine 處于作業
狀態，這時也會隨主執行緒的退出而銷毀
通過全域變數加鎖同步來實作通訊，也并不利用多個協程對全域變數的讀寫操作，
上面種種分析都在呼喚一個新的通訊機制-channel

channel 的基本介紹

channle 本質就是一個資料結構-佇列
資料是先進先出【FIFO : first in first out】
執行緒安全，多 goroutine 訪問時，不需要加鎖，就是說 channel 本身就是執行緒安全的
channel 有型別的，一個 string 的 channel 只能存放 string 型別資料，

定義/宣告 channel

var 變數名 chan 資料型別
舉例：

var intChan chan int (intChan 用于存放 int 資料)
var mapChan chan map[int]string (mapChan 用于存放 map[int]string 型別)
var perChan chan Person
var perChan2 chan *Person
...

說明
channel 是參考型別
channel 必須初始化才能寫入資料, 即 make 后才能使用
管道是有型別的，intChan 只能寫入整數 int

package main
import (
	"fmt"
)

func main() {

	//演示一下管道的使用
	//1. 創建一個可以存放3個int型別的管道
	var intChan chan int
	intChan = make(chan int, 3)

	//2. 看看intChan是什么
	fmt.Printf("intChan 的值=%v intChan本身的地址=%p\n", intChan, &intChan)


	//3. 向管道寫入資料
	intChan<- 10
	num := 211
	intChan<- num
	intChan<- 50
	// //如果從channel取出資料后，可以繼續放入
	<-intChan
	intChan<- 98//注意點, 當我們給管寫入資料時，不能超過其容量


	//4. 看看管道的長度和cap(容量)
	fmt.Printf("channel len= %v cap=%v \n", len(intChan), cap(intChan)) // 3, 3

	//5. 從管道中讀取資料

	var num2 int
	num2 = <-intChan 
	fmt.Println("num2=", num2)
	fmt.Printf("channel len= %v cap=%v \n", len(intChan), cap(intChan))  // 2, 3

	//6. 在沒有使用協程的情況下，如果我們的管道資料已經全部取出，再取就會報告 deadlock

	num3 := <-intChan
	num4 := <-intChan

	//num5 := <-intChan

	fmt.Println("num3=", num3, "num4=", num4/*, "num5=", num5*/)

}

管道的初始化，寫入資料到管道，從管道讀取資料及基本的注意事項

channel 使用的注意事項

channel 中只能存放指定的資料型別
channle 的資料放滿后，就不能再放入了
如果從 channel 取出資料后，可以繼續放入
在沒有使用協程的情況下，如果 channel 資料取完了，再取，就會報 dead lock

讀寫 channel 案例演示

package main
import (
	"fmt"
)

type Cat struct {
	Name string
	Age int
}

func main() {

	//定義一個存放任意資料型別的管道 3個資料
	//var allChan chan interface{}
	allChan := make(chan interface{}, 3)

	allChan<- 10
	allChan<- "tom jack"
	cat := Cat{"小花貓", 4}
	allChan<- cat

	//我們希望獲得到管道中的第三個元素，則先將前2個推出
	<-allChan
	<-allChan

	newCat := <-allChan //從管道中取出的Cat是什么?

	fmt.Printf("newCat=%T , newCat=%v\n", newCat, newCat)
	//下面的寫法是錯誤的!編譯不通過
	//fmt.Printf("newCat.Name=%v", newCat.Name)
	//使用型別斷言
	a := newCat.(Cat) 
	fmt.Printf("newCat.Name=%v", a.Name)
	

}

channel 的遍歷和關閉

channel 的關閉

使用內置函式 close 可以關閉 channel, 當 channel 關閉后，就不能再向 channel 寫資料了，但是仍然可以從該 channel 讀取資料

channel 的遍歷

channel 支持 for--range 的方式進行遍歷，請注意兩個細節

在遍歷時，如果 channel 沒有關閉，則回出現 deadlock 的錯誤
在遍歷時，如果 channel 已經關閉，則會正常遍歷資料，遍歷完后，就會退出遍歷，

channel 遍歷和關閉的案例演示

package main
import (
   "fmt"
)

func main() {

   intChan := make(chan int, 3)
   intChan<- 100
   intChan<- 200
   close(intChan) // close
   //這是不能夠再寫入數到channel
   //intChan<- 300
   fmt.Println("okook~")
   //當管道關閉后，讀取資料是可以的
   n1 := <-intChan
   fmt.Println("n1=", n1)


   //遍歷管道
   intChan2 := make(chan int, 100)
   for i := 0; i < 100; i++ {
   	intChan2<- i * 2  //放入100個資料到管道
   }

   //遍歷管道不能使用普通的 for 回圈
   // for i := 0; i < len(intChan2); i++ {

   // }
   //在遍歷時，如果channel沒有關閉，則會出現deadlock的錯誤
   //在遍歷時，如果channel已經關閉，則會正常遍歷資料，遍歷完后，就會退出遍歷
   close(intChan2)
   for v := range intChan2 {
   	fmt.Println("v=", v)
   }


}

應用示例--channel與goroutine

package main
import (
	"fmt"
	"time"
)


//write Data
func writeData(intChan chan int) {
	for i := 1; i <= 50; i++ {
		//放入資料
		intChan<- i //
		fmt.Println("writeData ", i)
		//time.Sleep(time.Second)
	}
	close(intChan) //關閉
}

//read data
func readData(intChan chan int, exitChan chan bool) {

	for {
		v, ok := <-intChan
		if !ok {
			break
		}
		time.Sleep(time.Second)
		fmt.Printf("readData 讀到資料=%v\n", v) 
	}
	//readData 讀取完資料后，即任務完成
	exitChan<- true
	close(exitChan)

}

func main() {

	//創建兩個管道
	intChan := make(chan int, 10)
	exitChan := make(chan bool, 1)
	
	go writeData(intChan)
	go readData(intChan, exitChan)

	time.Sleep(time.Second * 10)
	for {
		_, ok := <-exitChan
		if !ok {
			break
		}
	}

}

應用實體 2-阻塞

若上面的代碼，注釋掉go readData(intChan, exitChan)，會怎樣，因為管道有長度，所以當編譯器發現一個管道只有寫而沒有讀，改管道會阻塞（讀與寫的頻率不一致沒關系）！

應用實體 3

需求：
要求統計 1-200000 的數字中，哪些是素數？這個問題在本章開篇就提出了，現在我們有 goroutine和 channel 的知識后，就可以完成了 [測驗資料: 80000]

分析思路：

傳統的方法，就是使用一個回圈，回圈的判斷各個數是不是素數【ok】，
使用并發/并行的方式，將統計素數的任務分配給多個(4 個)goroutine 去完成，完成任務時間短，

傳統方法，一個協程

package main
import (
	"time"
	"fmt"
)

func main() {

		start := time.Now().Unix()
		for num := 1; num <= 80000; num++ {

			flag := true //假設是素數
			//判斷num是不是素數
			for i := 2; i < num; i++ {
				if num % i == 0 {//說明該num不是素數
					flag = false
					break
				}
			}

			if flag {
				//將這個數就放入到primeChan
				//primeChan<- num
			}

		}
		end := time.Now().Unix()
		fmt.Println("普通的方法耗時=", end - start)
		
}

開了四個協程

package main
import (
	"fmt"
	"time"
)



//向 intChan放入 1-8000個數
func putNum(intChan chan int) {

	for i := 1; i <= 80000; i++ {    
		intChan<- i
	}

	//關閉intChan
	close(intChan)
}

// 從 intChan取出資料，并判斷是否為素數,如果是，就
// 	//放入到primeChan
func primeNum(intChan chan int, primeChan chan int, exitChan chan bool) {

	//使用for 回圈
	// var num int
	var flag bool // 
	for {
		//time.Sleep(time.Millisecond * 10)
		num, ok := <-intChan //intChan 取不到..
		
		if !ok { 
			break
		}
		flag = true //假設是素數
		//判斷num是不是素數
		for i := 2; i < num; i++ {
			if num % i == 0 {//說明該num不是素數
				flag = false
				break
			}
		}

		if flag {
			//將這個數就放入到primeChan
			primeChan<- num
		}
	}

	fmt.Println("有一個primeNum 協程因為取不到資料，退出")
	//這里我們還不能關閉 primeChan
	//向 exitChan 寫入true
	exitChan<- true	

}

func main() {

	intChan := make(chan int , 1000)
	primeChan := make(chan int, 20000)//放入結果
	//標識退出的管道
	exitChan := make(chan bool, 8) // 4個



	start := time.Now().Unix()
	
	//開啟一個協程，向 intChan放入 1-8000個數
	go putNum(intChan)
	//開啟4個協程，從 intChan取出資料，并判斷是否為素數,如果是，就
	//放入到primeChan
	for i := 0; i < 8; i++ {
		go primeNum(intChan, primeChan, exitChan)
	}

	//這里我們主執行緒，進行處理
	//直接
	go func(){
		for i := 0; i < 8; i++ {
			<-exitChan
		}

		end := time.Now().Unix()
		fmt.Println("使用協程耗時=", end - start)

		//當我們從exitChan 取出了4個結果，就可以放心的關閉 prprimeChan
		close(primeChan)
	}()


	//遍歷我們的 primeChan ,把結果取出
	for {
		_, ok := <-primeChan
		if !ok{
			break
		}
		//將結果輸出
		//fmt.Printf("素數=%d\n", res)
	}

	fmt.Println("main執行緒退出")


	
}

結論：使用 go 協程后，執行的速度，理論上比普通方法提高至少 4 倍(我這是兩倍)

channel 使用細節和注意事項

channel 可以宣告為只讀，或者只寫性質【案例演示】

package main
import (
	"fmt"
)

func main() {
	//管道可以宣告為只讀或者只寫

	//1. 在默認情況下下，管道是雙向
	//var chan1 chan int //可讀可寫
	
	//2 宣告為只寫
	var chan2 chan<- int
	chan2 = make(chan int, 3)
	chan2<- 20
	//num := <-chan2 //error

	fmt.Println("chan2=", chan2)

	//3. 宣告為只讀
	var chan3 <-chan int
	num2 := <-chan3
	//chan3<- 30 //err
	fmt.Println("num2", num2)

}

使用 select 可以解決從管道取資料的阻塞問題

package main
import (
	"fmt"
	"time"
)

func main() {

	//使用select可以解決從管道取資料的阻塞問題

	//1.定義一個管道 10個資料int
	intChan := make(chan int, 10)
	for i := 0; i < 10; i++ {
		intChan<- i
	}
	//2.定義一個管道 5個資料string
	stringChan := make(chan string, 5)
	for i := 0; i < 5; i++ {
		stringChan <- "hello" + fmt.Sprintf("%d", i)
	}

	//傳統的方法在遍歷管道時，如果不關倍訓阻塞而導致 deadlock

	//問題，在實際開發中，可能我們不好確定什么關閉該管道.
	//可以使用select 方式可以解決
	//label:
	for {
		select {
			//注意: 這里，如果intChan一直沒有關閉，不會一直阻塞而deadlock
			//，會自動到下一個case匹配
			case v := <-intChan : 
				fmt.Printf("從intChan讀取的資料%d\n", v)
				time.Sleep(time.Second)
			case v := <-stringChan :
				fmt.Printf("從stringChan讀取的資料%s\n", v)
				time.Sleep(time.Second)
			default :
				fmt.Printf("都取不到了，不玩了, 程式員可以加入邏輯\n")
				time.Sleep(time.Second)
				return 
				//break label
		}
	}
}

goroutine 中使用 recover，解決協程中出現 panic，導致程式崩潰問題

如果我們開了一個協程，但這個協程出現panic，就會導致整個程式崩潰，這時我們可以在goroutine中使用recover來捕獲panic，這樣及時協程發生問題，主執行緒依然不受影響

package main
import (
	"fmt"
	"time"
)

//函式
func sayHello() {
	for i := 0; i < 10; i++ {
		time.Sleep(time.Second)
		fmt.Println("hello,world")
	}
}
//函式
func test() {
	//這里我們可以使用defer + recover
	defer func() {
		//捕獲test拋出的panic
		if err := recover(); err != nil {
			fmt.Println("test() 發生錯誤", err)
		}
	}()
	//定義了一個map
	var myMap map[int]string
	myMap[0] = "golang" //error
}

func main() {

	go sayHello()
	go test()


	for i := 0; i < 10; i++ {
		fmt.Println("main() ok=", i)
		time.Sleep(time.Second)
	}

}

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/61800.html

標籤：Go

上一篇：求助，idea上java專案問題

下一篇：[Go] gocron原始碼閱讀-判斷是否使用root用戶執行

go-goroutine 和 channel

goroutine 和 channel

goroutine-基本介紹

并發和并行

Go 協程和 Go 主執行緒

goroutine-快速入門

快速入門小結

goroutine 的調度模型

MPG 模式基本介紹

MPG 模式運行的狀態一

設定 Golang 運行的 cpu 數

channel(管道)-看個需求

不同 goroutine 之間如何通訊

使用全域變數加鎖同步改行程式

為什么需要 channel

channel 的基本介紹

定義/宣告 channel

管道的初始化，寫入資料到管道，從管道讀取資料及基本的注意事項

channel 使用的注意事項

讀寫 channel 案例演示

channel 的遍歷和關閉

channel 的關閉

channel 的遍歷

應用示例--channel與goroutine

應用實體 2-阻塞

應用實體 3

channel 使用細節和注意事項