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

Question

我目前正在尝试构建一个作业调度程序，如下所示。我的目标是能够安排任意函数的启动（这里是（Long）-

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

microssinceepoch（）中的now的实现是错误的。毫秒值应用两次。

为了说明这一点，下面是Java代码，用于打印nowInMicrosSinceEpoch（）中使用的值：

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.9999999z滚动到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;