延迟（）准确性问题/作业调度程序的奇怪行为

Question

我目前正在尝试构建一个作业调度程序，如下所示。我的目标是能够在启动时间上尽可能准确地安排任意函数（此处为 (Long) -> Unit））的启动（理想的情况是亚毫秒）。

import java.util.*
import kotlinx.coroutines.*
import java.util.concurrent.PriorityBlockingQueue
import kotlin.math.max
import java.time.Instant

fun nowInMicrosSinceEpoch() : Long {
    val now = Instant.now()
    return now.toEpochMilli() * 1000L + (now.getNano().toLong() / 1000L)
}

open class TimeCallback(open var time : Long, open val callback : (Long) -> Unit) {
    open fun run(){
        callback(time)
    }

    override fun toString() : String {
        return "(TimeCallback - T:${time/1000L})"
    }
}

class PulseCallback(override var time : Long,
                    override val callback : (Long) -> Unit,
                    val pulsePeriod : Long,
                    val callbackQueue : AbstractQueue<TimeCallback>) : TimeCallback(time, callback) {
    override fun run(){
        callback(time)
        time += pulsePeriod
        callbackQueue.add(this)
    }

    override fun toString() : String {
        return "(PulseCallback - T:${time/1000L} - PP:${pulsePeriod/1000L})"
    }
}

abstract class Clock {
    protected abstract var currentTime: Long
    protected val comparator : Comparator<TimeCallback> = compareBy<TimeCallback> { x -> x.time }

    abstract fun start()
    abstract fun stop()
    abstract fun addCallback(time: Long, callback: (Long) -> Unit)
    abstract fun addPulseCallback(time: Long, pulsePeriod: Long, callback: (Long) -> Unit)
    abstract fun getTime() : Long
}

class LiveClock : Clock() {
    override var currentTime : Long = nowInMicrosSinceEpoch()
    private val callbacks : PriorityBlockingQueue<TimeCallback> = PriorityBlockingQueue<TimeCallback>(10000, comparator)

    private var clockCoroutine : Job? = null

    override fun start(){
        clockCoroutine = GlobalScope.launch {
            try{
                var waitTime : Long
                while(true) {
                    println(callbacks)
                    val callback: TimeCallback = callbacks.take()
                    currentTime = nowInMicrosSinceEpoch()
                    waitTime = max(callback.time - currentTime, 0L) / 1000L
                    println("Now is ${currentTime/1000L}, waiting $waitTime ms until ${callback.time/1000L}")
                    delay(waitTime)
                    callback.run()
                }
            } finally {
                println("Clock was stopped by CancellationException.")
            }
        }
    }

    override fun stop(){
        // Cannot stop before starting!
        clockCoroutine!!.cancel()
    }

    override fun addCallback(time: Long, callback: (Long) -> Unit){
        callbacks.add(TimeCallback(
            time = time,
            callback = callback
        ))
    }

    override fun addPulseCallback(firstPulse: Long, pulsePeriod: Long, callback: (Long) -> Unit){
        callbacks.add(PulseCallback(
            time = firstPulse,
            pulsePeriod = pulsePeriod,
            callback = callback,
            callbackQueue = callbacks
        ))
    }

    override fun getTime() : Long {
        return nowInMicrosSinceEpoch()
    }
}

fun printTest(t : Long){
    println("Time difference: ${nowInMicrosSinceEpoch()/1000L - (t/1000L)} ms")
}

fun main(args: Array<String>) {
    val clock = LiveClock()
    clock.addPulseCallback(nowInMicrosSinceEpoch(), 1000*1000L, ::printTest)
    clock.addPulseCallback(nowInMicrosSinceEpoch(), 500*1000L, ::printTest)
    clock.start()
    runBlocking {
        // Run for 100 seconds...
        delay(100000L)
    }
}

但是，即使使用上面非常简单的示例（在 main() 中），我也会在计划时间和计划函数实际运行的时间之间获得显着的时间差异。有些甚至在预定时间之前运行（见下面的最后一行，负时差），这对我来说仍然是个谜。回调怎么可能在调用 time delay() 之前运行？

谢谢！

[(PulseCallback - T:1547692545172 - PP:1000), (PulseCallback - T:1547692545184 - PP:500)]
Now is 1547692545262, waiting 0 ms until 1547692545172
1547692545264 - Time difference: 92 ms
[(PulseCallback - T:1547692545184 - PP:500), (PulseCallback - T:1547692546172 - PP:1000)]
Now is 1547692545264, waiting 0 ms until 1547692545184
1547692545264 - Time difference: 80 ms
[(PulseCallback - T:1547692545684 - PP:500), (PulseCallback - T:1547692546172 - PP:1000)]
Now is 1547692545264, waiting 420 ms until 1547692545684
1547692546110 - Time difference: 426 ms
[(PulseCallback - T:1547692546172 - PP:1000), (PulseCallback - T:1547692546184 - PP:500)]
Now is 1547692546110, waiting 62 ms until 1547692546172
1547692546234 - Time difference: 62 ms
[(PulseCallback - T:1547692546184 - PP:500), (PulseCallback - T:1547692547172 - PP:1000)]
Now is 1547692546234, waiting 0 ms until 1547692546184
1547692546234 - Time difference: 50 ms
[(PulseCallback - T:1547692546684 - PP:500), (PulseCallback - T:1547692547172 - PP:1000)]
Now is 1547692546234, waiting 450 ms until 1547692546684
1547692546136 - Time difference: -548 ms
[(PulseCallback - T:1547692547172 - PP:1000), (PulseCallback - T:1547692547184 - PP:500)]
Now is 1547692546136, waiting 1036 ms until 1547692547172

Answer 1

nowInMicrosSinceEpoch() 的实现是错误的。毫秒值应用两次。

为了说明这一点，这里是打印 nowInMicrosSinceEpoch() 中使用的值的 Java 代码：

Instant now = Instant.now();
System.out.println(now);
System.out.printf("%23d        toEpochMilli()%n", now.toEpochMilli());
System.out.printf("%26d     toEpochMilli() * 1000 = a%n", now.toEpochMilli() * 1000L);
System.out.printf("%29d  getNano()%n", now.getNano());
System.out.printf("%26d     getNano() / 1000 = b%n", now.getNano() / 1000L);
System.out.printf("%26d     a + b%n", now.toEpochMilli() * 1000L + now.getNano() / 1000L);

输出

2019-02-02T00:16:58.999999999Z
          1549066618999        toEpochMilli()
          1549066618999000     toEpochMilli() * 1000 = a
                    999999999  getNano()
                    999999     getNano() / 1000 = b
          1549066619998999     a + b

因此，当时钟从 x:58.999999999Z 翻转到 x:59.000000000Z 时，您会得到：

2019-02-02T00:16:59.000000000Z
          1549066619000        toEpochMilli()
          1549066619000000     toEpochMilli() * 1000 = a
                    000000000  getNano()
                    000000     getNano() / 1000 = b
          1549066619000000     a + b

1 纳秒后的值返回一个早于 998999 微秒的值。
计算值是双倍速运行，每秒跳回1秒。

正确的公式是（在 Java 中）：

Instant now = Instant.now();
return now.getEpochSecond() * 1000000L + now.getNano() / 1000L;