如何在 Go 中并行化 time.Sleep 但保持有效执行时间？

哈喽！今天心血来潮给大家带来了《如何在 Go 中并行化 time.Sleep 但保持有效执行时间？》，想必大家应该对Golang都不陌生吧，那么阅读本文就都不会很困难，以下内容主要涉及到，若是你正在学习Golang，千万别错过这篇文章~希望能帮助到你！

我正在开发一个与 golang 中的慢查询日志重放器相关的慢查询日志解析器包。对于重放器，请使用以下代码（为了便于阅读，我在其中添加了注释）：

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:字段，这会减少前面代码片段中time.sleep(sleeping)的数量。 然而，对于 mysql 风格的慢查询日志，查询头中总是有一个 time: 字段，这意味着每个查询都会进行睡眠（即使是 µs 睡眠持续时间）。

我注意到 mariadb 和 mysql 日志之间的重播时间存在巨大差异； mariadb 重放持续时间与实际时间非常相似（日志文件的第一个和最后一个查询之间的时间差），但另一方面 mysql 重放时间比 irl 长得多。在玩了 pprof 后，我注意到问题来自 time.sleep，尤其是 runtime.futex ，它消耗 cpu 时间。

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

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

正确答案

我提出的解决方案如下：

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 的新结构中查询旁边的工作线程。

当工作人员通过通道接收作业时，他会计算等待他可以执行查询的时间。如果<=0，则立即执行查询，否则等待。

理论要掌握，实操不能落！以上关于《如何在 Go 中并行化 time.Sleep 但保持有效执行时间？》的详细介绍，大家都掌握了吧！如果想要继续提升自己的能力，那么就来关注golang学习网公众号吧！