如何在多goroutine中运行Postgres Tx，解决pq连接相关报错问题

问题根因 #

1. 事务/连接并发不安全：pq驱动的单个数据库连接、事务实例tx不支持跨goroutine并发操作，多个goroutine共用同一个tx或底层连接会导致协议响应乱序，触发pq: unexpected Parse response 'D'类错误。
2. 提前关闭连接池：你当前的错误处理逻辑是只要收到任意一个goroutine返回的错误就立即return，此时defer odooDB.Close()会直接关闭整个数据库连接池，另一个还在运行的goroutine操作数据库时就会拿到已关闭的连接，触发pq: bad connection错误。
3. wg.Done位置错误：你把defer wg.Done()写在了goroutine逻辑的末尾，如果SaveFromOdoo执行过程中发生panic，wg.Done()永远不会被调用，会导致负责关闭errs channel的goroutine永久阻塞，造成资源泄漏。

修复方案 #

1. 调整错误处理逻辑，避免提前关闭连接池 #

必须等所有goroutine都执行完成后再关闭数据库连接池，不能拿到第一个错误就直接返回。你可以收集所有错误，最后返回任意一个非空错误即可。

2. 修正sync.WaitGroup的使用姿势 #

把defer wg.Done()放在每个业务goroutine的第一行，确保无论函数执行成功、报错还是panic，都会正确计数。

3. 保证事务隔离性 #

如果你的两个SaveFromOdoo方法内部都需要开启事务，确保每个方法内部自己独立从odooDB连接池获取连接、开启独立的tx，不要把tx实例传到goroutine间共享。如果两个方法的操作需要保证事务一致性，就不要拆分到不同goroutine执行，跨goroutine的分布式事务逻辑复杂度极高，没必要的话就串行执行。

修复后的参考代码 #

func (uc OdooUseCase) SynchronizeTourPackage(tourPackageID, partnerID string, odooPackageID int64) (err error) {
    var wg sync.WaitGroup
    var mu sync.Mutex // 用来保护err变量避免并发写

    odooUc := OdooUseCase{UcContract: uc.UcContract}
    odooDB, err := odooUc.BuildConnection(partnerID)
    if err != nil {
        logruslogger.Log(logruslogger.WarnLevel, err.Error(), functioncaller.PrintFuncName(), "odoo-build-connection")
        return err
    }
    defer odooDB.Close()

    tourPackagePriceUseCase := TourPackagePriceUseCase{UcContract: uc.UcContract, odooDB: odooDB}
    tourPackageMealUseCase := TourPackageMealUseCase{UcContract: uc.UcContract, odooDB: odooDB}

    wg.Add(2) // 直接加2个业务goroutine的计数，不需要额外的监听goroutine
    // 处理套餐价格
    go func() {
        defer wg.Done() // 放第一行保证必执行
        saveErr := tourPackagePriceUseCase.SaveFromOdoo(tourPackageID, partnerID, odooPackageID)
        if saveErr != nil {
            logruslogger.Log(logruslogger.WarnLevel, saveErr.Error(), functioncaller.PrintFuncName(), "save-tour-package-price", uc.ReqID)
            mu.Lock()
            if err == nil { // 只保留第一个错误
                err = saveErr
            }
            mu.Unlock()
        }
    }()
    // 处理套餐餐食
    go func() {
        defer wg.Done() // 放第一行保证必执行
        saveErr := tourPackageMealUseCase.SaveFromOdoo(tourPackageID, partnerID, odooPackageID)
        if saveErr != nil {
            logruslogger.Log(logruslogger.WarnLevel, saveErr.Error(), functioncaller.PrintFuncName(), "save-tour-package-meal", uc.ReqID)
            mu.Lock()
            if err == nil {
                err = saveErr
            }
            mu.Unlock()
        }
    }()

    wg.Wait() // 等所有goroutine都执行完再返回
    return err
}

额外注意点 #

• 确认BuildConnection返回的是标准库的*sql.DB连接池实例，而不是单个底层连接。*sql.DB本身是线程安全的，可以在多个goroutine间共享，内部会自动调度连接，只要每个操作或者事务单独从池里拿连接就行。
• 如果你的SaveFromOdoo方法内部确实需要共用同一个事务，就不要拆分到两个goroutine执行，改成串行执行两个方法即可，避免并发操作事务的问题。

内容的提问来源于stack exchange，提问作者Yonathan Pamungkas Putra