多线程调试

1package main
2
3import (
4	"bytes"
5	"fmt"
6	"runtime"
7	"strconv"
8	"sync"
9	"time"
10)
11
12// 辅助函数用于获取协程ID
13func getGID() uint64 {
14	// 创建长度为64的字节切片
15	b := make([]byte, 64)
16	// 调用runtime包中的stack函数获取协程栈跟踪信息
17	// 它会将切片裁剪为实际的栈跟踪内容
18	b = b[:runtime.Stack(b, false)]
19	// 移除栈跟踪中的 "goroutine " 前缀
20	b = bytes.TrimPrefix(b, []byte("goroutine "))
21	// 通过解析字节直到第一个空格字符来获取协程ID
22	b = b[:bytes.IndexByte(b, ' ')]
23	// 将协程ID从字符串转换为uint64
24	n, _ := strconv.ParseUint(string(b), 10, 64)
25	return n
26}
27
28// 日志函数，记录带时间戳和协程ID的字符串
29func log(s string) {
30	prefix := fmt.Sprintf("%v [%d]", time.Now(), getGID())
31	logs := fmt.Sprintf("%v : %s", prefix, s)
32	fmt.Println(logs)
33}
34
35// 队列实现
36type Queue struct {
37	items        []int   // 实际队列
38	mu           sync.Mutex // 互斥锁，保护队列
39	backpressure int // 背压计数器
40}
41
42// Enqueue：向队列添加元素
43func (q *Queue) Enqueue(item int) {
44	q.mu.Lock() // 获取锁
45	defer q.mu.Unlock() // 函数返回时释放锁
46	log("向队列添加元素")
47	q.items = append(q.items, item) // 将元素添加到队列
48}
49
50// Dequeue：从队列移除元素
51func (q *Queue) Dequeue() int {
52	q.mu.Lock() // 获取锁
53	defer q.mu.Unlock() // 函数返回时释放锁
54	if len(q.items) == 0 {
55		return -1 // 队列为空时返回-1
56	}
57	log("从队列前端移除元素")
58	front := q.items[0]
59	q.items = q.items[1:] // 移除队列的第一个元素
60	return front
61}
62
63func (q *Queue) ResetQueue() {
64	q.mu.Lock() // 获取锁
65	defer q.mu.Unlock() // 函数返回时释放锁
66	q.items = []int{} // 重置队列
67	q.ResetBP() // 重置背压
68}
69
70// 增加和减少背压计数器
71
72// IncBP 将背压计数器增加1
73// 可在向队列添加元素时使用，表示有更多工作待处理
74func (q *Queue) IncBP() {
75	q.backpressure = q.backpressure + 1
76}
77
78// DecBP 将背压计数器减少1
79// 可在处理（移除）队列中的元素时使用，表示剩余工作减少
80func (q *Queue) DecBP() {
81	q.backpressure = q.backpressure - 1
82}
83
84// ResetBP 将背压计数器重置为0
85// 可在清空或重置队列时使用，表示没有剩余工作
86func (q *Queue) ResetBP() {
87	q.backpressure = 0
88}
89
90// 从队列中移除元素并进行处理
91func ProcessWork(q *Queue) {
92	log("处理队列")
93	q.DecBP() // 减少背压
94	q.Dequeue() // 从队列中移除元素
95}
96
97// 向队列添加元素
98func AddWorkItem(q *Queue, i int) {
99	log("向队列添加元素")
100	q.IncBP() // 增加背压
101	q.Enqueue(i) // 将元素添加到队列
102}
103
104// 批量向队列添加元素
105func BulkAdds(q *Queue, items []int) {
106	log(fmt.Sprintf("向队列添加 %d 个元素", len(items)))
107	for _, n := range items {
108		AddWorkItem(q, n)
109	}
110}
111
112// 客户端代码，使用上述库
113func main() {
114	queue := Queue{}
115	var wg sync.WaitGroup
116	N := 100
117	wg.Add(N)
118	for i := 0; i < (N / 2); i++ {
119		n := i
120		// 启动一个协程向队列添加工作项
121		go func() {
122			AddWorkItem(&queue, n)
123			wg.Done() // 表示该协程已完成
124		}()
125	}
126	for i := 0; i < (N / 2); i++ {
127		// 启动一个协程处理队列中的工作项
128		go func() {
129			ProcessWork(&queue)
130			wg.Done() // 表示该协程已完成
131		}()
132	}
133
134	wg.Wait() // 等待所有协程完成
135	log(fmt.Sprintf("完成 %d", queue.backpressure))
136}