Kotlin中的异步定时任务

Question

我正在尝试了解在Kotlin中以最佳方式使用异步作业定期触发，同时应用程序正在正常运行其正常任务。假设我有一个简单的应用程序，只需每秒打印出“ ......”，但每5秒钟我想要另一个作业/线程/协程（适合最好的）来打印“你有一条消息！”。对于异步作业，我有一个名为NotificationProducer的类，它看起来像这样。

class NotificationProducer {

    fun produce() {
        println("You have a message!")
    }
} 

然后，我的主要方法看起来像这样。

    while (true) {
        println("...")
        sleep(1000)
    }

我想知道我应该使用GlobalScope.async、Timer().schedule(...)还是一些Quartz job来实现我的目标？非常感谢任何建议。关键问题是通知必须来自另一个类（例如NotificationProducer）。

- wesleyy

---

我建议使用Java标准方法 - 使用Executors.newScheduledThreadPool()创建执行器，然后使用scheduleWithFixedDelay创建任务。请注意，任务Runnable中的异常会取消执行，可能需要使用try/catch。 - Eugene Petrenko

@EugenePetrenko，为什么不使用协程来执行“delay”函数呢？ - Sanat Serikuly

Answer 1

如果我理解的没错，使用 Kotlin 协程可以按照以下方式实现它：

class Presenter : CoroutineScope { // implement CoroutineScope to create local scope
    private var job: Job = Job()
    override val coroutineContext: CoroutineContext
        get() = Dispatchers.Default + job 

    // this method will help to stop execution of a coroutine. 
    // Call it to cancel coroutine and to break the while loop defined in the coroutine below    
    fun cancel() {
        job.cancel()
    }

    fun schedule() = launch { // launching the coroutine
        var seconds = 1
        val producer = NotificationProducer()
        while (true) {
            println("...")
            delay(1000)

            if (seconds++ == 5) {
                producer.produce()
                seconds = 1
            }
        }
    }
}

接下来，您可以使用Presenter类的实例来启动协程并停止它：

val presenter = Presenter()
presenter.schedule() // calling `schedule()` function launches the coroutine

//...

presenter.cancel() // cancel the coroutine when you need

- Sergio

---

很好，这是我的实现：interface IPresenter : CoroutineScope { val producers: Set val logger: Logger var jobs: CoroutineContext override val coroutineContext: CoroutineContext get() = Dispatchers.IO + jobs fun cancel() { jobs.cancel() } fun schedule(intervalMillis: Long, producer: IProducer) { jobs += launch { while (true) { producer.produce() delay(intervalMillis) } } } } - qwert_ukg

Answer 2

对于简单的调度需求，您可以考虑使用协程：

class NotificationProducerScheduler(val service: NotificationProducer, val interval: Long, val initialDelay: Long?) :
    CoroutineScope {
    private val job = Job()

    private val singleThreadExecutor = Executors.newSingleThreadExecutor()

    override val coroutineContext: CoroutineContext
        get() = job + singleThreadExecutor.asCoroutineDispatcher()


    fun stop() {
        job.cancel()
        singleThreadExecutor.shutdown()
    }

    fun start() = launch {
        initialDelay?.let {
            delay(it)
        }
        while (isActive) {
            service.produce()
            delay(interval)
        }
        println("coroutine done")
    }
}

否则，Java的并发API也非常可靠。

class NotificationProducerSchedulerJavaScheduler(
    val service: NotificationProducer,
    val interval: Long,
    val initialDelay: Long = 0
) {

    private val scheduler = Executors.newScheduledThreadPool(1)
    private val task = Runnable { service.produce() }

    fun stop() {
        scheduler.shutdown()
    }

    fun start() {
        scheduler.scheduleWithFixedDelay(task, initialDelay, interval, TimeUnit.MILLISECONDS)
    }
}

- s1m0nw1

---

抱歉，我在协程方面是个新手。但是我该如何启动NotificationProducerScheduler？ - Somaiah Kumbera

Answer 3

此函数将在后台运行一个任务，同时进行控制后台作业生命周期的“主”任务。以下是使用示例。

/**
 * Runs a task in the background in IO while the op proceeds.
 * The job is canceled when op returns.
 * This is useful for updating caches and the like.
 */
suspend fun withBackgroundTask(task: suspend () -> Unit, op: suspend () -> Unit) {
    val job = CoroutineScope(Dispatchers.IO).launch { task() }
    try {
        op()
    } finally {
        job.cancel()
    }
}

/**
 * Updates the cache in a background task while op runs.
 */ 
suspend fun withCache(cache: Cache<*>, op: suspend () -> Unit) {
    suspend fun cacheUpdate() {
        cache.fetchInternal()
        while (true) {
            delay(cache.cycle)
            cache.fetchInternal()
        }
    }
    withBackgroundTask(::cacheUpdate, op)
}

- F. P. Freely