如何改进简易RabbitMQ连接池，实现单连接阻塞而非全局阻塞？

当然可以改造。你的代码用了全局sync.Mutex，每次调用GetConnection都会锁住整个连接池，一旦某个连接断开需要重建，所有请求都得等锁释放，这就是全局阻塞的根源。要改成单连接阻塞，核心是给每个连接单独加锁，让连接的检查和重建只占用自身的锁，不影响其他连接的使用。

改造后的代码实现 #

import (
    "sync"
    "sync/atomic"
    "github.com/streadway/amqp"
)

// 单个连接的封装结构体，包含连接、专属锁和重建标记
type PoolConn struct {
    conn          *amqp.Connection
    mu            sync.Mutex
    isReconnecting bool // 标记是否正在重建，避免重复发起重建请求
}

type RabbitMQConnectionPool struct {
    counter    uint32 // 用原子类型维护轮询计数，无需全局锁
    connections []*PoolConn
    connString string
}

func (p *RabbitMQConnectionPool) GetConnection() (*amqp.Connection, int) {
    // 原子递增计数并取模，实现无锁的连接轮选
    idx := int(atomic.AddUint32(&p.counter, 1) % uint32(len(p.connections)))
    poolConn := p.connections[idx]

    // 仅锁定当前连接的专属锁，而非整个连接池
    poolConn.mu.Lock()
    defer poolConn.mu.Unlock()

    // 检查连接是否可用
    if poolConn.conn != nil && !poolConn.conn.IsClosed() {
        return poolConn.conn, idx
    }

    // 若已有goroutine在重建该连接，直接等待锁释放后返回新连接
    if poolConn.isReconnecting {
        return poolConn.conn, idx
    }

    // 标记开始重建
    poolConn.isReconnecting = true
    defer func() {
        poolConn.isReconnecting = false
    }()

    // 循环尝试重建连接
    for {
        conn, err := amqp.Dial(p.connString)
        if err != nil {
            vibelog.Stdout.Error("67eea0ee-ae1a-436b-8490-c6f232e6e033", err)
            // 可添加短暂延迟避免高频报错
            // time.Sleep(100 * time.Millisecond)
            continue
        }
        poolConn.conn = conn
        return conn, idx
    }
}

关键改动说明 #

• 缩小锁粒度：把全局锁替换为每个连接的专属锁，操作单个连接时只会阻塞请求该连接的调用者，其他连接的请求不受影响。
• 原子计数轮询：用atomic.Uint32维护轮询计数器，无需全局锁就能实现安全的连接轮选逻辑。
• 重建冲突避免：添加isReconnecting标记，防止多个goroutine同时重建同一个连接，避免不必要的资源消耗。

改造后，只有当某个连接需要重建时，请求该连接的调用者会被阻塞，其他连接的请求可以正常处理，实现了单连接级别的阻塞而非全局阻塞。

内容的提问来源于stack exchange，提问作者Alexander Mills