Go協(xié)程和channel的實(shí)現(xiàn)

更新時間：2026年01月28日 11:26:22 作者：JetTsang

本文主要介紹了GO利用channel協(xié)調(diào)協(xié)程的實(shí)現(xiàn),文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

一：?? 協(xié)程

Goroutine 是 Go 運(yùn)行時管理的輕量級線程

在 go 中，開啟一個協(xié)程是非常簡單的

package main

import (
  "fmt"
  "time"
)

func sing() {
  fmt.Println("唱歌")
  time.Sleep(1 * time.Second)
  fmt.Println("唱歌結(jié)束")
}

func main() {
  go sing()
  go sing()
  go sing()
  go sing()
  time.Sleep(2 * time.Second)
}

如果我把這個主線程中的延時去掉之后，你會發(fā)現(xiàn)程序沒有任何輸出就結(jié)束了

這是為什么呢

那是因?yàn)橹骶€程結(jié)束協(xié)程自動結(jié)束，主線程不會等待協(xié)程的結(jié)束

?? WaitGroup

我們只需要讓主線程等待協(xié)程就可以了，它的用法是這樣的

package main

import (
  "fmt"
  "sync"
  "time"
)

var (
  wait = sync.WaitGroup{}
)

func sing() {
  fmt.Println("唱歌")
  time.Sleep(1 * time.Second)
  fmt.Println("唱歌結(jié)束")
  wait.Done()
}

func main() {
  wait.Add(4)
  go sing()
  go sing()
  go sing()
  go sing()
  wait.Wait()
  fmt.Println("主線程結(jié)束")
}

二：?? channel

有沒有想過一個問題，我在協(xié)程里面產(chǎn)生了數(shù)據(jù)，咋傳遞給主線程呢？

或者是怎么傳遞給其他協(xié)程函數(shù)呢？

這個時候 channel 來了

基本定義

package main

import "fmt"

func main() {
  var c chan int // 聲明一個傳遞整形的通道
  // 初始化通道
  c = make(chan int, 1) //  初始化一個 有一個緩沖位的通道
  c <- 1
  //c <- 2 // 會報錯 deadlock
  fmt.Println(<-c) // 取值
  //fmt.Println(<-c) // 再取也會報錯  deadlock

  c <- 2
  n, ok := <-c
  fmt.Println(n, ok)
  close(c) // 關(guān)閉協(xié)程
  c <- 3   // 關(guān)閉之后就不能再寫或讀了  send on closed channel
  fmt.Println(c)
}

當(dāng)然，在同步模式下，channel 沒有任何意義

需要在異步模式下使用 channel，在協(xié)程函數(shù)里面寫，在主線程里面接收數(shù)據(jù)

package main

import (
  "fmt"
  "sync"
  "time"
)

var moneyChan = make(chan int) // 聲明并初始化一個長度為0的信道

func pay(name string, money int, wait *sync.WaitGroup) {
  fmt.Printf("%s 開始購物\n", name)
  time.Sleep(1 * time.Second)
  fmt.Printf("%s 購物結(jié)束\n", name)

  moneyChan <- money

  wait.Done()
}

// 協(xié)程
func main() {
  var wait sync.WaitGroup
  startTime := time.Now()
  // 現(xiàn)在的模式，就是購物接力
  //shopping("張三")
  //shopping("王五")
  //shopping("李四")
  wait.Add(3)
  // 主線程結(jié)束，協(xié)程函數(shù)跟著結(jié)束
  go pay("張三", 2, &wait)
  go pay("王五", 3, &wait)
  go pay("李四", 5, &wait)

  go func() {
    defer close(moneyChan)
    // 在協(xié)程函數(shù)里面等待上面三個協(xié)程函數(shù)結(jié)束
    wait.Wait()
  }()

  for {
    money, ok := <-moneyChan
    fmt.Println(money, ok)
    if !ok {
      break
    }
  }

  //time.Sleep(2 * time.Second)

  fmt.Println("購買完成", time.Since(startTime))
  fmt.Println("moneyList", moneyList)
}

如果這樣接收數(shù)據(jù)不太優(yōu)雅，那還有更優(yōu)雅的寫法

  for money := range moneyChan {
    moneyList = append(moneyList, money)
  }

如果通道被 close，for 循環(huán)會自己結(jié)束，十分優(yōu)雅

?? select

如果一個協(xié)程函數(shù)，往多個 channel 里面寫東西，在主線程里面怎么拿數(shù)據(jù)呢？

go 為我們提供了 select，用于異步的從多個 channel 里面去取數(shù)據(jù)

package main

import (
	"fmt"
	"sync"
	"time"
)

// 信道 存 int 類型
var moneyChan1 = make(chan int) // 聲明并初始化一個長度為0的信道
var nameChan1 = make(chan string)
var doneChan = make(chan struct{})

func send(name string, money int, wait *sync.WaitGroup) {
	fmt.Printf("%s 開始購物\n", name)
	time.Sleep(1 * time.Second)
	fmt.Printf("%s 購物結(jié)束\n", name)

	moneyChan1 <- money // 信道賦值語句
	nameChan1 <- name

	wait.Done()
}

func main() {
	var wait sync.WaitGroup

	startTime := time.Now()
	wait.Add(3)
	// 協(xié)程
	go send("zhangsan", 2, &wait)
	go send("lisi", 3, &wait)
	go send("wangwu", 5, &wait)

	// 再開一個協(xié)程函數(shù)判斷是否結(jié)束
	go func() {
		defer close(moneyChan1)
		defer close(nameChan1)
		defer close(doneChan)
		wait.Wait()
		// 再創(chuàng)建一個信道用于關(guān)閉
		// close(moneyChan)
	}()

	// 等價于下面的寫法
	var moneyList []int
	var nameList []string

	// 多個 channel 的寫法
	var event = func() {
		for {
			select {
			case money := <-moneyChan1:
				moneyList = append(moneyList, money)
			case name := <-nameChan1:
				nameList = append(nameList, name)
			case <-doneChan:
				return
			}
		}
	}
	event()

	fmt.Println("購買完成", time.Since(startTime))
	fmt.Println("moneyList", moneyList)
	fmt.Println("nameList", nameList)
}

?? 協(xié)程超時處理

package main

import (
  "fmt"
  "time"
)

var done = make(chan struct{})

func event() {
  fmt.Println("event執(zhí)行開始")
  time.Sleep(2 * time.Second)
  fmt.Println("event執(zhí)行結(jié)束")
  close(done)
}

func main() {
  go event()

  select {
  case <-done:
    fmt.Println("協(xié)程執(zhí)行完畢")
  case <-time.After(1 * time.Second):
    fmt.Println("超時")
    return
  }

}

三：?? 線程安全

什么是線程安全？

現(xiàn)在有兩個協(xié)程，同時觸發(fā)，一個協(xié)程對一個全局變量進(jìn)行 100 完成 ++ 操作，另一個對全局變量—的操作

那么，兩個協(xié)程結(jié)束，最后的值應(yīng)該是0才對

package main

import (
  "fmt"
  "sync"
)

var num int
var wait sync.WaitGroup

func add() {
  for i := 0; i < 1000000; i++ {
    num++
  }
  wait.Done()
}
func reduce() {
  for i := 0; i < 1000000; i++ {
    num--
  }
  wait.Done()
}

func main() {
  wait.Add(2)
  go add()
  go reduce()
  wait.Wait()
  fmt.Println(num)

}

但是你會發(fā)現(xiàn)，這個輸出的結(jié)果完全無法預(yù)測

這是為什么呢？

根本原因是 CPU 的調(diào)度方法為搶占式執(zhí)行，隨機(jī)調(diào)度

?? 同步鎖

那么我們能不能通過給操作加鎖來解決這個問題呢

答案是可以的

package main

import (
  "fmt"
  "sync"
)

var num int
var wait  sync.WaitGroup
var lock  sync.Mutex

func add() {
  // 誰先搶到了這把鎖，誰就把它鎖上，一旦鎖上，其他的線程就只能等著
  lock.Lock()
  for i := 0; i < 1000000; i++ {
    num++
  }
  lock.Unlock()
  wait.Done()
}
func reduce() {
  lock.Lock()
  for i := 0; i < 1000000; i++ {
    num--
  }
  lock.Unlock()
  wait.Done()
}

func main() {
  wait.Add(2)
  go add()
  go reduce()
  wait.Wait()
  fmt.Println(num)

}

?? 線程安全下的 map

如果我們在一個協(xié)程函數(shù)下，讀寫 map 就會引發(fā)一個錯誤

concurrent map read and map write

希望大家見到這個錯誤，就能知道，這個就是 map 的線程安全錯誤

package main

import (
  "fmt"
  "sync"
  "time"
)

var wait sync.WaitGroup
var mp = map[string]string{}

func reader() {
  for {
    fmt.Println(mp["time"])
  }
  wait.Done()
}
func writer() {
  for {
    mp["time"] = time.Now().Format("15:04:05")
  }
  wait.Done()
}

func main() {
  wait.Add(2)
  go reader()
  go writer()
  wait.Wait()

}

我們不能在并發(fā)模式下讀寫 map

如果要這樣做

給讀寫操作加鎖
使用sync.Map

加鎖

package main

import (
  "fmt"
  "sync"
  "time"
)

var wait sync.WaitGroup
var mp = map[string]string{}
var lock sync.Mutex

func reader() {
  for {
    lock.Lock()
    fmt.Println(mp["time"])
    lock.Unlock()
  }
  wait.Done()
}
func writer() {
  for {
    lock.Lock()
    mp["time"] = time.Now().Format("15:04:05")
    lock.Unlock()
  }
  wait.Done()
}

func main() {
  wait.Add(2)
  go reader()
  go writer()
  wait.Wait()
}

sync.Map

package main

import (
  "fmt"
  "sync"
  "time"
)

var wait sync.WaitGroup
var mp = sync.Map{}

func reader() {
  for {

    fmt.Println(mp.Load("time"))
  }
  wait.Done()
}
func writer() {
  for {
    mp.Store("time", time.Now().Format("15:04:05"))
  }
  wait.Done()
}

func main() {
  wait.Add(2)
  go reader()
  go writer()
  wait.Wait()

}

其實(shí)看它源碼，它的內(nèi)部也是用了同步鎖的

到此這篇關(guān)于Go協(xié)程和channel的實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)Go協(xié)程和channel內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: