multithreading - sqlx.Next和sqlx.StructScan是否可以安全同时使用？

Question

我在MySQL数据库中有一张大表，我正试图尽可能高效地读取该表。我当时在考虑通过添加多个工作程序来加速代码，但是当我这样做时，在运行它的开始(并且仅在开始时)出现编码(marshal)处理错误，它看起来像这样:

{"caller":"mysql.go:repository.(*MySQLRepo).GetNextBatch#428","error":"DBGetRecordException:could not marshal episode comments: sql: Scan error on column index 4,name "created_at": unsupported Scan, storing driver.Value type[]uint8 into type*time.Time","level":"error","ts":"2020-07-13T20:42:03.9621 658Z"}

如果我从ImportLegacyComments中删除工作程序代码并正常循环，则不会出现此错误。 sqlx.next和sqlx.StructScan对多线程是否安全？如果没有，是否可以安全地进行此操作？

import (
    _ "github.com/go-sql-driver/mysql"
    "github.com/jmoiron/sqlx"
)

type BatchResult struct {
    rows *sqlx.Rows
}

func (m *MySQLRepo) GetNextBatch(b *BatchResult) ([]model.EpisodeComment, error) {
    var episodeComments []model.EpisodeComment
    for i := 0; i < 1000 && b.rows.Next(); i++ {
        var episodeComment model.EpisodeComment
        err := b.rows.StructScan(&episodeComment)
        if err != nil {
            return nil, err
        }
        episodeComments = append(episodeComments, episodeComment)
    }
    return episodeComments, nil
}
func (m *MySQLRepo) FetchAllEpisodeComments() (*BatchResult, error) {
    rows, err := m.db.Queryx("SELECT * FROM episode_comment")
    if err != nil {
        return nil, err
    }
    return &BatchResult{
        rows: rows,
    }, nil
}

func (svc *ImportService) ImportLegacyComments(ctx context.Context) error {
    batchResult, err := svc.legacyCommentsRepo.FetchAllEpisodeComments()
    var wg sync.WaitGroup
    processor := func() {
        comments, err := svc.legacyCommentsRepo.GetNextBatch(batchResult)
        if err != nil {
            svc.logger.Error(err)
        }
        for len(comments) > 0 {
            comments, err = svc.legacyCommentsRepo.GetNextBatch(batchResult)
            if err != nil {
                svc.logger.Error(err)
            }
            svc.logger.Info("batch", "completed 1000")
        }
        wg.Done()
    }

    for i := 0; i < 20; i++ {
        go processor()
        wg.Add(1)
    }

    wg.Wait()

    return err
}

Answer 1

sqlx.Next和sqlx.StructScan不能安全同时使用。
如果您为代码组合了一个简单的单元测试，并使用竞速检测器go test -race运行它，它将在"database/sql".Rows结构的未导出字段上报告竞态条件:

Write at 0x00c00000e080 by goroutine 22:
  github.com/lib/pq.(*rows).Next()
      /Users/blackgreen/go/pkg/mod/github.com/lib/pq@v1.2.0/conn.go:1464 +0x8ec
  ...

Previous read at 0x00c00000e080 by goroutine 20:
  database/sql.(*Rows).Scan()
      /usr/local/go/src/database/sql/sql.go:3041 +0x2fa
  ...

如果我们找出哪个字段使种族探测器感到提示，我们可以看到正确记录了并发使用的迹象:

    // lastcols is only used in Scan, Next, and NextResultSet which are expected
    // not to be called concurrently.
    lastcols []driver.Value