如何在Go中同步线程

作者头像东城·2026-01-31 23:26:37·
Golang大师课程（8部分系列）
如何Go：基础设施的宠儿
Golang大师课程：接口驱动设计
Golang大师课程：结构体嵌入
Golang大师课程：函数式选项
Golang大师课程：依赖注入(DI)
Golang大师课程：单例结构体将拯救你的项目
如何在Golang中构建美丽的GUI：3种Web UI路径
如何在Go中同步线程
单线程代码已经带来头痛。添加第二个线程，就是从基础头痛升级了。
解决方案？互斥锁：线程和数据的交通警察。
一旦你理解了它们，线程同步就变成了第二本能，语言无关。
在C++和Go中工作，我遇到过所有常见的混乱：
有时会吞噬数据的竞态条件
线程践踏内存导致的段错误
还有沉默的杀手：死锁
最后一个是最糟糕的，没有崩溃，没有错误。只是一个死程序，卡在永恒的线程对峙中。
但当你理解了互斥锁背后的核心思想时，一切都开始变得清晰。
最好的部分？每种语言都说互斥锁：
Go → sync.Mutex
C++ → std::mutex
Python → threading.Lock()
Java → ReentrantLock
在这篇文章中，我将分解互斥锁的概念，展示死锁是如何发生的，并给你足够的直觉来处理任何语言中的线程代码。
学一次 → 到处应用。
互斥锁：互斥锁
线程引入了全新类别的问题，特别是在Go中，生成数千个线程实际上是免费的。
现在想象两个线程在完全相同的时间击中同一个数据源。那就是混乱。竞态条件、数据损坏、神秘bug，你不想调试的东西，更不用说向你的团队解释了。
进入互斥锁： 线程和共享数据之间的交通警察。
没有锁：
  1. 线程A ---> 数据源 <--- 线程B
有锁（两个线程之间共享）：
  1. 线程A [锁]---> 数据源 <---[锁] 线程B
互斥锁的工作很简单：一次只有一个线程进入。
如果线程A拥有锁，线程B被告知："等待你的轮次。"
这里是一个简单的切片访问示例，有和没有锁：
没有锁：
  1. package main

  3. import (
  4. "fmt"
  5. "time"
  6. )

  8. func main() {
  9. var numbers []int

  11. // 启动5个goroutine，都向同一个切片追加
  12. for i := 0; i < 5; i++ {
  13. go func(n int) {
  14. // 这里没有锁定，这很可能会导致数据竞争
  15. numbers = append(numbers, n)
  16. fmt.Println("追加", n, "→", numbers)
  17. }(i)
  18. }

  20. // 给它们一点时间运行
  21. time.Sleep(1 * time.Second)
  22. }
go
有锁：
  1. package main

  3. import (
  4. "fmt"
  5. "sync"
  6. "time"
  7. )

  9. func main() {
  10. var (
  11. numbers []int
  12. mu sync.Mutex
  13. )

  15. for i := 0; i < 5; i++ {
  16. go func(n int) {
  17. mu.Lock() // 获取锁
  18. defer mu.Unlock() // 确保它被释放，即使在panic时

  20. numbers = append(numbers, n)
  21. fmt.Println("追加", n, "→", numbers)
  22. }(i)
  23. }

  25. time.Sleep(1 * time.Second)
  26. }
go
注意我们如何做：
  1. mu.Lock()
  2. defer mu.Unlock()
go
defer保证无论我们如何退出那个goroutine，正常返回或panic，锁都会被释放。
一旦goroutine接触共享数据，就锁定它。 相信我，未来的你会感激的。
那么，死锁到底是什么？
死锁
回到我们的交通警察类比：
  1. 线程A [锁]---> 数据源 <---[锁] 线程B
这有效是因为一个共享锁控制访问。但当我们引入同一车道中的两个共享锁时会发生什么？
  1. 线程A [锁]--[锁]-> 数据源 <---[锁] 线程B
现在你有两个交通警察，都不知道谁负责。线程A卡在等待两者，永远在困惑中来回弹跳。这就是经典死锁。
通常的嫌疑人？相同的嵌套锁，调用一个从已经持有锁的另一个函数内部获取锁的函数。
这是一个真实世界的例子：
  1. func (m *ScheduledTask) Create(...) (task, error) {
  2. m.mu.Lock() // 锁1
  3. defer m.mu.Unlock() // 在结束时解锁1

  5. // ... 设置任务 ...

  7. if err := m.saveTasks(); err != nil { // 内部的锁2
  8. return task{}, err
  9. }

  11. return t, nil
  12. }
go
现在看看saveTasks内部：
  1. func (m *ScheduledTask) saveTasks() error {
  2. m.mu.Lock() // 锁2（再次）
  3. defer m.mu.Unlock()

  5. data, err := json.MarshalIndent(m.tasks, "", " ")
  6. if err != nil {
  7. return err
  8. }

  10. return os.WriteFile(tasks, data, 0644)
  11. }
go
死锁。
为什么？因为Create()已经持有锁，而saveTasks()试图再次获取它，在第一个被释放之前。Go例程不会抱怨，它们只是默默地冻结。没有崩溃，没有堆栈跟踪，只是一个僵尸线程吞噬资源。
主线程呢？完全不知道。继续运行，而你的程序挂在边缘。
如果你认真对待构建真实世界的软件，你需要理解同步。
这些概念适用于所有语言。这是C++版本：
  1. std::lock_guard<std::mutex> lk(globalIPCData.mapMutex); // 在访问前锁定
  2. UIelement& u = uiSet.get(entityId);
cpp
学好这个。
一旦你将互斥锁视为具有绝对权威的交通警察，大多数线程问题就消失了。
我将在Substack上发布更多关于后端主题、JavaScript、Golang、C++和低级系统的深度探讨。希望你在那里；来打个招呼：
Coffee & Kernels
更多内容：
深度学习指南
深度学习搭便车指南：Python和JS示例。
深度学习：Pytorch和Tensorflow.js示例。
如何不那么糟糕地使用数据库
如何不那么糟糕地使用数据库和数据系统，带JavaScript示例。
数据库、数据系统和意图语言。
多语言互斥锁示例
C++ → std::mutex
  1. #include <iostream>
  2. #include <mutex>
  3. #include <thread>
  4. #include <vector>

  6. int main() {
  7. int counter = 0;
  8. std::mutex m;
  9. std::vector<std::thread> threads;

  11. for (int i = 0; i < 5; ++i) {
  12. threads.emplace_back([&](){
  13. std::lock_guard<std::mutex> lock(m); // RAII：构造函数锁定，析构函数解锁
  14. ++counter; // 临界区
  15. std::cout << "C++计数器: " << counter << "\n";
  16. });
  17. }

  19. for (auto &t : threads) t.join();
  20. return 0;
  21. }
cpp
Python → threading.Lock()
  1. import threading

  3. counter = 0
  4. lock = threading.Lock()

  6. def worker():
  7. global counter
  8. with lock: # 上下文管理器获取和释放
  9. counter += 1 # 临界区
  10. print(f"Python计数器: {counter}")

  12. threads = []
  13. for _ in range(5):
  14. t = threading.Thread(target=worker)
  15. t.start()
  16. threads.append(t)

  18. for t in threads:
  19. t.join()
python
Java → ReentrantLock
  1. import java.util.concurrent.locks.ReentrantLock;

  3. public class Main {
  4. private static int counter = 0;
  5. private static final ReentrantLock lock = new ReentrantLock();

  7. public static void main(String[] args) throws InterruptedException {
  8. Thread[] threads = new Thread[5];

  10. for (int i = 0; i < 5; i++) {
  11. threads[i] = new Thread(() -> {
  12. lock.lock(); // 🔒 获取
  13. try {
  14. counter++; // 临界区
  15. System.out.println("Java计数器: " + counter);
  16. } finally {
  17. lock.unlock(); // 🔓 释放
  18. }
  19. });
  20. threads[i].start();
  21. }

  23. for (Thread t : threads) t.join();
  24. }
  25. }
java
🔑 关键要点
无论什么语言，配方都是：
获取锁/互斥锁，然后接触共享数据
做你最小的临界工作
释放锁/互斥锁（或使用作用域/RAII/上下文自动释放）
实际应用示例
1. 银行账户示例
  1. package main

  3. import (
  4. "fmt"
  5. "sync"
  6. "time"
  7. )

  9. type BankAccount struct {
  10. balance int
  11. mu sync.Mutex
  12. }

  14. func (ba *BankAccount) Deposit(amount int) {
  15. ba.mu.Lock()
  16. defer ba.mu.Unlock()
  18. ba.balance += amount
  19. fmt.Printf("存款 %d，余额: %d\n", amount, ba.balance)
  20. }

  22. func (ba *BankAccount) Withdraw(amount int) bool {
  23. ba.mu.Lock()
  24. defer ba.mu.Unlock()
  26. if ba.balance >= amount {
  27. ba.balance -= amount
  28. fmt.Printf("取款 %d，余额: %d\n", amount, ba.balance)
  29. return true
  30. }
  32. fmt.Printf("余额不足，无法取款 %d\n", amount)
  33. return false
  34. }

  36. func (ba *BankAccount) GetBalance() int {
  37. ba.mu.Lock()
  38. defer ba.mu.Unlock()
  40. return ba.balance
  41. }

  43. func main() {
  44. account := &BankAccount{balance: 1000}
  46. // 启动多个goroutine同时操作账户
  47. for i := 0; i < 5; i++ {
  48. go func(id int) {
  49. account.Deposit(100)
  50. time.Sleep(10 * time.Millisecond)
  51. account.Withdraw(50)
  52. }(i)
  53. }
  55. time.Sleep(1 * time.Second)
  56. fmt.Printf("最终余额: %d\n", account.GetBalance())
  57. }
go
2. 缓存示例
  1. package main

  3. import (
  4. "fmt"
  5. "sync"
  6. "time"
  7. )

  9. type Cache struct {
  10. data map[string]interface{}
  11. mu sync.RWMutex // 读写锁，允许多个读取者
  12. }

  14. func NewCache() *Cache {
  15. return &Cache{
  16. data: make(map[string]interface{}),
  17. }
  18. }

  20. func (c *Cache) Set(key string, value interface{}) {
  21. c.mu.Lock()
  22. defer c.mu.Unlock()
  24. c.data[key] = value
  25. fmt.Printf("设置 %s = %v\n", key, value)
  26. }

  28. func (c *Cache) Get(key string) (interface{}, bool) {
  29. c.mu.RLock() // 读锁，允许多个goroutine同时读取
  30. defer c.mu.RUnlock()
  32. value, exists := c.data[key]
  33. if exists {
  34. fmt.Printf("获取 %s = %v\n", key, value)
  35. } else {
  36. fmt.Printf("键 %s 不存在\n", key)
  37. }
  38. return value, exists
  39. }

  41. func (c *Cache) Delete(key string) {
  42. c.mu.Lock()
  43. defer c.mu.Unlock()
  45. delete(c.data, key)
  46. fmt.Printf("删除键 %s\n", key)
  47. }

  49. func main() {
  50. cache := NewCache()
  52. // 启动多个读取者
  53. for i := 0; i < 3; i++ {
  54. go func(id int) {
  55. for j := 0; j < 5; j++ {
  56. cache.Get("key1")
  57. time.Sleep(50 * time.Millisecond)
  58. }
  59. }(i)
  60. }
  62. // 启动写入者
  63. go func() {
  64. for i := 0; i < 3; i++ {
  65. cache.Set("key1", fmt.Sprintf("value%d", i))
  66. time.Sleep(100 * time.Millisecond)
  67. }
  68. }()
  70. time.Sleep(1 * time.Second)
  71. }
go
3. 工作池示例
  1. package main

  3. import (
  4. "fmt"
  5. "sync"
  6. "time"
  7. )

  9. type WorkerPool struct {
  10. jobs chan int
  11. results chan int
  12. wg sync.WaitGroup
  13. }

  15. func NewWorkerPool(numWorkers int) *WorkerPool {
  16. return &WorkerPool{
  17. jobs: make(chan int, 100),
  18. results: make(chan int, 100),
  19. }
  20. }

  22. func (wp *WorkerPool) Start(numWorkers int) {
  23. for i := 0; i < numWorkers; i++ {
  24. wp.wg.Add(1)
  25. go wp.worker(i)
  26. }
  27. }

  29. func (wp *WorkerPool) worker(id int) {
  30. defer wp.wg.Done()
  32. for job := range wp.jobs {
  33. fmt.Printf("工作者 %d 处理任务 %d\n", id, job)
  34. time.Sleep(100 * time.Millisecond) // 模拟工作
  35. wp.results <- job * 2 // 返回结果
  36. }
  37. }

  39. func (wp *WorkerPool) AddJob(job int) {
  40. wp.jobs <- job
  41. }

  43. func (wp *WorkerPool) Close() {
  44. close(wp.jobs)
  45. wp.wg.Wait()
  46. close(wp.results)
  47. }

  49. func main() {
  50. pool := NewWorkerPool(3)
  51. pool.Start(3)
  53. // 添加任务
  54. for i := 1; i <= 10; i++ {
  55. pool.AddJob(i)
  56. }
  58. // 关闭工作池并等待完成
  59. go func() {
  60. pool.Close()
  61. }()
  63. // 收集结果
  64. for result := range pool.results {
  65. fmt.Printf("收到结果: %d\n", result)
  66. }
  67. }
go
4. 条件变量示例
  1. package main

  3. import (
  4. "fmt"
  5. "sync"
  6. "time"
  7. )

  9. type ProducerConsumer struct {
  10. buffer []int
  11. mu sync.Mutex
  12. cond *sync.Cond
  13. size int
  14. }

  16. func NewProducerConsumer(size int) *ProducerConsumer {
  17. pc := &ProducerConsumer{
  18. buffer: make([]int, 0, size),
  19. size: size,
  20. }
  21. pc.cond = sync.NewCond(&pc.mu)
  22. return pc
  23. }

  25. func (pc *ProducerConsumer) Produce(item int) {
  26. pc.mu.Lock()
  27. defer pc.mu.Unlock()
  29. // 等待缓冲区有空间
  30. for len(pc.buffer) >= pc.size {
  31. fmt.Printf("生产者等待，缓冲区已满\n")
  32. pc.cond.Wait()
  33. }
  35. pc.buffer = append(pc.buffer, item)
  36. fmt.Printf("生产: %d，缓冲区大小: %d\n", item, len(pc.buffer))
  38. // 通知消费者
  39. pc.cond.Signal()
  40. }

  42. func (pc *ProducerConsumer) Consume() int {
  43. pc.mu.Lock()
  44. defer pc.mu.Unlock()
  46. // 等待缓冲区有数据
  47. for len(pc.buffer) == 0 {
  48. fmt.Printf("消费者等待，缓冲区为空\n")
  49. pc.cond.Wait()
  50. }
  52. item := pc.buffer[0]
  53. pc.buffer = pc.buffer[1:]
  54. fmt.Printf("消费: %d，缓冲区大小: %d\n", item, len(pc.buffer))
  56. // 通知生产者
  57. pc.cond.Signal()
  59. return item
  60. }

  62. func main() {
  63. pc := NewProducerConsumer(3)
  65. // 启动生产者
  66. go func() {
  67. for i := 1; i <= 10; i++ {
  68. pc.Produce(i)
  69. time.Sleep(200 * time.Millisecond)
  70. }
  71. }()
  73. // 启动消费者
  74. go func() {
  75. for i := 0; i < 10; i++ {
  76. pc.Consume()
  77. time.Sleep(300 * time.Millisecond)
  78. }
  79. }()
  81. time.Sleep(5 * time.Second)
  82. }
go
最佳实践
1. 锁的粒度
  1. // 不好的做法：锁太大
  2. type BadExample struct {
  3. mu sync.Mutex
  4. data1 map[string]int
  5. data2 map[string]int
  6. }

  8. func (b *BadExample) UpdateData1(key string, value int) {
  9. b.mu.Lock()
  10. defer b.mu.Unlock()
  12. b.data1[key] = value // 只需要锁定data1
  13. // 但锁住了整个结构体
  14. }

  16. // 好的做法：细粒度锁
  17. type GoodExample struct {
  18. mu1 sync.Mutex
  19. mu2 sync.Mutex
  20. data1 map[string]int
  21. data2 map[string]int
  22. }

  24. func (g *GoodExample) UpdateData1(key string, value int) {
  25. g.mu1.Lock()
  26. defer g.mu1.Unlock()
  28. g.data1[key] = value // 只锁定需要的部分
  29. }
go
2. 避免死锁
  1. // 可能导致死锁的代码
  2. func (s *Service) Method1() {
  3. s.mu1.Lock()
  4. defer s.mu1.Unlock()
  6. s.Method2() // 可能尝试获取mu1，导致死锁
  7. }

  9. func (s *Service) Method2() {
  10. s.mu1.Lock() // 死锁！
  11. defer s.mu1.Unlock()
  12. }

  14. // 解决方案：使用sync.RWMutex或重新设计
  15. type Service struct {
  16. mu sync.RWMutex
  17. }

  19. func (s *Service) Method1() {
  20. s.mu.Lock()
  21. defer s.mu.Unlock()
  23. s.Method2() // 现在安全了
  24. }

  26. func (s *Service) Method2() {
  27. s.mu.RLock() // 读锁，不会死锁
  28. defer s.mu.RUnlock()
  29. }
go
3. 使用sync.Once
  1. package main

  3. import (
  4. "fmt"
  5. "sync"
  6. )

  8. type Singleton struct {
  9. data string
  10. }

  12. var (
  13. instance *Singleton
  14. once sync.Once
  15. )

  17. func GetInstance() *Singleton {
  18. once.Do(func() {
  19. instance = &Singleton{data: "单例数据"}
  20. fmt.Println("创建单例实例")
  21. })
  22. return instance
  23. }

  25. func main() {
  26. // 多个goroutine同时调用
  27. for i := 0; i < 5; i++ {
  28. go func(id int) {
  29. instance := GetInstance()
  30. fmt.Printf("Goroutine %d 获取实例: %v\n", id, instance)
  31. }(i)
  32. }
  34. // 等待所有goroutine完成
  35. time.Sleep(1 * time.Second)
  36. }
go
性能考虑
1. 锁竞争
  1. // 高竞争情况
  2. func HighContention() {
  3. var counter int
  4. var mu sync.Mutex
  6. for i := 0; i < 1000; i++ {
  7. go func() {
  8. mu.Lock()
  9. counter++
  10. mu.Unlock()
  11. }()
  12. }
  13. }

  15. // 减少竞争：使用原子操作
  16. import "sync/atomic"

  18. func LowContention() {
  19. var counter int64
  21. for i := 0; i < 1000; i++ {
  22. go func() {
  23. atomic.AddInt64(&counter, 1)
  24. }()
  25. }
  26. }
go
2. 读写锁优化
  1. type OptimizedCache struct {
  2. data map[string]interface{}
  3. mu sync.RWMutex
  4. }

  6. func (c *OptimizedCache) Get(key string) (interface{}, bool) {
  7. c.mu.RLock() // 多个读取者可以同时访问
  8. defer c.mu.RUnlock()
  10. return c.data[key], true
  11. }

  13. func (c *OptimizedCache) Set(key string, value interface{}) {
  14. c.mu.Lock() // 写入者独占访问
  15. defer c.mu.Unlock()
  17. c.data[key] = value
  18. }
go
调试技巧
1. 使用race detector
  1. go run -race your_program.go
bash
2. 添加调试信息
  1. type DebugMutex struct {
  2. sync.Mutex
  3. name string
  4. }

  6. func (dm *DebugMutex) Lock() {
  7. fmt.Printf("尝试锁定: %s\n", dm.name)
  8. dm.Mutex.Lock()
  9. fmt.Printf("已锁定: %s\n", dm.name)
  10. }

  12. func (dm *DebugMutex) Unlock() {
  13. fmt.Printf("解锁: %s\n", dm.name)
  14. dm.Mutex.Unlock()
  15. }
go