Go-13-并发深入：sync、context 与连接池

字数统计: 1.9k字   |   阅读时长≈ 9分

Go-13-并发深入：sync、context 与连接池

本节目标：掌握 Go 高级并发原语：sync.Mutex/RWMutex/Once/Pool，context 包做超时和取消，以及连接池模式。

1. sync.Mutex（互斥锁）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
type Counter struct {
    mu    sync.Mutex
    count int
}

func (c *Counter) Inc() {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.count++
}

func (c *Counter) Value() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.count
}

**互斥锁的”金科玉律”**：

  • Lock 和 Unlock 必须成对出现
  • 用 defer Unlock() 保证一定会释放
  • 不要在持锁时调用可能阻塞的函数（如 I/O、channel 操作）
  • 不要在持锁时再次请求同一把锁（死锁）

复制陷阱：sync.Mutex 不能被复制（包含状态）：

1
2
3
4
5
6
7
8
9
10
// 错：
type Counter struct {
    mu sync.Mutex  // ← 复制 Counter 会复制 mu
}

c1 := Counter{}
c2 := c1  // 复制，c1.mu 和 c2.mu 独立 ← 死锁风险

// 对：永远用指针传递
func process(c *Counter) { c.Inc() }

2. sync.RWMutex（读写锁）

多个读可以并发，写是排他的：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
type Cache struct {
    mu   sync.RWMutex
    data map[string]string
}

func (c *Cache) Get(k string) (string, bool) {
    c.mu.RLock()         // 读锁（可多个并发）
    defer c.mu.RUnlock()
    v, ok := c.data[k]
    return v, ok
}

func (c *Cache) Set(k, v string) {
    c.mu.Lock()          // 写锁（排他）
    defer c.mu.Unlock()
    c.data[k] = v
}

何时用 RWMutex：

  • 读多写少（读占比 > 90%）
  • 读操作明显比写操作耗时
  • 否则用 sync.Mutex 即可（RWMutex 自身开销略大）

3. sync.Once（一次性初始化）

保证函数只执行一次，常用于单例模式：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
var (
    instance *Database
    once     sync.Once
)

func GetDB() *Database {
    once.Do(func() {
        fmt.Println("初始化数据库连接")
        instance = &Database{...}
    })
    return instance
}

// 多次调用只初始化一次
GetDB()
GetDB()  // 不再打印

Once 的内部实现：

1
2
3
4
5
6
7
8
9
10
type Once struct {
    done uint32
    m    Mutex
}

func (o *Once) Do(f func()) {
    if atomic.LoadUint32(&o.done) == 0 {
        o.doSlow(f)
    }
}

4. sync.Pool（对象池）

复用对象，减少 GC 压力：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
var bufPool = sync.Pool{
    New: func() interface{} {
        return new(bytes.Buffer)
    },
}

func process(data []byte) string {
    buf := bufPool.Get().(*bytes.Buffer)
    defer func() {
        buf.Reset()        // 清空
        bufPool.Put(buf)   // 放回池里
    }()

    buf.Write(data)
    // 处理
    return buf.String()
}

Pool 的特性：

  • GC 时会被清空（不保证持久）
  • 适合短生命周期高频创建的对象（如 bytes.Buffer、临时切片）
  • 不适合做连接池（连接需要持久）

5. atomic 包（无锁编程）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import "sync/atomic"

var counter int64

// 原子加
atomic.AddInt64(&counter, 1)

// 原子读
v := atomic.LoadInt64(&counter)

// 原子写
atomic.StoreInt64(&counter, 100)

// CAS（Compare-And-Swap）
ok := atomic.CompareAndSwapInt64(&counter, 100, 200)

适用场景：简单的计数器、标志位。
不适用：复合操作（需要事务语义）。

6. context：超时和取消

context 是 Go 的超时和取消机制的标准：

6.1 三种 context

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// 根 context
ctx := context.Background()

// 带取消
ctx, cancel := context.WithCancel(ctx)
defer cancel()

// 带超时
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()

// 带截止时间
deadline := time.Now().Add(5 * time.Second)
ctx, cancel := context.WithDeadline(ctx, deadline)
defer cancel()

6.2 传递值

1
2
3
4
5
6
ctx := context.WithValue(context.Background(), "userID", 42)
ctx = context.WithValue(ctx, "requestID", "abc123")

// 取出
userID := ctx.Value("userID").(int)
requestID := ctx.Value("requestID").(string)

注意：

  • Value 应该用自定义类型作为 key（避免冲突）：
    1
    2
    3
    type ctxKey int
    const userIDKey ctxKey = iota
    ctx := context.WithValue(ctx, userIDKey, 42)

6.3 实战：HTTP 请求带超时

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
func fetch(ctx context.Context, url string) (string, error) {
    req, _ := http.NewRequestWithContext(ctx, "GET", url, nil)
    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        return "", err
    }
    defer resp.Body.Close()
    body, _ := io.ReadAll(resp.Body)
    return string(body), nil
}

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
    defer cancel()
    
    body, err := fetch(ctx, "https://example.com")
    if err != nil {
        if errors.Is(err, context.DeadlineExceeded) {
            fmt.Println("请求超时")
        } else {
            fmt.Println("请求失败:", err)
        }
        return
    }
    fmt.Println(body)
}

6.4 实战：goroutine 链 + 取消传播

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
func worker(ctx context.Context, id int) {
    for {
        select {
        case <-ctx.Done():
            fmt.Printf("worker %d 退出: %v\n", id, ctx.Err())
            return
        case <-time.After(time.Second):
            fmt.Printf("worker %d 工作中...\n", id)
        }
    }
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    
    for i := 1; i <= 3; i++ {
        go worker(ctx, i)
    }
    
    time.Sleep(3 * time.Second)
    cancel()  // 通知所有 worker 退出
    time.Sleep(100 * time.Millisecond)
}

7. errgroup：多 goroutine 错误聚合

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
import "golang.org/x/sync/errgroup"

func main() {
    var g errgroup.Group
    
    urls := []string{"https://a.com", "https://b.com", "https://c.com"}
    results := make([]string, len(urls))
    
    for i, url := range urls {
        i, url := i, url
        g.Go(func() error {
            resp, err := http.Get(url)
            if err != nil {
                return err
            }
            defer resp.Body.Close()
            body, _ := io.ReadAll(resp.Body)
            results[i] = string(body)
            return nil
        })
    }
    
    if err := g.Wait(); err != nil {
        log.Fatal(err)
    }
    
    for _, r := range results {
        fmt.Println(r)
    }
}

特性：

  • 任意一个 goroutine 报错，其他都立即停止
  • 第一个错误被返回
  • 内部用 context.WithCancel 实现

8. 连接池实战

Go 标准库没有通用连接池，但有现成模式：

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
51
52
53
type Conn struct {
    addr string
    // 实际连接（如 net.Conn、*sql.DB、redis client）
}

type Pool struct {
    mu      sync.Mutex
    conns   []*Conn
    cap     int
    factory func(addr string) (*Conn, error)
    closed  bool
}

func NewPool(cap int, factory func(addr string) (*Conn, error)) *Pool {
    return &Pool{
        cap:     cap,
        factory:  factory,
        conns:   make([]*Conn, 0, cap),
    }
}

func (p *Pool) Get(addr string) (*Conn, error) {
    p.mu.Lock()
    if n := len(p.conns); n > 0 {
        conn := p.conns[n-1]
        p.conns = p.conns[:n-1]
        p.mu.Unlock()
        return conn, nil
    }
    p.mu.Unlock()
    return p.factory(addr)
}

func (p *Pool) Put(conn *Conn) {
    p.mu.Lock()
    defer p.mu.Unlock()
    if p.closed || len(p.conns) >= p.cap {
        // 池已关闭或已满，关闭连接
        conn.Close()
        return
    }
    p.conns = append(p.conns, conn)
}

func (p *Pool) Close() {
    p.mu.Lock()
    defer p.mu.Unlock()
    p.closed = true
    for _, c := range p.conns {
        c.Close()
    }
    p.conns = nil
}

生产用：

  • HTTP 连接池：http.Client 内部自动管理
  • 数据库连接池：*sql.DB 自动管理
  • Redis：github.com/redis/go-redis/v9 客户端池
  • 通用池：github.com/jolestar/go-commons-pool 或 github.com/silenceper/pool

9. 实战：fan-out / fan-in 模式

扇出（多个 worker 并行处理）→ 扇入（合并结果）：

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
func fanOut(in <-chan int, workers int) []<-chan int {
    channels := make([]<-chan int, workers)
    for i := 0; i < workers; i++ {
        ch := make(chan int)
        channels[i] = ch
        go func(ch chan int) {
            defer close(ch)
            for v := range in {
                ch <- process(v)
            }
        }(ch)
    }
    return channels
}

func fanIn(channels ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    for _, ch := range channels {
        ch := ch
        wg.Add(1)
        go func() {
            defer wg.Done()
            for v := range ch {
                out <- v
            }
        }()
    }
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

// 用法
in := make(chan int)
go func() {
    for i := 0; i < 10; i++ {
        in <- i
    }
    close(in)
}()

workers := fanOut(in, 3)
out := fanIn(workers...)
for v := range out {
    fmt.Println(v)
}

10. 限流（rate limit）

1
2
3
4
5
6
7
8
9
10
11
12
import "golang.org/x/time/rate"

// 每秒 10 个，最多突发 5 个
limiter := rate.NewLimiter(rate.Limit(10), 5)

for i := 0; i < 20; i++ {
    if err := limiter.Wait(ctx); err != nil {
        log.Fatal(err)
    }
    // 处理请求
    fmt.Println(i)
}

令牌桶算法：

  • rate.Limit(10)：每秒 10 个令牌
  • burst=5：桶容量 5（允许短时间突发）
  • Wait(ctx)：阻塞直到拿到令牌（带超时）
  • Allow()：非阻塞，返回 bool

11. 最佳实践

  1. 优先用 channel 通信，必要时用 mutex
  2. 每个公共函数第一个参数是 ctx（如果是可能被取消的）
  3. goroutine 数量 = CPU 核数 × 2~10（I/O 密集型）
  4. 永远考虑 goroutine 退出路径
  5. 用 -race 跑测试
  6. 池化高频短生命周期对象（bytes.Buffer、临时切片）
  7. 生产用 errgroup 替代裸 WaitGroup（带 ctx 取消）

小结

同步原语 用途
sync.Mutex 互斥
sync.RWMutex 读多写少
sync.Once 一次性初始化
sync.Pool 对象复用
atomic 简单原子操作
context 超时/取消/传值
errgroup 多 goroutine 错误聚合
rate.Limiter 限流

下一节讲反射（reflect）和元数据编程。

赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

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