有没有办法并行化 time.Sleep 但在 Go 中保持有效的执行时间？

Question

我正在开发一个慢查询日志解析器包，它与 golang 中的慢查询日志回放器相关联。对于回放器，有以下代码（我在其中添加了 cmets 以提高可读性）：

for {
        // method from my package that returns a Query object, containing headers values
        // and the query itself
        q := p.GetNext()
        if q == (query.Query{}) {
            break
        }
        db.logger.Tracef("query: %s", q.Query)

        // we send the SQL query to a chan that is read by workers.
        // workers just execute the query on the database, that's all.
        // results from the db query are handled in another piece of the code, it doesn't really
        // matter here
        queries <- q.Query

        // We need a reference time
        if firstPass {
            firstPass = false
            previousDate = q.Time
            continue
        }
        
        // Time field contains the Time: field value in the query header
        now := q.Time
        sleeping := now.Sub(previousDate)
        db.logger.Tracef("next sleeping time: %s", sleeping)
        time.Sleep(sleeping) // Here is my issue. For MariaDB the value is < 0 so no sleep is done

        // For MariaDB, when there is multiple queries in a short amount of
        // time, the Time field is not repeated, so we do not have to update
        // the previous date.
        if now != (time.Time{}) {
            previousDate = now
        }
    }

我遇到了一个有趣的问题： 在 MariaDB 慢查询日志中，如果 2 个（或更多）查询彼此接近，则头部没有 Time: 字段，这减少了前面代码 sn-p 中的time.Sleep(sleeping) 的数量。 但是，对于 MySQL 风格的慢查询日志，查询头中总是有一个 Time: 字段，这意味着对每个查询都进行睡眠（即使是 µs 睡眠时长）。

我注意到 MariaDB 和 MySQL 日志之间的重放时间差异很大； MariaDB 重播时间与实时非常相似（日志文件的第一次和最后一次查询之间的时间差），但另一方面 MySQL 重播时间比 IRL 高得多。在玩了pprof之后，我注意到问题来自time.Sleep，尤其是runtime.Futex，这很耗时。

我做了一些基准测试，持续时间结果与完成的time.Sleep 的数量相关（MySQL 高于 MariaDB）。

因此，我不是在单个线程中执行所有time.Sleep，而是在寻找一种不同的方法来并行执行它们而不改变有效时间，但我想不出办法来做到这一点。

Answer 1

我提出的解决方案如下：

type job struct {
    query string
    idle  time.Time
}

...
    var reference time.Time
    start := time.Now()
    for {
        q := p.GetNext()
        if q == (query.Query{}) {
            s.Stop()
            break
        }
        db.logger.Tracef("query: %s", q.Query)

        r.queries++
        s.Suffix = " queries replayed: " + strconv.Itoa(r.queries)

        // We need a reference time
        if firstPass {
            firstPass = false
            reference = q.Time
        }

        var j job
        delta := q.Time.Sub(reference)
        j.idle = start.Add(delta)
        j.query = q.Query
        db.logger.Tracef("next sleeping time: %s", j.idle)
        jobs <- j
    }

...

func (db database) worker(jobs chan job, errors chan error, wg *sync.WaitGroup) {
    defer wg.Done()
    for {
        j, ok := <-jobs
        if !ok {
            db.logger.Trace("channel closed, worker exiting")
            return
        }
        sleep := time.Until(j.idle)
        if sleep > 0 {
            time.Sleep(sleep)
        }
        rows, err := db.drv.Query(j.query)
        if err != nil {
            errors <- err
            db.logger.Debugf("failed to execute query:\n%s\nerror: %s", j.query, err)
        }
        if rows != nil {
            rows.Close()
        }
    }
}

说明：

我们将程序的开始保存在一个变量中（这里是start）。 接下来，我们设置一个参考时间（reference），这是慢查询日志文件的第一个时间戳。它永远不会改变。

然后，在每个新查询中，我们计算reference 和当前查询时间戳q.Time 之间的持续时间。让我们把它存储在delta。

我们将delta 添加到start 中，并且我们的时间线中有一个时间戳（不像过去的慢查询日志文件中那样）。我们在我们创建的名为job 的新结构中将此时间戳发送到查询旁边的工作人员。

当worker通过channel接收到job时，他会计算等到可以查询的时间。如果