Golang协程中Spanner只读事务查询渐慢问题排查与解决

尝试从Spanner表中查询约10000行数据，试过limit、offset等方案未达预期，转而用每个goroutine查询单行。原本预期单行查询耗时~5ms，但批量1万行查询耗时超20s。

简化代码 #

func queryEmp(IDs[]string, spannerClient *spanner.Client) (Employee,error){
query := "Select name from Employee Where id = @id"

    g, gCtx := errgroup.WithContext(ctx)
    for _, ID := range IDs {
        id := ID
        g.Go(func() error {
    
            tx := spannerClient.Single() 
            defer tx.Close()

            stmt2 := spanner.NewStatement(query)
            stmt2.Params = map[string]interface{}{
                "ID": id,
            }

            qstart := time.Now()
            it := tx.Query(gCtx, stmt2)
            defer it.Stop()
            logrus.Debugf("%s took %v \n", "query execution.", time.Since(qstart))

            for {
                row, err := it.Next()
                if err == iterator.Done {
                    break
                }
                if err != nil {
                    return err
                }

                var eID string
                if err := row.Column(0, &eID); err != nil {
                    return err
                }

            }

            return nil
        })
    }
    err = g.Wait()
}

查询Trace日志 #

{"message":"query execution. took 39.677µs \n","severity":"debug","time":"2023-11-03T20:51:29-04:00"}
{"message":"query execution. took 34.125µs \n","severity":"debug","time":"2023-11-03T20:51:29-04:00"}
...
{"message":"query execution. took 6.779527006s \n","severity":"debug","time":"2023-11-03T20:52:00-04:00"}

查询初始速度符合预期，但耗时持续增加。spannerClient的MaxSessions与MinSessions均设为100，按预期仅在超过100后出现轻微变慢，但实际并非如此。

Google文档说明：

Sessions can execute only one transaction at a time. Standalone reads, writes, and queries use a transaction internally, and count toward the one transaction limit.

使用非迭代查询（如ReadRow）也出现相同结果，将tx := spannerClient.Single()放在循环外也无改善。

疑问 #

1. 为何在goroutine中调用spannerClient.Single()后，协程仍可能复用同一会话/事务？
2. 如何修改上述代码解决该问题？

问题1：协程复用同一会话/事务的原因 #

Spanner客户端维护了会话池，spannerClient.Single()会从池中获取会话创建只读事务。当启动10000个goroutine并发请求时，初始100个请求会占满池中的所有会话，但会话释放不是即时的——tx.Close()只是将会话归还池，但归还、重新分配需要时间，且池的会话创建速率有上限。

后续goroutine会等待池中有可用会话，随着等待队列变长，每个请求的整体耗时持续增加。你看到的query execution日志只统计了tx.Query()的调用时间，没包含等待会话的时间，所以初始日志显示耗时极短，但后续请求因等待会话，实际整体耗时被大幅拉长。

另外，同一时间一个会话只能处理一个事务，即使每个goroutine都调用Single()创建独立事务，会话池的复用机制会让多个请求排队使用同一个会话，最终导致串行等待，累积出超长总耗时。

问题2：代码修改方案 #

方案1：控制并发数，匹配会话池上限 #

既然会话池最大只有100个会话，就把并发goroutine数量限制在100以内，避免请求排队等待会话。用带缓冲的通道做并发控制：

import (
    "sync"
    "cloud.google.com/go/spanner"
    "google.golang.org/api/iterator"
    "golang.org/x/sync/errgroup"
)

type Employee struct {
    ID   string
    Name string
}

func queryEmp(IDs []string, spannerClient *spanner.Client, ctx context.Context) ([]Employee, error) {
    query := "Select name from Employee Where id = @id"
    g, gCtx := errgroup.WithContext(ctx)
    // 用带缓冲通道控制并发数，和MaxSessions一致
    sem := make(chan struct{}, 100)

    var results []Employee
    mu := sync.Mutex{} // 保护结果切片的并发写入

    for _, ID := range IDs {
        id := ID
        sem <- struct{}{} // 占用一个并发槽位
        g.Go(func() error {
            defer func() { <-sem }() // 释放槽位

            tx := spannerClient.Single()
            defer tx.Close()

            stmt2 := spanner.NewStatement(query)
            stmt2.Params = map[string]interface{}{
                "ID": id,
            }

            it := tx.Query(gCtx, stmt2)
            defer it.Stop()

            var e Employee
            e.ID = id
            for {
                row, err := it.Next()
                if err == iterator.Done {
                    break
                }
                if err != nil {
                    return err
                }
                if err := row.Column(0, &e.Name); err != nil {
                    return err
                }
            }

            mu.Lock()
            results = append(results, e)
            mu.Unlock()
            return nil
        })
    }
    if err := g.Wait(); err != nil {
        return nil, err
    }
    return results, nil
}

方案2：改用批量查询，减少会话占用 #

单行查询效率极低，建议用IN语句批量查询，一次获取多行数据，大幅减少请求次数和会话占用：

import (
    "sync"
    "fmt"
    "strings"
    "cloud.google.com/go/spanner"
    "google.golang.org/api/iterator"
    "golang.org/x/sync/errgroup"
)

type Employee struct {
    ID   string
    Name string
}

func queryEmpBatch(IDs []string, spannerClient *spanner.Client, ctx context.Context) ([]Employee, error) {
    // 拆分ID为多个批量组，每组数量控制在合理范围（如100个）
    batchSize := 100
    var allResults []Employee
    g, gCtx := errgroup.WithContext(ctx)
    sem := make(chan struct{}, 10) // 控制批量查询的并发数

    mu := sync.Mutex{}

    for i := 0; i < len(IDs); i += batchSize {
        end := i + batchSize
        if end > len(IDs) {
            end = len(IDs)
        }
        batchIDs := IDs[i:end]

        sem <- struct{}{}
        g.Go(func() error {
            defer func() { <-sem }()

            tx := spannerClient.Single()
            defer tx.Close()

            // 构建IN语句的参数和占位符
            params := make(map[string]interface{})
            placeholders := make([]string, len(batchIDs))
            for idx, id := range batchIDs {
                paramName := fmt.Sprintf("id%d", idx)
                placeholders[idx] = fmt.Sprintf("@%s", paramName)
                params[paramName] = id
            }
            query := fmt.Sprintf("Select id, name from Employee Where id IN (%s)", strings.Join(placeholders, ","))

            stmt := spanner.NewStatement(query)
            stmt.Params = params

            it := tx.Query(gCtx, stmt)
            defer it.Stop()

            var batchResults []Employee
            for {
                row, err := it.Next()
                if err == iterator.Done {
                    break
                }
                if err != nil {
                    return err
                }
                var e Employee
                if err := row.Columns(&e.ID, &e.Name); err != nil {
                    return err
                }
                batchResults = append(batchResults, e)
            }

            mu.Lock()
            allResults = append(allResults, batchResults...)
            mu.Unlock()
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        return nil, err
    }
    return allResults, nil
}

补充优化建议 #

• 调整会话池参数：若业务需要更高并发，可适当调高MaxSessions，但不要超过Spanner实例的配额限制。
• 完善日志统计：把计时点放在goroutine开始处，统计包含等待会话在内的完整请求耗时，方便排查问题。

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