移动互联网架构深度实践：从弱网优化到亿级长连接网关的全链路设计

class AdaptiveNetworkInterceptor : Interceptor {
    override fun intercept(chain: Interceptor.Chain): Response {
        val request = chain.request()
        // 1. 超时自适应：根据当前网络类型调整超时
        val timeout = when (NetworkDetector.getNetworkType()) {
            NetworkType.WIFI -> 5000L
            NetworkType._4G -> 8000L
            NetworkType._3G -> 15000L
            else -> 20000L
        }
        // 2. 重试策略：指数退避 + 域名切换
        var retryCount = 0
        var lastException: IOException? = null
        while (retryCount < 3) {
            try {
                val newRequest = request.newBuilder()
                    .header("X-Retry-Count", retryCount.toString())
                    .build()
                return chain.withConnectTimeout(timeout, TimeUnit.MILLISECONDS)
                            .withReadTimeout(timeout, TimeUnit.MILLISECONDS)
                            .proceed(newRequest)
            } catch (e: IOException) {
                lastException = e
                retryCount++
                // 切换备用域名
                if (retryCount == 2) {
                    DnsManager.switchToBackup()
                }
                Thread.sleep(200L * (1L shl retryCount)) // 指数退避
            }
        }
        throw lastException ?: IOException("重试耗尽")
    }
}

// DNS 优化：HTTPDNS 防劫持 + 预解析
object DnsManager {
    private val ipCache = ConcurrentHashMap<String, String>()
    fun preResolve(hosts: List<String>) {
        hosts.forEach { host ->
            // 异步从 HTTPDNS 服务获取 IP
            async { 
                val ip = HttpDnsClient.resolve(host)
                ipCache[host] = ip
            }
        }
    }
    fun getIp(host: String): String = ipCache[host] ?: host
}

@ChannelHandler.Sharable
public class GatewayServerHandler extends SimpleChannelInboundHandler<MessageProtos.Message> {
    // 全局连接管理（使用内存+Redis双层存储）
    private static final ConnectionRegistry registry = ConnectionRegistry.getInstance();
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MessageProtos.Message msg) {
        String userId = msg.getUserId();
        String deviceId = msg.getDeviceId();
        // 1. 注册连接（绑定 userId -> Channel）
        registry.register(userId, deviceId, ctx.channel());
        
        // 2. 根据消息类型路由到不同处理链
        switch (msg.getType()) {
            case HEARTBEAT:
                ctx.writeAndFlush(buildHeartbeatAck(msg.getSeq()));
                break;
            case JOIN_ROOM:
                RoomManager.joinRoom(userId, msg.getRoomId(), ctx.channel());
                break;
            case CHAT:
                // 走消息分发总线，异步投递
                MessageBus.deliver(msg);
                break;
        }
    }
    
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        // 清理连接（心跳超时或主动断连）
        registry.unregister(ctx.channel());
        ctx.close();
    }
}

// 连接注册表（双层存储：本地内存 + Redis，支持横向扩展）
public class ConnectionRegistry {
    private final Map<String, Channel> localCache = new ConcurrentHashMap<>();
    private final JedisPool jedisPool;
    
    public void register(String userId, String deviceId, Channel channel) {
        localCache.put(userId + ":" + deviceId, channel);
        try (Jedis jedis = jedisPool.getResource()) {
            jedis.sadd("online:" + userId, deviceId);
            jedis.expire("online:" + userId, 600); // 10分钟TTL
        }
    }
    
    public Channel getChannel(String userId, String deviceId) {
        return localCache.get(userId + ":" + deviceId);
    }
}

@Service
public class MessageBus {
    @Autowired private KafkaTemplate<String, String> kafkaTemplate;
    @Autowired private RedisTemplate<String, String> redisTemplate;
    
    public void deliver(MessageProtos.Message msg) {
        // 1. 持久化到 Kafka（用于回放和审计）
        kafkaTemplate.send("chat-logs", msg.getRoomId(), msg.toJson());
        
        // 2. 实时广播到房间内所有在线用户（通过 Redis Pub/Sub）
        String roomChannel = "room:" + msg.getRoomId();
        redisTemplate.convertAndSend(roomChannel, msg.toJson());
        
        // 3. 特殊消息走推送（如系统公告）
        if (msg.getPriority() > 8) {
            PushService.sendToAll(msg);
        }
    }
    
    // Redis 订阅者（每个网关节点订阅自己负责的房间）
    @Component
    public class RoomSubscriber implements MessageListener {
        @Override
        public void onMessage(Message message, byte[] pattern) {
            String json = message.toString();
            MessageProtos.Message msg = parseJson(json);
            // 获取该房间内的所有 Channel 并下发
            List<Channel> channels = RoomManager.getChannels(msg.getRoomId());
            for (Channel ch : channels) {
                ch.writeAndFlush(msg);
            }
        }
    }
}

class ConnectionManager {
    private var currentLevel = 0  // 0-3 逐级降级
    private val heartbeatTimer = Timer()
    
    fun start() {
        // 优先尝试 WebSocket
        tryConnectWebSocket()
    }
    
    private fun tryConnectWebSocket() {
        // 如果 3 次重连失败，降级
        if (failCount >= 3) {
            downgrade()
            return
        }
        // 实际连接逻辑...
    }
    
    private fun downgrade() {
        currentLevel++
        when (currentLevel) {
            1 -> startHttpPolling(2000)
            2 -> startBatchPolling(5000)  // 合并请求
            3 -> showOfflineMode()
        }
        // 上报降级事件到监控
        Monitor.report("connection_downgrade", currentLevel)
    }
    
    private fun startHttpPolling(interval: Long) {
        // 定期 HTTP 请求拉取消息
        timer.scheduleAtFixedRate(0, interval) {
            httpClient.get("/api/poll?lastSeq=$lastSeq")
                .onSuccess { messages ->
                    processMessages(messages)
                    // 检测网络恢复，尝试升级
                    if (NetworkDetector.isStable()) {
                        tryReconnectWebSocket()
                    }
                }
        }
    }
}

- alert: HighMessageLatency
  expr: histogram_quantile(0.99, sum(rate(msg_delivery_latency_bucket[5m]))) > 3000
  annotations:
    summary: "消息P99延迟超过3秒，可能网络或Kafka积压"
- alert: GatewayConnectionDrop
  expr: rate(gateway_disconnect_total[1m]) / rate(gateway_connect_total[1m]) > 0.3
  annotations:
    summary: "连接断连率超过30%，可能遭受攻击或网络故障"