Golang pgx库嵌套调用Query()引发连接繁忙问题排查

错误原因 #

你遇到的“conn is busy”本质是：当前数据库连接被订单结果集rows占用着——在rows.Next()遍历过程中，驱动会锁定该连接用于读取订单数据，这时用同一个连接执行新的商品查询，就会触发连接冲突。要是提前关闭订单的rows，后续的遍历自然就没法继续了。

解决方案 #

方案1：用数据库连接池（最推荐） #

如果你的r.client是单个数据库连接（比如sql.Conn），换成sql.DB（连接池）。sql.DB会自动维护空闲连接池，每次调用Query时会分配新的空闲连接，不会和正在迭代的订单rows抢资源。

代码改动很小，只要确保r.client是*sql.DB实例即可：

rows, err := r.client.Query(ctx, q)
if err != nil {
    return nil, err
}
defer rows.Close()

orders := make([]Order, 0)
for rows.Next() {
    var ord Order
    // 扫描订单基础数据（原有代码不变）
    err = rows.Scan(
        &ord.OrderUID,
        &ord.TrackNumber,
        &ord.Entry,
        &ord.Delivery.Name,
        &ord.Delivery.Phone,
        &ord.Delivery.Zip,
        &ord.Delivery.City,
        &ord.Delivery.Address,
        &ord.Delivery.Region,
        &ord.Delivery.Email,
        &ord.Payment.Transaction,
        &ord.Payment.RequestID,
        &ord.Payment.Currency,
        &ord.Payment.Provider,
        &ord.Payment.Amount,
        &ord.Payment.PaymentDT,
        &ord.Payment.Bank,
        &ord.Payment.DeliveryCost,
        &ord.Payment.GoodsTotal,
        &ord.Payment.CustomFee,
        &ord.Locale,
        &ord.InternalSignature,
        &ord.CustomerID,
        &ord.DeliveryService,
        &ord.ShardKey,
        &ord.SmID,
        &ord.DateCreated,
        &ord.OofShard,
    )
    if err != nil {
        return nil, err
    }

    // 查询当前订单的商品，连接池会自动分配空闲连接
    iq := `SELECT id, chrt_id, track_number, price, rid, name, sale, size, total_price, nm_id, brand, status FROM items WHERE order_uid = $1`
    itemRows, err := r.client.Query(ctx, iq, ord.OrderUID)
    if err != nil {
        return nil, err
    }
    // 记得关闭商品结果集，避免连接泄漏
    defer itemRows.Close()

    items := make([]item.Item, 0)
    for itemRows.Next() {
        var item item.Item
        err = itemRows.Scan(
            &item.ID,
            &item.ChrtID,
            &item.TrackNumber,
            &item.Price,
            &item.Rid,
            &item.Name,
            &item.Sale,
            &item.Size,
            &item.TotalPrice,
            &item.NmID,
            &item.Brand,
            &item.Status,
        )
        if err != nil {
            return nil, err
        }
        items = append(items, item)
    }
    // 检查商品遍历过程中的错误
    if err = itemRows.Err(); err != nil {
        return nil, err
    }

    ord.Items = items
    orders = append(orders, ord)
}
// 检查订单遍历过程中的错误
if err = rows.Err(); err != nil {
    return nil, err
}
return orders, nil

方案2：先加载所有订单到内存，再批量查商品 #

如果想减少数据库请求次数（尤其订单多的时候），可以先把所有订单的基础信息读进内存，关闭订单rows后，一次性批量查询所有商品，最后通过OrderUID映射到对应订单：

rows, err := r.client.Query(ctx, q)
if err != nil {
    return nil, err
}
defer rows.Close()

orders := make([]Order, 0)
orderUIDs := make([]string, 0)
// 第一步：读取所有订单基础数据
for rows.Next() {
    var ord Order
    err = rows.Scan(
        &ord.OrderUID,
        &ord.TrackNumber,
        &ord.Entry,
        &ord.Delivery.Name,
        &ord.Delivery.Phone,
        &ord.Delivery.Zip,
        &ord.Delivery.City,
        &ord.Delivery.Address,
        &ord.Delivery.Region,
        &ord.Delivery.Email,
        &ord.Payment.Transaction,
        &ord.Payment.RequestID,
        &ord.Payment.Currency,
        &ord.Payment.Provider,
        &ord.Payment.Amount,
        &ord.Payment.PaymentDT,
        &ord.Payment.Bank,
        &ord.Payment.DeliveryCost,
        &ord.Payment.GoodsTotal,
        &ord.Payment.CustomFee,
        &ord.Locale,
        &ord.InternalSignature,
        &ord.CustomerID,
        &ord.DeliveryService,
        &ord.ShardKey,
        &ord.SmID,
        &ord.DateCreated,
        &ord.OofShard,
    )
    if err != nil {
        return nil, err
    }
    orders = append(orders, ord)
    orderUIDs = append(orderUIDs, ord.OrderUID)
}
if err = rows.Err(); err != nil {
    return nil, err
}

// 第二步：批量查询所有商品（用ANY子句，注意数据库对参数数量的限制）
iq := `SELECT id, chrt_id, track_number, price, rid, name, sale, size, total_price, nm_id, brand, status, order_uid FROM items WHERE order_uid = ANY($1)`
itemRows, err := r.client.Query(ctx, iq, orderUIDs)
if err != nil {
    return nil, err
}
defer itemRows.Close()

// 第三步：把商品按OrderUID分组
itemMap := make(map[string][]item.Item)
for itemRows.Next() {
    var item item.Item
    var orderUID string
    err = itemRows.Scan(
        &item.ID,
        &item.ChrtID,
        &item.TrackNumber,
        &item.Price,
        &item.Rid,
        &item.Name,
        &item.Sale,
        &item.Size,
        &item.TotalPrice,
        &item.NmID,
        &item.Brand,
        &item.Status,
        &orderUID, // 多扫一个order_uid用于分组
    )
    if err != nil {
        return nil, err
    }
    itemMap[orderUID] = append(itemMap[orderUID], item)
}
if err = itemRows.Err(); err != nil {
    return nil, err
}

// 第四步：将商品映射到对应订单
for i := range orders {
    orders[i].Items = itemMap[orders[i].OrderUID]
}

return orders, nil

方案3：循环内临时获取独立连接 #

如果必须用单个连接，每次查商品时临时获取新连接：

rows, err := r.client.Query(ctx, q)
if err != nil {
    return nil, err
}
defer rows.Close()

orders := make([]Order, 0)
for rows.Next() {
    var ord Order
    // 扫描订单数据（原有代码不变）
    err = rows.Scan(
        &ord.OrderUID,
        &ord.TrackNumber,
        &ord.Entry,
        &ord.Delivery.Name,
        &ord.Delivery.Phone,
        &ord.Delivery.Zip,
        &ord.Delivery.City,
        &ord.Delivery.Address,
        &ord.Delivery.Region,
        &ord.Delivery.Email,
        &ord.Payment.Transaction,
        &ord.Payment.RequestID,
        &ord.Payment.Currency,
        &ord.Payment.Provider,
        &ord.Payment.Amount,
        &ord.Payment.PaymentDT,
        &ord.Payment.Bank,
        &ord.Payment.DeliveryCost,
        &ord.Payment.GoodsTotal,
        &ord.Payment.CustomFee,
        &ord.Locale,
        &ord.InternalSignature,
        &ord.CustomerID,
        &ord.DeliveryService,
        &ord.ShardKey,
        &ord.SmID,
        &ord.DateCreated,
        &ord.OofShard,
    )
    if err != nil {
        return nil, err
    }

    // 临时获取新连接查询商品
    conn, err := r.client.Conn(ctx)
    if err != nil {
        return nil, err
    }
    // 用完立即关闭连接
    defer conn.Close()

    iq := `SELECT id, chrt_id, track_number, price, rid, name, sale, size, total_price, nm_id, brand, status FROM items WHERE order_uid = $1`
    itemRows, err := conn.QueryContext(ctx, iq, ord.OrderUID)
    if err != nil {
        return nil, err
    }
    defer itemRows.Close()

    // 遍历商品数据（原有代码不变）
    items := make([]item.Item, 0)
    for itemRows.Next() {
        var item item.Item
        err = itemRows.Scan(
            &item.ID,
            &item.ChrtID,
            &item.TrackNumber,
            &item.Price,
            &item.Rid,
            &item.Name,
            &item.Sale,
            &item.Size,
            &item.TotalPrice,
            &item.NmID,
            &item.Brand,
            &item.Status,
        )
        if err != nil {
            return nil, err
        }
        items = append(items, item)
    }

    ord.Items = items
    orders = append(orders, ord)
}
return orders, nil

注意事项 #

• 方案1是最优解，连接池自动管理连接，代码改动最小，性能稳定。
• 方案2能把N次商品查询缩减为1次，适合订单量较大的场景，性能更优。
• 所有结果集（rows、itemRows）都要记得关闭，避免连接泄漏。
• 一定要检查rows.Err()，不要遗漏遍历过程中出现的错误。

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