Go-10-回调、闭包与并发入门

字数统计: 1.7k字   |   阅读时长≈ 8分

Go-10-回调、闭包与并发入门

本节目标：理解 Go 的并发模型（CSP），掌握 goroutine + channel + select，理解”用通信共享内存而不是用内存通信”。

1. goroutine：轻量级线程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
func say(s string) {
    for i := 0; i < 3; i++ {
        time.Sleep(100 * time.Millisecond)
        fmt.Println(s)
    }
}

func main() {
    go say("world")  // 启动 goroutine
    say("hello")      // 主 goroutine
}
// 输出（顺序不保证）：
// hello
// world
// hello
// ...

关键事实：

  • goroutine 由 Go runtime 管理，不是 OS 线程
  • 初始栈 2KB（可动态增长），可以轻松开上百万个
  • go 关键字启动一个 goroutine，立即返回，不等待

主 goroutine 退出时，所有子 goroutine 立即终止（程序退出）：

1
2
3
4
5
6
7
func main() {
    go func() {
        time.Sleep(time.Second)
        fmt.Println("子 goroutine")
    }()
    // 立即退出，子 goroutine 不会执行
}

同步等待：

1
2
3
4
5
6
7
8
9
10
var wg sync.WaitGroup

for i := 0; i < 3; i++ {
    wg.Add(1)
    go func(id int) {
        defer wg.Done()
        fmt.Printf("worker %d done\n", id)
    }(i)
}
wg.Wait()  // 阻塞直到 wg 计数为 0

2. channel：goroutine 之间的通信

channel 是类型化的管道，用 make 创建：

1
2
3
4
5
6
7
8
9
ch := make(chan int)     // 无缓冲 channel
ch := make(chan int, 5) // 缓冲 5 个元素的 channel

// 发送
ch <- 42

// 接收
v := <-ch
v, ok := <-ch  // ok = false 表示 channel 已关闭

2.1 无缓冲 channel：同步

无缓冲 = 发送方阻塞直到接收方准备好：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
ch := make(chan int)

go func() {
    fmt.Println("before send")
    ch <- 1                // 阻塞，直到有接收方
    fmt.Println("after send")
}()

time.Sleep(time.Second)    // 让 goroutine 先跑到 send
fmt.Println("before recv")
v := <-ch                  // 接收到，sender 才解锁
fmt.Println("received:", v)
// 输出：
// before send
// before recv
// received: 1
// after send

2.2 缓冲 channel：异步

缓冲 channel 发送方在缓冲区未满时不会阻塞：

1
2
3
4
ch := make(chan int, 2)
ch <- 1  // 不阻塞
ch <- 2  // 不阻塞
ch <- 3  // 阻塞！缓冲区满

2.3 关闭 channel

发送方 close，接收方检测关闭：

1
2
3
4
5
6
7
8
9
10
11
12
ch := make(chan int)

go func() {
    for i := 0; i < 3; i++ {
        ch <- i
    }
    close(ch)  // 关闭 channel（只能关闭一次）
}()

for v := range ch {  // 不断接收直到 channel 关闭
    fmt.Println(v)   // 0 1 2
}

关闭原则：

  • 关闭后不能再发送（会 panic）
  • 接收方不会因关闭而阻塞，可以继续接收剩余值
  • 不要从接收方关闭 channel（容易 panic）
  • 不要关闭已关闭的 channel

3. select：多路复用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
ch1 := make(chan string)
ch2 := make(chan string)

go func() {
    time.Sleep(100 * time.Millisecond)
    ch1 <- "from ch1"
}()
go func() {
    time.Sleep(200 * time.Millisecond)
    ch2 <- "from ch2"
}()

for i := 0; i < 2; i++ {
    select {
    case msg := <-ch1:
        fmt.Println(msg)
    case msg := <-ch2:
        fmt.Println(msg)
    case <-time.After(50 * time.Millisecond):
        fmt.Println("timeout")
    }
}

select 的关键特性：

  • 多个 case 同时就绪时随机选一个（避免饥饿）
  • default 让 select 变成”非阻塞”
  • time.After(d) 在 d 时间后发送一个值，常用于超时

4. 实战：并发安全的工作池

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
type Job struct {
    ID   int
    Data string
}

func worker(id int, jobs <-chan Job, results chan<- string) {
    for job := range jobs {
        time.Sleep(100 * time.Millisecond)  // 模拟工作
        results <- fmt.Sprintf("worker %d processed job %d: %s", id, job.ID, job.Data)
    }
}

func main() {
    jobs := make(chan Job, 10)
    results := make(chan string, 10)

    // 启动 3 个 worker
    for w := 1; w <= 3; w++ {
        go worker(w, jobs, results)
    }

    // 派发 5 个 job
    for j := 1; j <= 5; j++ {
        jobs <- Job{ID: j, Data: fmt.Sprintf("data-%d", j)}
    }
    close(jobs)

    // 收集结果
    for r := 0; r < 5; r++ {
        fmt.Println(<-results)
    }
}

5. 实战：超时控制

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
func fetchWithTimeout(url string, timeout time.Duration) (string, error) {
    result := make(chan string, 1)
    errCh := make(chan error, 1)

    go func() {
        resp, err := http.Get(url)
        if err != nil {
            errCh <- err
            return
        }
        defer resp.Body.Close()
        body, _ := io.ReadAll(resp.Body)
        result <- string(body)
    }()

    select {
    case body := <-result:
        return body, nil
    case err := <-errCh:
        return "", err
    case <-time.After(timeout):
        return "", fmt.Errorf("timeout after %v", timeout)
    }
}

6. 实战：done channel 模式（取消）

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
func doWork(done <-chan struct{}) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for i := 0; ; i++ {
            select {
            case <-done:
                return  // 被取消
            case out <- i:
                time.Sleep(100 * time.Millisecond)
            }
        }
    }()
    return out
}

func main() {
    done := make(chan struct{})
    out := doWork(done)

    for i := 0; i < 3; i++ {
        fmt.Println(<-out)
    }
    close(done)  // 通知 goroutine 退出
    // 给 goroutine 时间退出
    time.Sleep(100 * time.Millisecond)
}

7. CSP 模型：通信顺序进程

Go 的并发哲学：

“不要通过共享内存来通信；而要通过通信来共享内存。”

传统方式（共享内存 + 锁）：

1
2
3
4
5
6
7
8
var counter int
var mu sync.Mutex

func increment() {
    mu.Lock()
    counter++
    mu.Unlock()
}

Go 风格（通过 channel 通信）：

1
2
3
4
5
6
7
8
9
10
11
12
counter := make(chan int)

func increment() {
    n := <-counter  // 接收当前值
    counter <- n + 1  // 发送新值
}

func main() {
    counter <- 0
    go increment()
    fmt.Println(<-counter)
}

两种方式都有效，但 channel 风格：

  • 数据所有权明确（谁发送谁拥有）
  • 避免竞态条件
  • 更容易组合（多个 channel 通过 select 复用）

8. 并发陷阱

8.1 死锁

1
2
ch := make(chan int)
ch <- 1  // 死锁！没有接收方

8.2 竞态条件（race condition）

1
2
3
4
5
6
7
8
var counter int

for i := 0; i < 1000; i++ {
    go func() {
        counter++  // 多个 goroutine 同时写，未保护
    }()
}
// counter 值不确定

检测竞态：go test -race 或 go run -race main.go

8.3 goroutine 泄漏

1
2
3
4
5
6
7
8
func leak() {
    ch := make(chan int)
    go func() {
        val := <-ch  // 永远阻塞
        fmt.Println(val)
    }()
    // goroutine 永远不会退出
}

避免：用 context、超时、done channel 通知退出。

9. 并发 Map（sync.Map）

普通 map 多 goroutine 写会 panic，用 sync.Map 或加锁：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
var m sync.Map

m.Store("key1", "value1")
m.Store("key2", "value2")

if v, ok := m.Load("key1"); ok {
    fmt.Println(v.(string))
}

m.Range(func(k, v interface{}) bool {
    fmt.Println(k, v)
    return true  // 继续遍历
})

m.Delete("key1")

或者用 map + sync.RWMutex：

1
2
3
4
5
6
7
8
9
10
11
type SafeMap struct {
    mu sync.RWMutex
    m  map[string]int
}

func (s *SafeMap) Get(k string) (int, bool) {
    s.mu.RLock()
    defer s.mu.RUnlock()
    v, ok := s.m[k]
    return v, ok
}

10. WaitGroup 详解

1
2
3
4
5
6
7
8
9
var wg sync.WaitGroup

wg.Add(1)       // 计数 +1
go func() {
    defer wg.Done()  // 计数 -1（在 goroutine 退出时）
    // do work
}()

wg.Wait()       // 阻塞直到计数 = 0

注意：

  • wg.Add 必须在 go 语句之前调用
  • 计数不能为负（Done 比 Add 多会 panic）

小结

概念 用法
go func() 启动 goroutine
ch := make(chan T) 无缓冲 channel（同步）
ch := make(chan T, n) 缓冲 channel（异步）
ch <- v / <-ch 发送/接收
close(ch) 关闭 channel
for v := range ch 接收直到关闭
select 多 channel 复用
sync.WaitGroup 等多个 goroutine
sync.Mutex / RWMutex 互斥锁
sync.Map 并发安全的 map
time.After / time.Tick 超时/定时

最佳实践：

  1. 能用 channel 就别用共享内存（但也别走极端）
  2. 永远关心 goroutine 退出（避免泄漏）
  3. 用 -race 跑测试
  4. buffered channel 的缓冲大小要算清楚
  5. 不要在不知道对方会如何关闭时 close channel

下一节进入并发深入：sync 包、context、连接池。

赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

主要涉及技术：
Java后端开发、Linux开发、
公众号开发、开源爱好者、Linux

联系QQ:562054870