Golang goroutine channel 实现并发和并行
作者：cocoCola91667
本文手把手教你从零掌握goroutine+channel实战,涵盖并发编程、线程调度和协程通信核心技巧,点击立即提升你的Go并发能力,具有一定的参考价值,感兴趣的可以了解一下
一、为什么要使用goroutine
需求：统计1-10000000的数字中哪些是素数，并打印这些素数。
素数：除了1和它本身，不能被其他数整除的数。
实现方法：
- 传统方法：通过一个for循环判断各个数是不是素数
- 使用并发/并行方式：将统计素数的任务分配给多个goroutine完成
- goroutine结合channel
二、进程、线程以及并行、并发
1. 关于进程和线程
进程（Process）：程序在操作系统中的一次执行过程，是系统进行资源分配和调度的基本单位。每个进程都有自己的地址空间。一个进程至少有5种基本状态：初始态、执行态、等待状态、就绪状态、终止状态。
通俗地讲，进程就是一个正在执行的程序。
线程：进程的一个执行实例，是程序执行的最小单元，是比进程更小的能独立运行的基本单位。
- 一个进程可以创建多个线程
- 同一个进程中的多个线程可以并发执行
- 一个程序至少有一个进程
2. 关于并行和并发
| 概念 | 说明 |
|---|---|
| 并发 | 多个线程同时竞争一个位置，每个时间段只有一个线程在执行 |
| 并行 | 多个线程可以同时执行，每个时间段可以有多个线程同时执行 |
通俗地讲：
- 多线程程序在单核CPU上运行 → 并发
- 多线程程序在多核CPU上运行 → 并行
- 线程数 > CPU核数 → 既有并行又有并发
三、Golang中的协程（goroutine）以及主线程
golang 中的主线程：（可以理解为线程/也可以理解为进程），在一个 Golang 程序的主线程上可以起多个协程。Golang 中多协程可以实现并行或者并发。
协程：可以理解为用户级线程，这是对内核透明的，也就是系统并不知道有协程的存在， 是完全由用户自己的程序进行调度的。Golang 的一大特色就是从语言层面原生支持协程，在函数或者方法前面加go关键字就可创建一个协程。可以说Golang中的协程就是goroutine 。
Golang 中的多协程有点类似其他语言中的多线程。
多协程 vs 多线程
| 对比项 | goroutine（协程） | OS线程(操作系统线程） |
|---|---|---|
| 默认占用内存 | ~2KB | ~2MB |
| 切换调度开销 | 小 | 大 |
这也是越来越多大公司使用Golang的原因之一。
四、Goroutine的使用以及sync.WaitGroup
需求
在主线程中开启一个goroutine，每隔50毫秒输出"你好golang"，主线程也每隔50毫秒输出"你好golang"，输出10次后退出程序，要求主线程和goroutine同时执行。
基本用法（存在的问题）
package main
import (
"fmt"
"strconv"
"time"
)
func test() {
for i := 1; i <= 10; i++ {
fmt.Println("test() hello,world " + strconv.Itoa(i))
time.Sleep(time.Millisecond * 10)
}
}
func main() {
go test() // 开启一个协程
for i := 1; i <= 10; i++ {
fmt.Println(" main() hello,golang" + strconv.Itoa(i))
time.Sleep(time.Millisecond)
}
}
注意：主线程执行完毕后，即使协程没有执行完毕，程序也会退出。
使用 sync.WaitGroup 解决
package main
import (
"fmt"
"strconv"
"sync"
"time"
)
// 定义一个全局变量，类型是sync.WaitGroup
var wg sync.WaitGroup
func test() {
for i := 1; i <= 10; i++ {
fmt.Println("test() hello,world " + strconv.Itoa(i))
time.Sleep(time.Millisecond * 50)
}
wg.Done() // 协程计数器-1
}
func main() {
wg.Add(1) // 协程计数器+1
go test() // 开启一个协程
for i := 1; i <= 2; i++ {
fmt.Println(" main() 你好golang" + strconv.Itoa(i))
time.Sleep(time.Millisecond * 50)
}
wg.Wait() // 等待所有登记的goroutine都结束
fmt.Println("主线程退出...")
}
五、启动多个Goroutine
package main
import (
"fmt"
"sync"
)
var wg sync.WaitGroup
func hello(i int) {
defer wg.Done() // goroutine结束就登记-1
fmt.Println("Hello Goroutine!", i)
}
func main() {
for i := 0; i < 10; i++ {
wg.Add(1) // 启动一个goroutine就登记+1
go hello(i)
}
wg.Wait() // 等待所有登记的goroutine都结束
}
多次执行上面的代码，每次打印的顺序都不一致，因为10个goroutine是并发执行的，调度是随机的。
六、设置Golang并行运行的时候占用的CPU数量
Go 运行时的调度器使用 GOMAXPROCS 参数来确定需要使用多少个 OS 线程来同时执行 Go代码。默认值是机器上的 CPU 核心数。 例如在一个 8 核心的机器上，调度器会把 Go 代码同时调度到 8 个 OS 线程上。
Go 语言中可以通过 runtime.GOMAXPROCS()函数设置当前程序并发时占用的 CPU 逻辑核心数。
Go1.5 版本之前，默认使用的是单核心执行。Go1.5 版本之后，默认使用全部的 CPU 逻辑核心数。
package main
import (
"fmt"
"runtime"
)
func main() {
// 获取当前计算机上的CPU个数
cpuNum := runtime.NumCPU()
fmt.Println("cpuNum=", cpuNum)
// 可以自己设置使用多个CPU
runtime.GOMAXPROCS(cpuNum - 1)
}
七、Goroutine统计素数
package main
import (
"fmt"
"time"
)
// 需求：要统计 1-120000 的数字中那些是素数？for循环实现
// 素数：除了1和它本身，不能被其他数整除的数
// 为什么从 2 开始，到 num-1 结束？
// 因为一个数如果能被整除，除数不可能等于 1（1 不算），也不可能等于它自己（自己当然能被自己整除，但这是废话）。
// 比如判断 7 是不是素数，只需要试 2、3、4、5、6 能不能整除 7。
func main() {
start := time.Now().Unix()
for num := 2; num < 120000; num++ {
var flag = true
for i := 2; i < num; i++ {
if num%i == 0 {
flag = false
break
}
}
if flag {
fmt.Println(num, "是素数")
}
}
end := time.Now().Unix()
fmt.Println(end - start) // 11毫秒
}
使用4个goroutine并行统计素数：
package main
import (
"fmt"
"sync"
"time"
)
/*
1 协程 统计 1-30000
2 协程 统计 30001-60000
3 协程 统计 60001-90000
4 协程 统计 90001-120000
start:(n-1)*30000+1 end:n*30000
*/
var wg sync.WaitGroup
func test(n int) {
for num := (n-1)*30000 + 1; num < n*30000; num++ {
if num > 1 {
var flag = true
for i := 2; i < num; i++ {
if num%i == 0 {
flag = false
break
}
}
if flag {
// fmt.Println(num, "是素数")
// 打印注释掉，不然比较耗时
}
}
}
wg.Done()
}
func main() {
start := time.Now().Unix()
for i := 1; i <= 4; i++ {
wg.Add(1)
go test(i)
}
wg.Wait()
fmt.Println("执行完毕")
end := time.Now().Unix()
fmt.Println(end - start) //4毫秒
}
上面算法有两个问题：
- 协程数4写死了，不方便
- 无法打印（因为打印比较耗时，会影响时间统计）
如果想统计数据和打印数据同时进行，可以使用channel管道。
八、Channel管道
管道是 Golang 在语言级别上提供的 goroutine 间的通讯方式，我们可以使用 channel 在多个 goroutine 之间传递消息。 如果说 goroutine 是 Go 程序并发的执行体，channel 就是它们之间的连接。channel 是可以让一个 goroutine 发送特定值到另一个 goroutine 的通信机制。
Golang 的并发模型是 CSP （Communicating Sequential Processes） ，提倡通过通信共享内存而不是通过共享内存而实现通信。
Go 语言中的管道（channel）是一种特殊的类型。管道像一个传送带或者队列，总是遵循先入先出（First In First Out）的规则，保证收发数据的顺序。每一个管道都是一个具体类型的导管，也就是声明 channel 的时候需要为其指定元素类型。
1. channel类型
channel是一种引用类型，声明时需指定元素类型：
var ch1 chan int // 声明一个传递整型的管道 var ch2 chan bool // 声明一个传递布尔型的管道 var ch3 chan []int // 声明一个传递int切片的管道
2. 创建channel
使用 make 函数初始化后才能使用：
make(chan 元素类型, 容量)
示例：
ch1 := make(chan int, 10) // 创建一个能存储10个int类型数据的管道 ch2 := make(chan bool, 4) // 创建一个能存储4个bool类型数据的管道 ch3 := make(chan []int, 3) // 创建一个能存储3个int切片类型数据的管道
3. channel操作
管道有 发送（send） 接收（receive） 关闭（close） 三种操作。
package main
import "fmt"
func main() {
//1、创建channel
ch := make(chan int, 3)
//2、给管道里面存储数据
ch <- 10
ch <- 21
ch <- 32
//3、获取管道里面的内容
a := <-ch
fmt.Println(a) //10
<-ch //从管道里面取值 //21
c := <-ch
fmt.Println(c) //32
// 管道里面的数据取完了，又可以继续存数据
ch <- 56
ch <- 66
//4、打印管道的长度和容量
fmt.Printf("值：%v 容量：%v 长度%v\n", ch, cap(ch), len(ch))
//值：0xc0000d0080 容量：3 长度2
// 5、管道的类型（引用数据类型）
ch1 := make(chan int, 4)
ch1 <- 34
ch1 <- 54
ch1 <- 64
ch2 := ch1
ch2 <- 69
<-ch1
<-ch1
<-ch1
d := <-ch1
fmt.Println(d) // 69
// 6、管道阻塞
ch6 := make(chan int, 1)
ch6 <- 34
ch6 <- 64 //all goroutines are asleep - deadlock!
// ① 管道放满了放不下了，再去放就会导致管道阻塞
ch7 := make(chan string, 2)
ch7 <- "数据1"
ch7 <- "数据2"
m1 := <-ch7
m2 := <-ch7
m3 := <-ch7
fmt.Println(m1, m2, m3)
// fatal error: all goroutines are asleep - deadlock!
// ② 在没有使用协程的情况下，如果我们的管道数据已经全部取出，再取也会阻塞
//正确的写法，写一个取一个
ch8 := make(chan int, 1)
ch8 <- 34
<-ch8
ch8 <- 67
<-ch8
ch8 <- 78
m4 := <-ch8
fmt.Println(m4) // 78
}
关于关闭管道需要注意的事情是，只有在通知接收方 goroutine 所有的数据都发送完毕的时候才需要关闭管道。管道是可以被垃圾回收机制回收的， 它和关闭文件是不一样的，在结束操作之后关闭文件是必须要做的，但关闭管道不是必须的。
关闭管道的特点：
- 对一个已关闭的管道再发送值就会导致 panic。
- 关闭一个已关闭的管道会导致 panic。
- 对一个已关闭的管道进行取值，能继续接收，直到取完所有值。
- 对一个已关闭的并且没有值的管道执行接收操作会得到对应类型的零值。
4. 管道阻塞
无缓冲管道（容量为0）
果创建管道的时候没有指定容量，那么我们可以叫这个管道为无缓冲的管道
func main() {
ch := make(chan int)
ch <- 10
fmt.Println("发送成功")
}
// 错误：fatal error: all goroutines are asleep - deadlock!
有缓冲管道（容量 > 0）
只要管道的容量大于零，那么该管道就是有缓冲的管道，管道的容量表示管道中能存放元素的数量。就像你小区的快递柜只有那么个多格子，格子满了就装不下了，就阻塞了，等到别人取走一个快递员就能往里面放一个。
func main() {
ch := make(chan int, 1)
ch <- 10
fmt.Println("发送成功")
}
5. for range 从管道循环取值
package main
import "fmt"
// 循环遍历管道数据
func main() {
var ch1 = make(chan int, 5)
// 发送数据
for i := 0; i < 5; i++ {
ch1 <- i + 1
}
close(ch1) // 关闭管道
// for range 会自动判断管道是否关闭，管道关闭且取完所有值后，自动退出循环
// 如果使用for range遍历管道，并且管道没有关闭会报错fatal error: all goroutines are asleep - deadlock!
for val := range ch1 {
fmt.Println(val)
}
fmt.Println("循环结束了")
}
| 管道状态 | for range 的行为 |
|---|---|
| 管道未关闭且有值 | 正常取值，但是取完值还会继续等新值，因为管道还没关，最后死锁 |
| 管道未关闭且空（无发送者） | ⚠️ 死锁报错：fatal error: all goroutines are asleep - deadlock! |
| 管道已关闭且有值 | 正常取完所有值，然后退出循环 |
| 管道已关闭且为空 | 自动退出循环，不会报错 |
原因分析：
for range不知道后面还有没有数据进来，它会一直等待。- 如果管道没有被关闭，
for range会认为“可能还有数据要来”，所以永远等在那里。 - 如果没有其他 goroutine 在往管道里写数据，整个程序就会卡死（死锁） 。
解决方法： 发送方必须 close() 管道，告诉 for range "数据已经全部发完了"。
6. for循环从管道循环取值
// 2、通过for循环遍历管道的时候管道可以不关闭，也不会报错
var ch2 = make(chan int, 10)
for i := 1; i <= 10; i++ {
ch2 <- i
}
// 正好取10次，但是如果你取的次数多了还是会报错的
for j := 0; j < 10; j++ {
fmt.Println(<-ch2)
}
与for range的本质区别
| 对比 | for range 遍历 | for i:=0; i<n; i++ 遍历 |
|---|---|---|
| 取多少次 | 不知道，直到管道为空或关闭 | 明确知道次数（用 n 控制） |
| 管道未关闭会怎样 | ❌ 不知道什么时候结束，一直等 → 死锁 | ✅ 取完指定次数就停止，不再等待 |
| 是否必须关闭管道 | ✅ 必须关闭，否则死锁 | ❌ 不需要关闭，因为知道取多少次 |
九、Goroutine结合Channel管道
1. 同步写入和读取
package main
import (
"fmt"
"sync"
)
// 保证两个协程执行完毕
var wg sync.WaitGroup
// 向管道写入100条数据
func fn1(inChan chan int) {
for i := 0; i < 100; i++ {
inChan <- i
}
close(inChan) // 写入完成关闭管道
wg.Done()
}
// 从管道读取数据
func fn2(inChan chan int) {
for v := range inChan {
fmt.Println(v)
}
wg.Done()
}
func main() {
inChan := make(chan int, 10)
wg.Add(1)
go fn1(inChan)
wg.Add(1)
go fn2(inChan)
// 保证两个协程执行完毕
wg.Wait()
// 写入和读取会同步进行，如果写入比较慢也没事，因为读取会等写入完成再读取
}
2. 统计素数（goroutine + channel）
package main
import (
"fmt"
"sync"
"time"
)
var wg sync.WaitGroup
//向 intChan放入 1-120000个数
func putNum(intChan chan int) {
for i := 2; i < 120000; i++ {
intChan <- i
}
close(intChan)
wg.Done()
}
// 从 intChan取出数据，并判断是否为素数,如果是，就把得到的素数放在primeChan
func primeNum(intChan chan int, primeChan chan int, exitChan chan bool) {
for num := range intChan {
var flag = true
for i := 2; i < num; i++ {
if num%i == 0 {
flag = false
break
}
}
if flag {
primeChan <- num //num是素数，放到primeChan里面
}
}
// 要关闭 primeChan
// close(primeChan)
// primeNum被开了16个协程去写数据，如果一个方法执行的快，执行完把primeChan关闭了，那么其他方法就无法向primeChan发送数据了
// 什么时候关闭primeChan?
// 给exitChan里面放入一条数据表示执行完了1次，不在这里关闭primeChan
exitChan <- true
wg.Done()
}
//printPrime打印素数的方法
func printPrime(primeChan chan int) {
// 不打印了，计算时间
// for v := range primeChan {
// fmt.Println(v)
// }
wg.Done()
}
func main() {
start := time.Now().Unix()
// 存放1-120000数字的管道
intChan := make(chan int, 1000)
// 存放读取的素数的管道
primeChan := make(chan int, 50000)
// 标识primeChan close的管道
exitChan := make(chan bool, 16)
// 1. 存放数字的协程
wg.Add(1)
go putNum(intChan)
// 2. 统计素数的协程
for i := 0; i < 16; i++ {
wg.Add(1)
go primeNum(intChan, primeChan, exitChan)
}
// 3. 打印素数的协程
wg.Add(1)
go printPrime(primeChan)
// 4. 判断exitChan是否存满值
wg.Add(1)
go func() {
for i := 0; i < 16; i++ {
<-exitChan // 如果管道没值了，会阻塞等待。这里的疑问解释在下面👇🏻
}
// 关闭primeChan
close(primeChan)
wg.Done()
}()
wg.Wait()
end := time.Now().Unix()
fmt.Println("执行完毕....", end-start, "毫秒") // 3毫秒
}
<-exitChan 的解释
管道只有在已关闭且为空时，读取才会立即返回零值（不报错），但这里的情况不同：
| 情况 | 结果 |
|---|---|
| 管道未关闭且有值 | 正常取值 |
| 管道未关闭且为空 | 阻塞等待（不是报错） |
| 管道已关闭且为空 | 返回零值（不报错） |
| 管道已关闭且有值 | 正常取值（取完后再取返回零值） |
这里的 exitChan 从未被关闭，所以当它为空时，<-exitChan 会阻塞等待，而不是报错。
- 对于for range，循环次数无限，当管道未关闭并且无值的时候会一直等待，并且因为有无限循环次数，就会一直等，就会死锁。
- 对于for，循环次数定死了就是16，而且exitChan里面早晚会有16条数据，所以早晚会循环完成16次，所以不会死锁。阻塞等待 ≠ 死锁。死锁是永远等不到，而这里是确定能等到。当然如果把16改成17也是会死锁的。
十、单向管道
管道默认是双向的，但可以声明为只读或只写：
// 只写管道 var chan1 chan<- int chan1 <- 30 // 可以写入 // <-chan1 // 不能读取 // 只读管道 var chan2 <-chan int num := <-chan2 // 可以读取 // chan2 <- 30 // 不能写入
十一、select多路复用
在某些场景下需要同时从多个通道接收数据，可以使用 select 关键字。
package main
import (
"fmt"
"time"
)
func main() {
// 定义管道：10个int数据
intChan := make(chan int, 10)
for i := 0; i < 10; i++ {
intChan <- i
}
// close(intChan)
// 如果这里关闭通道，会死循环，一直打印0值，原因如下👇🏻
// 定义管道：5个string数据
stringChan := make(chan string, 5)
for i := 0; i < 5; i++ {
stringChan <- "hello" + fmt.Sprintf("%d", i)
}
// close(stringChan)
// 如果这里关闭通道，会死循环，一直打印0值，原因如下👇🏻
// 当然也可以使用两个for循环去读取，但是那样就是串行执行了
// 当然也可以使用协程同时从多个管道获取数据，但是如果你只想在当前main方法里面实现，就只能用select了
for {
select {
case v := <-intChan:
fmt.Printf("从intChan读取的数据: %d\n", v)
case v := <-stringChan:
fmt.Printf("从stringChan读取的数据: %s\n", v)
default:
fmt.Println("都取不到了，不玩了")
time.Sleep(time.Second)
return // 注意退出，跳出for循环
}
}
}
select从多个 case 中随机选择一个可执行的- 如果多个 case 同时满足（管道都有数据），会随机选一个执行。
- 如果一个 case 都不满足（所有管道都为空且未关闭），就执行
default。所以不会死锁。 - 管道未关闭 → 数据取完 → 管道为空 → 阻塞等待 → 两个都阻塞 → 走 default → return ✅
- 管道已关闭 → 数据取完 → 管道为空 → 读取不阻塞 → 返回零值 → 走 case → 无限循环 ❌
十二、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() {
if err := recover(); err != nil {
fmt.Println("test()发生错误:", err)
}
}()
panic("test()中发生了panic")
}
func main() {
go sayHello()
go test()
// 为了防止主线程提前退出，睡眠一下，当然也可以用上面的协程
for i := 0; i < 10; i++ {
fmt.Println("main() ok")
time.Sleep(time.Second)
}
}
十三、Golang并发安全和锁
1. 互斥锁（Mutex）
互斥锁保证同一时间只有一个goroutine进入临界区。
有问题代码（存在资源竞争）：
package main
import (
"fmt"
"sync"
"time"
)
var count = 0
var wg sync.WaitGroup
func test() {
count++
fmt.Println("the count is : ", count)
time.Sleep(time.Millisecond)
wg.Done()
}
func main() {
for r := 0; r < 20; r++ {
wg.Add(1)
go test()
}
wg.Wait()
}
使用
go build -race main.go检测数据竞争。
使用互斥锁解决：
package main
//go build -race main.go 编译后运行查看
import (
"fmt"
"sync"
"time"
)
var count = 0
var wg sync.WaitGroup
// 如果不加mutex，会同时又多个协程竞争count
var mutex sync.Mutex
func test() {
mutex.Lock()
count++
fmt.Println("the count is : ", count)
time.Sleep(time.Millisecond)
mutex.Unlock()
wg.Done()
}
func main() {
for r := 0; r < 20; r++ {
wg.Add(1)
go test()
}
wg.Wait()
}
2. 读写互斥锁（RWMutex）
互斥锁的本质是当一个 goroutine 访问的时候， 其他 goroutine 都不能访问。 这样在资源同步，避免竞争的同时也降低了程序的并发性能。程序由原来的并行执行变成了串行执行。
其实，当我们对一个不会变化的数据只做“读”操作的话，是不存在资源竞争的问题的。因为数据是不变的，不管怎么读取，多少 goroutine 同时读取，都是可以的。
所以问题不是出在“读”上，主要是修改，也就是“写”。修改的数据要同步，这样其他 goroutine 才可以感知到。所以真正的互斥应该是读取和修改、修改和修改之间，读和读是没有互斥操作的必要的。
因此，衍生出另外一种锁，叫做读写互斥锁。
读写锁可以让多个读操作并发， 同时读取， 但是对于写操作是完全互斥的。 也就是说， 当一个 goroutine 进行写操作的时候，其他 goroutine 既不能进行读操作，也不能进行写操作。
package main
import (
"fmt"
"sync"
"time"
)
var wg sync.WaitGroup
// 读写互斥锁
var mutex sync.RWMutex
// 写的方法
func write() {
mutex.Lock()
fmt.Println("执行写操作")
time.Sleep(time.Second * 2)
mutex.Unlock()
wg.Done()
}
// 读的方法
func read() {
mutex.RLock()
fmt.Println("执行读操作")
time.Sleep(time.Second * 2)
mutex.RUnlock()
wg.Done()
}
func main() {
//开启10个协程执行读操作
for i := 0; i < 10; i++ {
wg.Add(1)
go write()
}
// 开启10个协程执行写操作
for i := 0; i < 10; i++ {
wg.Add(1)
go read()
}
wg.Wait()
}
到此这篇关于Golang goroutine channel 实现并发和并行的文章就介绍到这了,更多相关Golang goroutine channel并发和并行内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！