使用select选择“更快的结果”

缘起

在所接手的几个项目中，都用到了 NTP 时间。然而，在我阅读过这几个项目中的代码时却发现，都是使用 for 循环按顺序向多个NTP服务器发起请求的，如果发现有结果了，那么就跳出循环。否则，继续下一个请求。

网络时间协议，英文名称：Network Time Protocol（NTP）是用来使计算机时间同步化的一种协议，它可以使计算机对其服务器或时钟源（如石英钟，GPS等等)做同步化，它可以提供高精准度的时间校正（LAN上与标准间差小于1毫秒，WAN上几十毫秒），且可介由加密确认的方式来防止恶毒的协议攻击。NTP的目的是在无序的Internet环境中提供精确和健壮的时间服务。

以上摘自百度百科

简单来说，就是获取到一个无法被用户修改时间设置所影响的时间。

伪代码如下（代码一）：

private var realTimeout = 0

fun main(): Unit = runBlocking {
    // 创建 10 个 [10, 1000] 内的随机数，用于模拟这些请求各自所需的耗时
    val timeoutList = (1..10).map { (10..1000).random() }
    // 从请求发出到结束时所消耗的时间差
    val consumeTime = measureTimeMillis {
        for (timeout in timeoutList) {
            if (fetchFastResult(timeout)) {
                break
            }
        }
    }
    println("timeoutList is $timeoutList")
    println("realTimeout is $realTimeout ms")
    println("consumeTime is $consumeTime ms")
}

/**
 * 模拟网络请求的耗时操作
 */
private suspend fun fetchFastResult(timeout: Int): Boolean {
    // 转换为 Duration
    val delayTime = timeout.milliseconds
    delay(delayTime)
    println("fetchNetTime：===> timeout $timeout ms")
    if (timeout >= 500) {
        // 如果使用的时间超过了 500ms，则认为当前请求超时失败
        return false
    }
    realTimeout = timeout
    return true
}

运行以上代码，我们可以有以下的打印日志：

观察以上打印的情况，我们可以清楚的发现这段代码总是获取第一个成功请求的结果。如果第一个请求失败了，那么会获取第二个、第三个请求的值，直到有一个请求成功或遍历完所有的请求，这样的请求方式如果在第一个请求能够快速响应的情况下还是能够接受的，但是如果前面的情况迟迟无法返回，那么我们只能等待前面的请求超时后才能向其它服务器发起请求了。

这肯定是我们不想要的，我们希望的是能尽快的获取到最先返回的那个值，在获取到最先返回的值之后取消其它的请求。

那么，我们的伪代码可以像下面这样写（代码二）：

fun main(): Unit = runBlocking {
    // 创建 10 个 [10, 1000] 内的随机数，用于模拟这些请求各自所需的耗时
    val timeoutList = (1..10).map { (10..1000).random() }
    // 从请求发出到结束时所消耗的时间差
    val consumeTime = measureTimeMillis {
        val requestList = timeoutList.map { async { fetchFastResult(it) } }
        val realTimeout = select {
            requestList.forEach { deferred ->
                deferred.onAwait {
                    it
                }
            }
        }
        // 取消所有的请求
        requestList.forEach { it.cancel() }
        println("timeoutList is $timeoutList")
        println("realTimeout is $realTimeout ms")
    }
    println("consumeTime is $consumeTime ms")
}

/**
 * 模拟网络请求的耗时操作
 */
private suspend fun fetchFastResult(timeout: Int): Int {
    // 转换为 Duration
    val delayTime = timeout.milliseconds
    delay(delayTime)
    println("fetchNetTime：===> timeout $timeout ms")
    if (timeout >= 500) {
        // 如果使用的时间超过了 500ms，则认为当前请求超时失败
        throw CancellationException()
    }
    return timeout
}

运行以上代码，打印日志如下：

通过对比两段代码，我们可以清楚的发现，代码二的耗时是最少的，各个请求之间互不影响，无论最先发起的请求耗时是多久，最后总会获取最先返回的那个值。

小结

通过代码二的方式，我们可以极大的缩短获取NTP时间的耗时。缺点也是比较明显的，就是会在获取NTP时间的时候发起多个请求，这样子会比较代码一来说更耗费流量一些。这个时候我们可以通过对返回的数据做一个缓存来减少请求的次数，下一次想要获取NTP时间的时候可以通过当前时间距离上一次发起请求的时间差值加上上一次获取到的NTP时间。

那么可以有如下公式：

（当前时间戳 - 上一次请求时的时间戳）+ 上一次请求到的 NTP 时间戳 = 现在的时间

获取NTP时间的代码

import android.os.SystemClock
import com.blankj.utilcode.util.TimeUtils
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.selects.select
import kotlinx.coroutines.suspendCancellableCoroutine
import kotlinx.coroutines.withContext
import timber.log.Timber
import java.net.DatagramPacket
import java.net.DatagramSocket
import java.net.InetAddress
import kotlin.coroutines.resume

/**
 * author : A Lonely Cat
 * github : https://github.com/anjiemo/SunnyBeach
 * time   : 2024/02/11
 * desc   : NTP时间帮助类
 */
object NtpHelper {

    const val INVALID_TIME = -1L
    private const val NTP_PORT = 123
    private const val NTP_PACKET_SIZE = 48
    private const val NTP_MODE_CLIENT = 3
    private const val NTP_VERSION = 4
    private const val NTP_UNIX_EPOCH_DIFFERENCE: Long = 2208988800000

    // timeout 2000ms
    private const val DEFAULT_TIME_OUT = 2000
    private val ntpServiceList = arrayOf(
        "ntp1.aliyun.com",
        "ntp2.aliyun.com",
        "ntp3.aliyun.com",
        "ntp4.aliyun.com",
        "ntp5.aliyun.com",
        "ntp6.aliyun.com",
        "ntp7.aliyun.com",
        "time1.aliyun.com",
        "time2.aliyun.com",
        "time3.aliyun.com",
        "time4.aliyun.com",
        "time5.aliyun.com",
        "time6.aliyun.com",
        "time7.aliyun.com",
        "ntp.ntsc.ac.cn",
        "cn.ntp.org.cn",
        "s1a.time.edu.cn",
        "ntp1.nim.ac.cn",
        "ntp.tuna.tsinghua.edu.cn",
        "ntp.sjtu.edu.cn",
        "ntp.neusoft.edu.cn",
        "ntp2.nim.ac.cn",
        "time.ustc.edu.cn",
        "0.cn.pool.ntp.org",
        "1.cn.pool.ntp.org",
        "2.cn.pool.ntp.org",
        "3.cn.pool.ntp.org",
        "time.windows.com",
        "time.apple.com",
        "time.asia.apple.com",
    )
    private var originElapsedRealtime = SystemClock.elapsedRealtime()
    private var mNtpTime = INVALID_TIME

    /**
     * Make sure that the [fetchFastestNTPResponseTime] function was successfully called.
     * see：[fetchFastestNTPResponseTime]
     */
    fun getRealTime(): Long {
        return mNtpTime.takeUnless { it == INVALID_TIME }?.let {
            mNtpTime + (SystemClock.elapsedRealtime() - originElapsedRealtime)
        } ?: INVALID_TIME
    }

    /**
     * By default, the current time is estimated using the last obtained NTP time, and if it is not,
     * the current NTP time is immediately obtained and cached.
     * see：[getRealTime], [getNTPTime]
     */
    suspend fun fetchFastestNTPResponseTime(timeout: Int = DEFAULT_TIME_OUT, forceRefresh: Boolean = false): Long {
        takeUnless { forceRefresh }?.let {
            getRealTime().takeUnless { it == INVALID_TIME }?.let { return it }
        }
        return withContext(Dispatchers.IO) {
            val deferredList = ntpServiceList.map { host ->
                async { HostAndTime(host, getNTPTime(host, timeout)) }
            }
            // prioritizes the fastest one
            val fastestNTPTime = select {
                deferredList.forEach { deferred ->
                    deferred.onAwait { it }
                }
            }
            // cancel other requests
            deferredList.forEach { it.cancel() }
            Timber.d("getFastestNTPTime：===> host is ${fastestNTPTime.host}, at ${TimeUtils.millis2String(fastestNTPTime.time)}")
            fastestNTPTime.time.also { mNtpTime = it }
        }
    }

    /**
     * Get the current NTP time from the given host.
     */
    suspend fun getNTPTime(host: String, timeout: Int = DEFAULT_TIME_OUT): Long = suspendCancellableCoroutine { continuation ->
        runCatching {
            // 为 NTP 请求创建缓冲区
            val ntpRequest = ByteArray(NTP_PACKET_SIZE)
            // 设置 NTP 模式为客户端，版本为4
            ntpRequest[0] = (NTP_MODE_CLIENT or (NTP_VERSION shl 3)).toByte()

            val socket = DatagramSocket()
            continuation.invokeOnCancellation {
                socket.close()
            }
            socket.soTimeout = timeout
            val requestPacket = DatagramPacket(ntpRequest, ntpRequest.size, InetAddress.getByName(host), NTP_PORT)
            val t1 = System.currentTimeMillis()
            // 发送 NTP 请求到服务器
            socket.send(requestPacket)
            // 接收 NTP 响应
            val ntpResponse = ByteArray(NTP_PACKET_SIZE)
            val responsePacket = DatagramPacket(ntpResponse, ntpResponse.size)
            socket.receive(responsePacket)
            socket.use {
                val t4 = System.currentTimeMillis()
                // 将接收到的NTP响应数据字段转换为时间（从1900年起的毫秒值）
                val t2 = ((ntpResponse[32].toLong() and 0xFF shl 24) or
                        (ntpResponse[33].toLong() and 0xFF shl 16) or
                        (ntpResponse[34].toLong() and 0xFF shl 8) or
                        (ntpResponse[35].toLong() and 0xFF)) * 1000L - NTP_UNIX_EPOCH_DIFFERENCE
                val t3 = ((ntpResponse[40].toLong() and 0xFF shl 24) or
                        (ntpResponse[41].toLong() and 0xFF shl 16) or
                        (ntpResponse[42].toLong() and 0xFF shl 8) or
                        (ntpResponse[43].toLong() and 0xFF)) * 1000L - NTP_UNIX_EPOCH_DIFFERENCE
                val delay = (t4 - t1) - (t3 - t2)
                val offset = ((t2 - t1) + (t3 - t4)) / 2
                Timber.d("getNTPTime：===> Round trip delay: $delay ms")
                originElapsedRealtime = SystemClock.elapsedRealtime()
                System.currentTimeMillis() + offset
            }
        }.onSuccess { ntpTime ->
            continuation.resume(ntpTime)
        }.onFailure {
            throw CancellationException(it.message)
        }
    }
}

data class HostAndTime(val host: String, val time: Long)

食用方式

    override fun initData() {
        lifecycleScope.launch {
            mBinding.tvCopyrightInfo.text = getString(R.string.about_copyright, 2023)
            val ntpTime = NtpHelper.fetchFastestNTPResponseTime()
            takeUnless { ntpTime == NtpHelper.INVALID_TIME }?.let {
                val calendar = Calendar.getInstance().also { it.timeInMillis = ntpTime }
                mBinding.tvCopyrightInfo.text = getString(R.string.about_copyright, calendar[Calendar.YEAR])
            }
        }
    }

结语

好啦，就写到这里吧 ，如果你想看更多的项目代码，请查看我在 Github 上的 阳光沙滩APP项目 ，你也可以在这里点击下载我写的 阳光沙滩APP客户端 进行体验。

如果对你有帮助的话，欢迎一键三连+关注哦~