网络监控与容灾：让网络问题无处遁形

class NetMonitorListener(
private val reporter:
NetMetricReporter
) : EventListener() {private val t = NetTiming()override fun callStart(
call: Call
) {
t.callStart = System.nanoTime()
t.url = call.request().url
.toString()
}override fun dnsStart(
call: Call,
name: String
) { t.dnsStart = System.nanoTime() }override fun dnsEnd(
call: Call,
name: String,
ips: List<InetAddress>
) {
t.dnsEnd = System.nanoTime()
t.resolvedIp = ips.firstOrNull()
?.hostAddress
}override fun connectStart(
call: Call,
addr: InetSocketAddress,
proxy: Proxy
) { t.connectStart = System.nanoTime() }override fun secureConnectStart(
call: Call
) { t.tlsStart = System.nanoTime() }override fun secureConnectEnd(
call: Call,
hs: Handshake?
) { t.tlsEnd = System.nanoTime() }override fun connectEnd(
call: Call,
addr: InetSocketAddress,
proxy: Proxy,
protocol: Protocol?
) {
t.connectEnd = System.nanoTime()
t.protocol = protocol?.toString()
}override fun responseHeadersStart(
call: Call
) {
t.responseStart =
System.nanoTime()
}override fun responseBodyEnd(
call: Call,
byteCount: Long
) {
t.responseEnd = System.nanoTime()
t.bytes = byteCount
}override fun callEnd(
call: Call
) {
t.callEnd = System.nanoTime()
t.success = true
reporter.report(t.toMetric())
}override fun callFailed(
call: Call,
e: IOException
) {
t.callEnd = System.nanoTime()
t.success = false
t.errorType = classifyError(e)
reporter.report(t.toMetric())
}
}

private fun classifyError(
e: IOException
): String = when {
e is UnknownHostException
-> "dns_fail"
e is ConnectException
-> "connect_fail"
e is SSLException
-> "tls_fail"
e is SocketTimeoutException
-> "timeout"
e.message?.contains("canceled")
== true -> "canceled"
e.message?.contains("reset")
== true -> "conn_reset"
else ->
"unknown_${e.javaClass.simpleName}"
}

class SmartSampler(
private var sampleRate: Float
= 0.1f
) {
// 远程可配，出问题时调到1.0
fun updateRate(r: Float) {
sampleRate = r
}fun shouldReport(
metric: NetMetric
): Boolean = when {
// 失败请求：100%上报
!metric.success -> true
// 慢请求（>3s）：100%上报
metric.totalMs > 3000 -> true
// 正常请求：按采样率
else ->
Random.nextFloat() < sampleRate
}
}

class NetworkProber(
private val client: OkHttpClient
) {
// 探测endpoint：空body，只测链路RTT
private val probeUrl =
"https://probe.example.com/ping"enum class Quality {
EXCELLENT, // < 100ms
GOOD,      // 100-300ms
MODERATE,  // 300-1000ms
POOR,      // 1000-3000ms
TERRIBLE   // > 3s 或失败
}suspend fun probe(): Quality {
val start =
System.currentTimeMillis()
return try {
client.newCall(
Request.Builder()
.url(probeUrl)
.head().build()
).await()
val rtt =
System.currentTimeMillis() -
start
classify(rtt)
} catch (_: Exception) {
Quality.TERRIBLE
}
}private fun classify(
rtt: Long
) = when {
rtt < 100 -> Quality.EXCELLENT
rtt < 300 -> Quality.GOOD
rtt < 1000 -> Quality.MODERATE
rtt < 3000 -> Quality.POOR
else -> Quality.TERRIBLE
}
}

class AdaptiveStrategy(
private val prober: NetworkProber
) {
fun getTimeout(
q: Quality
): Long = when (q) {
EXCELLENT -> 10_000L
GOOD -> 15_000L
MODERATE -> 20_000L
POOR -> 30_000L
TERRIBLE -> 45_000L
}fun getImageQuality(
q: Quality
) = when (q) {
EXCELLENT, GOOD ->
ImageQuality.HIGH
MODERATE ->
ImageQuality.MEDIUM
POOR, TERRIBLE ->
ImageQuality.THUMBNAIL
}fun shouldPreload(
q: Quality
) = q == EXCELLENT || q == GOODfun getRetryCount(
q: Quality
) = when (q) {
EXCELLENT, GOOD -> 1
MODERATE -> 2
POOR, TERRIBLE -> 3
}
}

class WeakNetSimulator(
private val delayMs: Long = 2000,
private val jitterMs: Long = 500,
private val dropRate: Float = 0.1f
) : Interceptor {override fun intercept(
chain: Interceptor.Chain
): Response {
// 模拟随机丢包
if (Random.nextFloat()
< dropRate) {
throw SocketTimeoutException(
"[SIM] packet loss"
)
}
// 模拟延迟+抖动
val actual = delayMs +
Random.nextLong(
-jitterMs, jitterMs
)
Thread.sleep(actual)return chain.proceed(
chain.request()
)
}
}

class DomainFailover(
private val domains: List<String>
) : Interceptor {// ["api.example.com",
//  "api-bk.example.com",
//  "api.example.cn"]private val fails =
ConcurrentHashMap<
String, AtomicInteger>()
private val current =
AtomicReference(domains[0])override fun intercept(
chain: Interceptor.Chain
): Response {
val req = chain.request()
var lastErr: IOException? = nullfor (domain in sortedDomains()) {
try {
val newUrl = req.url
.newBuilder()
.host(domain).build()
val resp = chain.proceed(
req.newBuilder()
.url(newUrl).build()
)
// 成功 → 重置计数
fails[domain]?.set(0)
current.set(domain)
return resp
} catch (e: IOException) {
fails.getOrPut(domain) {
AtomicInteger()
}.incrementAndGet()
lastErr = e
}
}
throw lastErr
?: IOException(
"All domains failed")
}private fun sortedDomains() =
domains.sortedBy {
if (it == current.get()) 0
else fails[it]?.get() ?: 1
}
}

class ResilientRepo<T>(
private val api: ApiService,
private val cache: LocalCache<T>,
private val fallback:
AssetFallback<T>
) {
suspend fun getData(
key: String
): DataResult<T> {
// Level 1: 网络
try {
val data = api.fetch(key)
cache.save(key, data)
return DataResult.fresh(data)
} catch (_: Exception) { }// Level 2: 本地缓存
cache.get(key)?.let {
return DataResult.stale(it)
}// Level 3: 预埋兜底包
fallback.get(key)?.let {
return DataResult.fallback(it)
}return DataResult.empty()
}
}

class CdnHealthChecker {// IP → 解禁时间戳
private val blocked =
ConcurrentHashMap<String, Long>()fun markBad(ip: String) {
// 屏蔽5分钟
blocked[ip] =
System.currentTimeMillis() +
5 * 60_000
}fun filterHealthy(
ips: List<InetAddress>
): List<InetAddress> {
val now =
System.currentTimeMillis()
val healthy = ips.filter {
val exp =
blocked[it.hostAddress]
exp == null || now > exp
}
// 全挂了就返回全量，避免死锁
return healthy.ifEmpty { ips }
}
}

fun buildProductionClient():
OkHttpClient {
val health = CdnHealthChecker()
val reporter =
NetMetricReporter(reportUrl)return OkHttpClient.Builder()
// DNS优化（第2篇）+健康过滤
.dns(SmartDns(httpDns, health))
// 域名容灾（本篇）
.addInterceptor(
DomainFailover(domains)
)
// 连接池（第3篇）
.connectionPool(
ConnectionPool(
32, 5, MINUTES
)
)
// HTTP/2多路复用（第3篇）
.protocols(listOf(
Protocol.HTTP_2,
Protocol.HTTP_1_1
))
// 全链路监控（本篇）
.eventListenerFactory {
NetMonitorListener(reporter)
}
.build()
}