Goroutine
Go 协程是由 Go 运行时管理的轻量级线程。它们允许在单个进程中同时执行多个函数或方法,而无需显式地管理线程的生命周期。Go 协程比传统的操作系统线程更轻量,启动和切换的开销更小,因此可以轻松地创建成千上万个协程。
1 创建协程
func sayHello() {
fmt.Println("Hello from goroutine!")
}
func main() {
go sayHello() // 启动一个新的协程
fmt.Println("Hello from main!")
time.Sleep(1 * time.Second) // 等待协程完成
}
输出结果可能是:
Hello from main!
Hello from goroutine!
也可能是:
Hello from goroutine!
Hello from main!
因为两个打印操作是并发执行的,顺序不确定。
2 GMP 调度器
GMP 调度器是 Go 语言实现高效并发的关键组件。它通过将 goroutine 与操作系统线程解耦,并使用工作窃取、局部性原理、抢占式调度和网络轮询器等技术,实现了高效的并发调度和执行。
2.1 组成部分
- G(Goroutine):表示一个 goroutine,它是 Go 语言中的轻量级线程。每个 goroutine 都有自己的栈、程序计数器和局部变量等。
- M(Machine):表示一个操作系统线程。M 负责执行 G,并与操作系统进行交互。一个 M 可以执行多个 G,但是在任意时刻,一个 M 只能执行一个 G。
- P(Processor):表示一个逻辑处理器,它是 G 和 M 之间的中间层。P 负责管理和调度分配给它的 G。每个 P 都有自己的本地队列,用于存储待执行的 G。P 的数量决定了可以同时运行的 G 的最大数量。
2.2 工作原理
- 当创建一个新的 goroutine 时,它会被添加到当前 P 的本地队列中。
- M 从与其关联的 P 的本地队列中获取一个 G 来执行。如果本地队列为空,M 会尝试从其他 P 的本地队列中窃取 G,或者从全局队列中获取 G。
- 如果所有队列都为空,M 会进入休眠状态,等待新的 G 被创建或者被唤醒。
- 当一个 G 阻塞时(例如等待 I/O 操作完成),M 会将其与当前的 P 解除关联,并尝试从其他 P 的本地队列中获取一个新的 G 来执行。
- 当一个 G 完成执行时,它会返回到其所属的 P 的本地队列中,等待下一次调度。
2.3 优化
- 工作窃取(Work Stealing):当一个 P 的本地队列为空时,它会尝试从其他 P 的本地队列中窃取 G。这种机制可以平衡各个 P 之间的负载,提高整体调度效率。
- 局部性原理(Locality Principle):GMP 调度器会尽量将相关的 G 调度到同一个 P 上执行,以提高缓存命中率和减少内存访问延迟。
- 抢占式调度(Preemptive Scheduling):Go 1.14 引入了基于信号的抢占式调度,可以在长时间运行的 G 上强制插入调度点,以避免某个 G 长时间占用 M,导致其他 G 无法得到执行。
- 网络轮询器(Network Poller):Go 运行时内置了一个网络轮询器,用于高效地处理 I/O 多路复用。当 G 阻塞在 I/O 操作上时,M 可以与网络轮询器协作,继续执行其他 G,从而提高整体性能。
3 WaitGroup
WaitGroup
是 Go 语言中用于等待一组 goroutine 完成的同步原语。它属于 sync
包,提供了一种简单而有效的方式来协调多个并发执行的 goroutine,确保主程序在所有子任务完成后再继续执行。
- 创建一个
WaitGroup
var wg sync.WaitGroup
- 添加等待的 goroutine 数量
使用 Add
方法来设置需要等待的 goroutine 数量。这通常在启动 goroutine 之前调用。
wg.Add(3) // 表示需要等待 3 个 goroutine 完成
- 在每个 goroutine 完成时调用
Done
在每个 goroutine 的逻辑结束时,调用 Done
方法来通知 WaitGroup
该 goroutine 已完成。通常使用 defer
语句确保 Done
被调用。
go func() {
defer wg.Done()
// goroutine 的逻辑代码
}()
- 等待所有 goroutine 完成
在主 goroutine 中调用 Wait
方法,它会阻塞直到所有的 goroutine 调用了 Done
。
wg.Wait()
fmt.Println("All Done")
4 互斥锁
Go 语言中的互斥锁(Mutex)是一种同步原语,用于在多个 goroutine 之间保护共享资源的访问,防止数据竞争和不一致性。互斥锁确保在同一时间只有一个 goroutine 能够访问被保护的资源。
Go 的 sync
包提供了 Mutex
类型,用于实现互斥锁。以下是互斥锁的基本使用方法:
首先,需要导入 sync
包:
import "sync"
定义一个 sync.Mutex
变量:
var mu sync.Mutex
使用 Lock()
方法对互斥锁进行加锁,使用 Unlock()
方法进行解锁。通常,Unlock()
会在 defer
语句中调用,以确保在函数退出时释放锁。
func example() {
mu.Lock()
defer mu.Unlock()
// 访问共享资源
}
完整示例:
var counter int
var mu sync.Mutex
var wg = sync.WaitGroup{}
func count() {
mu.Lock()
defer mu.Unlock()
defer wg.Done()
counter++
}
func main() {
for i := 0; i < 1000000; i++ {
wg.Add(1)
go count()
}
wg.Wait()
fmt.Println(counter)
}
5 原子操作包
sync/atomic
包提供了一系列原子操作函数,这些函数可以对基本数据类型(如 int32
、int64
、uint32
、uint64
、uintptr
和 unsafe.Pointer
)进行原子性的读取、写入、比较和交换等操作。
以下是一些常用的 atomic
操作函数:
Add
:原子地将delta
添加到*addr
并返回新值。
var ops uint64 = 0
atomic.AddUint64(&ops, 1)
fmt.Println("ops:", ops)
CompareAndSwap
:原子地比较addr
的旧值和old
。如果相等,则将addr
的值设为新值并返回true
;否则返回false
。
var value int32 = 3
swapped := atomic.CompareAndSwapInt32(&value, 3, 5)
fmt.Println(swapped, value) // true, 5
Load
:原子地加载*addr
。
var value int32 = 18
loadedValue := atomic.LoadInt32(&value)
fmt.Println(loadedValue) // 18
Store
:原子地存储val
到*addr
。
var value int32 = 0
atomic.StoreInt32(&value, 20)
fmt.Println(value) // 20
Swap
:原子地将val
存储到*addr
并返回*addr
的旧值。
var value int32 = 4
oldValue := atomic.SwapInt32(&value, 5)
fmt.Println(oldValue, value) // 4, 5
原子操作在并发编程中非常有用,因为它们可以避免使用锁,从而减少性能开销和死锁的风险。然而,需要注意的是,并非所有操作都可以通过原子操作来实现。对于复杂的操作,仍然需要使用互斥锁(sync.Mutex
)或其他同步机制来确保数据的一致性。
6 读写互斥锁
RWMutex
(读写互斥锁)是 Go 语言中用于管理共享资源访问的一种同步机制,位于 sync
包下。与普通的互斥锁(Mutex
)不同,RWMutex
允许多个读操作同时进行,但在写操作时会阻塞所有其他读写操作。
6.1 主要特点
- 读锁(RLock):
- 多个 goroutine 可以同时持有读锁。
- 当有线程持有读锁时,其他 goroutine 仍然可以获取读锁,但不能获取写锁。
- 写锁(Lock):
- 写锁是独占的,当一个 goroutine 持有写锁时,其他任何 goroutine 都不能获取读锁或写锁。
- 写锁会阻塞所有其他读锁和写锁的获取,直到写锁被释放。
6.2 常用方法
方法 | 说明 |
---|---|
RLock() | 获取一个读锁 |
RUnlock() | 释放一个读锁 |
Lock() | 获取一个写锁 |
Unlock() | 释放一个写锁 |
7 Channel
- Channel:一种类型化的管道,可以通过它发送和接收特定类型的值。
- 发送(Send):将一个值发送到 Channel 中。
- 接收(Receive):从 Channel 中接收一个值。
- 阻塞(Blocking):如果发送方尝试向满的 Channel 发送数据,或者接收方尝试从空的 Channel 接收数据,操作将会阻塞,直到另一端准备好。
7.1 创建
ch := make(chan int) // 创建一个传递 int 类型的无缓冲 Channel
chBuffered := make(chan int, 10) // 创建一个传递 int 类型的有缓冲 Channel,缓冲区大小为 10
7.2 无缓冲
无缓冲 Channel 在发送和接收操作完成之前会阻塞。
ch := make(chan string)
go func() {
time.Sleep(2 * time.Second)
ch <- "Zhang"
}()
msg := <-ch
fmt.Println(msg)
7.3 有缓冲
有缓冲 Channel 允许在阻塞之前存储一定数量的元素。
ch := make(chan int, 2)
ch <- 1
ch <- 2
// ch <- 3 // 如果取消注释,这里会阻塞,因为缓冲区已满
fmt.Println(<-ch) // 输出 1
fmt.Println(<-ch) // 输出 2
7.4 关闭
关闭 Channel 表示不会再有更多的值发送到该 Channel。接收方可以通过检测 Channel 是否关闭来决定如何处理接收到的数据。
ch := make(chan int, 2)
ch <- 1
ch <- 2
close(ch)
for v := range ch {
fmt.Println(v)
}
7.5 多路复用(Select)
select
语句允许同时等待多个 Channel 操作,类似于 switch
语句,但用于 Channel。
ch1 := make(chan string)
ch2 := make(chan string)
go func() {
time.Sleep(1 * time.Second)
ch1 <- "ch1"
}()
go func() {
time.Sleep(2 * time.Second)
ch2 <- "ch2"
}()
select {
case msg1 := <-ch1:
fmt.Println(msg1)
case msg2 := <-ch2:
fmt.Println(msg2)
}
当多个 case
同时就绪时,select
会随机选择一个执行。这有助于避免某些 Goroutine 被饿死(即一直得不到执行机会)。
通常还会设置一个超时时间,避免阻塞等待的时间过长:
timer := time.NewTimer(3 * time.Second) // 会在 3s 后向 Channel timer.C 写入时间
select {
case msg1 := <-ch1:
fmt.Println(msg1)
case msg2 := <-ch2:
fmt.Println(msg2)
case <-timer.C:
fmt.Println("Timed out")
}
7.6 单向 Channel
单向Channel分为两种类型:
- 只发送通道(Send-only Channel):只能用于发送数据,不能用于接收数据。
- 只接收通道(Receive-only Channel):只能用于接收数据,不能用于发送数据。
首先,需要创建一个双向 Channel,然后可以将其转换为单向 Channel:
ch := make(chan int) // 创建一个双向的整数通道
var sendOnly chan<- int = ch // sendOnly 只能用于发送数据
var receiveOnly <-chan int = ch // receiveOnly 只能用于接收数据
单向Channel常用于函数参数,以确保函数只能以特定的方式使用通道:
// 生产者函数,只发送数据
func producer(sendCh chan<- int) {
for i := 0; i < 5; i++ {
sendCh <- i
fmt.Println("Produced:", i)
}
close(sendCh)
}
// 消费者函数,只接收数据
func consumer(receiveCh <-chan int) {
for num := range receiveCh {
fmt.Println("Consumed:", num)
}
}
func main() {
ch := make(chan int)
go producer(ch)
consumer(ch)
}
8 context
context
包提供了一种机制,用于在不同的 Goroutine 之间传递请求范围内的数据、取消信号以及截止时间(deadline)。
主要用途:
- 传递请求范围内的数据:如请求 ID、用户认证信息等。
- 取消信号:通知相关的 Goroutine 停止当前操作。
- 设置截止时间(Deadline):为操作设置一个超时时间,防止操作无限期地阻塞。
8.1 核心接口
type Context interface {
Deadline() (deadline time.Time, ok bool) // 返回上下文的截止时间
Done() <-chan struct{} // 返回一个通道,当上下文被取消或到达截止时间时关闭
Err() error // 返回上下文被取消的原因
Value(key interface{}) interface{} // 返回与 key 关联的值,如果没有则返回 nil
}
8.2 创建根上下文
根上下文是一个没有任何值、不会被取消且没有截止时间的上下文,通常作为其他上下文的父级使用。
ctx := context.Background()
8.3 派生上下文
可以从根上下文或其他派生上下文中创建新的上下文,添加取消功能、截止时间或键值对。
带取消功能的上下文:
ctx, cancel := context.WithCancel(context.Background())
defer cancel() // 确保在不需要时取消上下文,释放资源
带截止时间的上下文:
deadline := time.Now().Add(10 * time.Second)
ctx, cancel := context.WithDeadline(context.Background(), deadline)
defer cancel()
或者使用 WithTimeout
简化:
ctx, cancel := context.WithTimeout(context.Background(), 10 * time.Second)
defer cancel()
带值的上下文:
ctx := context.WithValue(context.Background(), "userID", 12345)