如何使用 Kotlin 协程实现计时器

Question

我想用 Kotlin 协程实现定时器，类似于用 RxJava 实现的东西：

       Flowable.interval(0, 5, TimeUnit.SECONDS)
                    .observeOn(AndroidSchedulers.mainThread())
                    .map { LocalDateTime.now() }
                    .distinctUntilChanged { old, new ->
                        old.minute == new.minute
                    }
                    .subscribe {
                        setDateTime(it)
                    }

它将每隔一分钟发出 LocalDateTime。

Answer 1

编辑：请注意，原始答案中建议的 API 现在标记为 @ObsoleteCoroutineApi：

Ticker 通道目前未与结构化并发集成，其 api 将在未来发生变化。

您现在可以使用Flow API 创建您自己的代码流：

import kotlin.time.*
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun tickerFlow(period: Duration, initialDelay: Duration = Duration.ZERO) = flow {
    delay(initialDelay)
    while (true) {
        emit(Unit)
        delay(period)
    }
}

您可以以与当前代码非常相似的方式使用它：

tickerFlow(Duration.seconds(5))
    .map { LocalDateTime.now() }
    .distinctUntilChanged { old, new ->
        old.minute == new.minute
    }
    .onEach {
        setDateTime(it)
    }
    .launchIn(viewModelScope) // or lifecycleScope or other

如果您不想要实验性的Duration API，也可以使用Long 毫秒。

注意：使用此处编写的代码，tickerFlow 未考虑处理元素所花费的时间，因此延迟可能不是定期的（这是元素处理之间的延迟）。如果您希望代码独立于每个元素的处理进行滴答，您可能需要使用buffer 或专用线程（例如通过flowOn）。

---

原答案

我相信它仍处于试验阶段，但您可以使用 TickerChannel 来生成每 X 毫秒的值：

val tickerChannel = ticker(delayMillis = 60_000, initialDelayMillis = 0)

repeat(10) {
    tickerChannel.receive()
    val currentTime = LocalDateTime.now()
    println(currentTime)
}

如果您需要在“订阅”为每个“滴答”做某事的同时继续工作，您可以launch 一个后台协程，该协程将从该频道读取并做您想做的事情：

val tickerChannel = ticker(delayMillis = 60_000, initialDelayMillis = 0)

launch {
    for (event in tickerChannel) {
        // the 'event' variable is of type Unit, so we don't really care about it
        val currentTime = LocalDateTime.now()
        println(currentTime)
    }
}

delay(1000)

// when you're done with the ticker and don't want more events
tickerChannel.cancel()

如果你想从循环内部停止，你可以简单地跳出它，然后取消通道：

val ticker = ticker(500, 0)

var count = 0

for (event in ticker) {
    count++
    if (count == 4) {
        break
    } else {
        println(count)
    }
}

ticker.cancel()

Answer 2

编辑：Joffrey 用更好的方法编辑了他的解决方案。

旧：

Joffrey 的解决方案对我有用，但我遇到了 for 循环的问题。

我必须像这样在 for 循环中取消我的代码：

            val ticker = ticker(500, 0)
            for (event in ticker) {
                if (...) {
                    ticker.cancel()
                } else {
                    ...
                    }
                }
            }

但是ticker.cancel() 抛出了一个cancellationException，因为for 循环一直在执行此操作。

我必须使用 while 循环来检查通道是否未关闭，以免出现此异常。

                val ticker = ticker(500, 0)
                while (!ticker.isClosedForReceive && ticker.iterator().hasNext()) {
                    if (...) {
                        ticker.cancel()
                    } else {
                        ...
                        }
                    }
                }

Answer 3

你可以像这样创建倒数计时器

GlobalScope.launch(Dispatchers.Main) {
            val totalSeconds = TimeUnit.MINUTES.toSeconds(2)
            val tickSeconds = 1
            for (second in totalSeconds downTo tickSeconds) {
                val time = String.format("%02d:%02d",
                    TimeUnit.SECONDS.toMinutes(second),
                    second - TimeUnit.MINUTES.toSeconds(TimeUnit.SECONDS.toMinutes(second))
                )
                timerTextView?.text = time
                delay(1000)
            }
            timerTextView?.text = "Done!"
        }

Answer 4

它没有使用 Kotlin 协程，但如果您的用例足够简单，您始终可以使用 fixedRateTimer 或 timer (docs here) 之类的东西，它们解析为 JVM 原生 Timer。

我在一个相对简单的场景中使用 RxJava 的 interval，当我切换到使用 Timers 时，我看到了显着的性能和内存改进。

您还可以使用 View.post() 或其多种变体在 Android 的主线程上运行您的代码。

唯一真正的烦恼是您需要自己跟踪旧时的状态，而不是依赖 RxJava 为您完成。

但这总是会快得多（如果您正在执行 UI 动画等性能关键的事情，这很重要）并且不会有 RxJava 的 Flowables 的内存开销。

这是使用fixedRateTimer 的问题代码：

var currentTime: LocalDateTime = LocalDateTime.now()

fixedRateTimer(period = 5000L) {
    val newTime = LocalDateTime.now()
    if (currentTime.minute != newTime.minute) {
        post { // post the below code to the UI thread to update UI stuff
            setDateTime(newTime)
        }
        currentTime = newTime
    }
}

Answer 5

这是一个使用 Kotlin Flow 的可能解决方案

fun tickFlow(millis: Long) = callbackFlow<Int> {
    val timer = Timer()
    var time = 0
    timer.scheduleAtFixedRate(
        object : TimerTask() {
            override fun run() {
                try { offer(time) } catch (e: Exception) {}
                time += 1
            }
        },
        0,
        millis)
    awaitClose {
        timer.cancel()
    }
}

用法

val job = CoroutineScope(Dispatchers.Main).launch {
   tickFlow(125L).collect {
      print(it)
   }
}

...

job.cancel()

Answer 6

另一种可能的解决方案是CoroutineScope 的可重用 kotlin 扩展

fun CoroutineScope.launchPeriodicAsync(
    repeatMillis: Long,
    action: () -> Unit
) = this.async {
    if (repeatMillis > 0) {
        while (isActive) {
            action()
            delay(repeatMillis)
        }
    } else {
        action()
    }
}

然后用法为：

var job = CoroutineScope(Dispatchers.IO).launchPeriodicAsync(100) {
  //...
}

然后打断它：

job.cancel()

Answer 7

一个非常实用的 Kotlin Flows 方法可能是：

// Create the timer flow
val timer = (0..Int.MAX_VALUE)
    .asSequence()
    .asFlow()
    .onEach { delay(1_000) } // specify delay

// Consume it
timer.collect { 
    println("bling: ${it}")
}

Answer 8

这是基于 Joffrey 的回答的 Flow 版本的 Observable.intervalRange(1, 5, 0, 1, TimeUnit.SECONDS)：

fun tickerFlow(start: Long,
               count: Long,
               initialDelayMs: Long,
               periodMs: Long) = flow<Long> {
    delay(initialDelayMs)

    var counter = start
    while (counter <= count) {
        emit(counter)
        counter += 1

        delay(periodMs)
    }
}

//...

tickerFlow(1, 5, 0, 1_000L)

Answer 9

制作Observable.intervalRange(0, 90, 0, 1, TimeUnit.SECONDS) 的副本（每 1 秒会在 90 秒内发射物品）：

fun intervalRange(start: Long, count: Long, initialDelay: Long = 0, period: Long, unit: TimeUnit): Flow<Long> {
        return flow<Long> {
            require(count >= 0) { "count >= 0 required but it was $count" }
            require(initialDelay >= 0) { "initialDelay >= 0 required but it was $initialDelay" }
            require(period > 0) { "period > 0 required but it was $period" }

            val end = start + (count - 1)
            require(!(start > 0 && end < 0)) { "Overflow! start + count is bigger than Long.MAX_VALUE" }

            if (initialDelay > 0) {
                delay(unit.toMillis(initialDelay))
            }

            var counter = start
            while (counter <= count) {
                emit(counter)
                counter += 1

                delay(unit.toMillis(period))
            }
        }
    }

用法：

lifecycleScope.launch {
intervalRange(0, 90, 0, 1, TimeUnit.SECONDS)
                .onEach {
                    Log.d(TAG, "intervalRange: ${90 - it}")
                }
                .lastOrNull()
}