线程加锁
线程安全
1
2
3
4
5
6
7
8
9
|
func unsafeWrite() {
conflictMap := map[int]int{}
for i := 0; i < 1<<10; i++ {
go func(i int) {
conflictMap[0] = i
}(i)
}
}
|
锁
Go 不仅支持基于 CSP 的通信模型，也支持基于共享内存的多线程数据访问
Sync 包提供了锁的基本原语
| 原语 |
描述 |
sync.Mutex |
互斥锁 |
sync.RWMutex |
读写分离锁 |
sync.WaitGroup |
等待一组 goroutine 返回 |
sync.Once |
保证某段代码只执行 1 次 |
sync.Cond |
让一组 goroutine 在满足特定条件时被唤醒 |
Mutex
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| type SafeMap struct {
sync.Mutex
safeMap map[int]int
}
func (m *SafeMap) Write(k, v int) {
m.Lock()
defer m.Unlock()
m.safeMap[k] = v
}
func safeWrite() {
m := SafeMap{
Mutex: sync.Mutex{},
safeMap: map[int]int{},
}
for i := 0; i < 1<<10; i++ {
go func(i int) {
m.Write(0, i)
}(i)
}
}
|
RWMutex
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
| func main() {
go rLock()
go wLock()
time.Sleep(time.Second)
}
func rLock() {
mutex := sync.RWMutex{}
for i := 0; i < 1<<2; i++ {
mutex.RLock()
defer mutex.RUnlock()
fmt.Println("rLock", i)
}
}
func wLock() {
mutex := sync.RWMutex{}
for i := 0; i < 1<<2; i++ {
mutex.Lock()
defer mutex.Unlock()
fmt.Println("wLock", i)
}
}
|
WaitGroup
1
2
3
4
5
6
7
8
9
10
11
12
13
| func waitByChannel(n int) {
ch := make(chan int, n)
for i := 0; i < n; i++ {
go func(i int) {
ch <- i
}(i)
}
for i := 0; i < n; i++ {
<-ch
}
}
|
1
2
3
4
5
6
7
8
9
10
11
12
| func waitByWG(n int) {
wg := sync.WaitGroup{}
wg.Add(n)
for i := 0; i < n; i++ {
go func(i int) {
wg.Done()
}(i)
}
wg.Wait()
}
|
Cond
生产者消费者模型
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
| type Queue struct {
queue []interface{}
cond *sync.Cond
}
func (q *Queue) Enqueue(item interface{}) {
q.cond.L.Lock()
defer q.cond.L.Unlock()
q.queue = append(q.queue, item)
q.cond.Broadcast()
fmt.Println("enqueue and broadcast")
}
func (q *Queue) Dequeue() (item interface{}) {
q.cond.L.Lock()
defer q.cond.L.Unlock()
if len(q.queue) == 0 {
fmt.Println("no data and wait")
q.cond.Wait()
}
item = q.queue[0]
q.queue = q.queue[1:]
return item
}
func main() {
q := Queue{
queue: []interface{}{},
cond: &sync.Cond{L: &sync.Mutex{}},
}
go func() {
for {
q.Enqueue("a")
time.Sleep(time.Second * 1 << 1)
}
}()
go func() {
for {
q.Dequeue()
time.Sleep(time.Second)
}
}()
time.Sleep(time.Second * 1 << 4)
}
|
Once
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
| type SliceNum []int
func (s *SliceNum) Add(item int) *SliceNum {
*s = append(*s, item)
fmt.Println("add", item)
return s
}
func NewSlice() SliceNum {
return make(SliceNum, 0)
}
func main() {
once := sync.Once{}
slice := NewSlice()
for i := 0; i < 1<<2; i++ {
once.Do(func() {
slice.Add(1 << 4)
})
}
}
|
线程调度
进程是资源分配的基本单位，线程是调度的基本单位
在 Kernel 视角，进程和线程本质无差别，都是以 task_struct 来描述
内存使用
页表：虚拟地址与物理地址的映射关系
1
2
3
4
| package main
func main(){
}
|
1
2
3
4
5
6
7
8
| $ go build
$ size go-demo
text data bss dec hex filename
758966 13328 227280 999574 f4096 go-demo
$ objdump -x go-demo | grep bss | grep runtime | head -n 1
00000000000e3420 l O .bss 0000000000000000 runtime.bss
|
1
2
| $ getconf PAGE_SIZE
4096
|
访问内存
- CPU 上有一个
MMU（MemoryManagementUnit）
- CPU 将
虚拟地址给到 MMU，而 MMU 去物理内存中查找页面，得到实际的物理地址
- CPU 维护一份
TLB（Translation Lookaside Buffer）：缓存虚拟地址和物理地址的映射关系
切换开销
内核进程
进程切换的开销很大
直接开销
- 切换
PGD
- 刷新
TLB
- 切换
内核态堆栈 - 系统调用
- 切换
硬件上下文
系统调度器的代码执行
间接开销
CPU 缓存失效：导致进程需要直接访问内存
内核线程
共享虚拟空间地址
- 线程本质上是一组
共享资源的进程，线程切换本质上依然需要内核进行进程切换（系统调用）
- 一个进程内的所有线程共享
虚拟地址空间，主要节省：虚拟空间的切换（PGD、TLB 等）
虽然内核线程的切换不需要切换虚拟地址空间，但内核线程本质上还是内核进程，切换依然需要系统调用
用户线程
无需内核帮助，应用程序在用户空间创建的可执行单元，创建和销毁完全在用户态完成
在 Kernel ，多个用户线程本质上是同一个内核线程，同一个内核线程内的用户线程之间的切换是不需要系统调用的
Goroutine 是用户线程的一种实现
Goroutine
Go 基于 GMP 模型实现用户线程
| Key |
Desc |
G = Goroutine |
每个 goroutine 有自己的栈空间（初始大小为 2k 左右）和定时器 |
M = Machine
= 内核线程 |
记录内核线程栈信息，当 goroutine 调度到内核线程时，使用 goroutine 自己的栈信息
数量一般与 CPU 数 相同，一个 CPU 一般对应一个 M |
P = Process
= 调度器 |
负责调度 goroutine
维护一个本地 goroutine 队列，M 从 P上获得 goroutine 并执行 |
虚线代表关系不稳定
GMP 模型
