我花了三个月，从零撸了一套支撑百万级的高并发消息推送系统

引言

大家好，给大家讲一下前段时间在公司的事情。那时候公司业务扩张，原来用的第三方推送服务越来越不够用，领导拍板让我带两个同事，从零搞一套自己的高并发消息推送系统。

说实话一开始我心里还挺没底，但嘴上还是硬着头皮说没问题。毕竟之前也用过不少开源IM和推送组件，总觉得原理都懂，不就是个WebSocket长连接嘛，能有多复杂？后面等自己真正的动手后才发现有多复杂，从设备注册、连接保活，到消息路由、离线缓存，再到多端SDK、监控告警，每一步都有要考虑的东西，而且有些坑都是踩进去才知道疼的那种。

先给大家透一下底：这个系统前前后后折腾了三个多月，系统终于稳定上线了。目前稳定支撑80多万在线长连接，峰值每秒推送能扛5万条，可用性99%以上。今天就把整个项目的设计思路、可运行的实现代码，还有我踩过的那些坑全部分享出来，算是给自己做个复盘，也给想做类似系统的朋友提供一些经验。

一、为啥要自己搞一套推送系统？

很多朋友可能会问，现在极光、个推这么成熟，直接用不行吗？说实话，我们前两年也一直用第三方，要不是实在受不了了，谁也不愿意费这劲搞这个东西。我总结下来主要有几个痛点：

第一是成本问题。第三方按推送量阶梯收费，而且很多增值功能还要单独加钱，算下来太不划算。

第二是数据合规风险。我们的推送内容里包含用户的订单、账户等敏感信息，所有消息都要过第三方的服务器，公司安全部门一直有顾虑，等保测评的时候也反复提这个问题。自己做的话，数据全在我们自己的服务器上，可控一些。

第三是定制化能力太差。比如我们想做“单用户离线消息最多存200条”“7天自动过期”“不同优先级消息走不同通道”这类自定义逻辑，第三方要么不支持，要么要开高价企业版，还不能完全按我们的需求改。想加个内部业务的扩展字段都麻烦的很。

第四是延迟不稳定。一到电商大促或者运营活动高峰期，第三方的推送延迟能从几百毫秒飙到十几秒，用户投诉等等乱的很。而且我们没法控制它的调度优先级，核心的订单通知经常被营销消息堵住。

我们也评估过开源的OpenIM、悟空IM，这些开源项目都是完整的IM系统，聊天、群组、音视频什么都有，对我们来说比较重。我们就只需要一个纯推送能力，集成进来冗余代码一大堆，二次开发还要理解别人的架构，还不如自己做一套轻量的，完全可控。

二、整体架构长啥样？

一开始我们想得特别简单：一个Netty服务，前端连WebSocket，后端接数据库，消息来了直接转发就行。结果第一次压测，1万连接就把服务压崩了，各种问题就全都出来了。

后来踩坑踩多了，就慢慢的迭代，并拆分成了分层架构，核心思路就是：分层解耦、异步削峰、水平扩展、可观测。最终的架构从上到下大概分了这么几层：

1. 接入层：推送网关
   这个是基于Netty实现，是所有设备长连接的入口，负责连接管理、心跳检测、协议编解码、消息收发。集群部署，加机器就能线性扩展连接数。

2. 业务逻辑层
   我们这个拆成了几个独立的微服务：设备管理服务管注册鉴权、在线状态服务管用户设备的在线状态、消息路由服务负责消息分发、离线消息服务管不在线时的消息存储。每个服务只干自己的那一件事，独立扩容。

3. 消息队列层
   是用Kafka做异步解耦和流量削峰。业务方的推送请求先到Kafka，后端服务慢慢消费，就算运营突然发全量推送，也不会直接把系统冲的有问题。

4. 存储层
   存储是用Redis存热点数据：在线状态、设备token、短期离线消息，读写快；MySQL存设备基础信息、用户绑定关系这些冷数据；MongoDB存历史消息和推送日志，海量数据查询也不慢，对俺们这个项目来说很适用。

5. 监控告警层
   Prometheus + Grafana做指标监控，Zipkin做链路追踪，企业微信推送告警。没监控的系统是肯定不行的，这个一定要提前做的，方面后面的维护和问题的一个及时解决处理。

6. 客户端SDK层
   Android、iOS原生SDK，外加Flutter封装层，统一的接口给业务方调用。

其实这些架构原则说起来谁都懂，但真做的时候，好几次为了赶进度想省事，都差点没有用上。比如一开始想把路由逻辑直接写在网关里，后来压测发现网关CPU直接飙满，到后面又把它拆成了独立服务。所以架构都是慢慢演进出来的，不是一开始就设计完美的。

三、第一步：设备注册与身份鉴权，先把身份捋明白

推送系统的第一步，就是得知道连上来的是谁，有没有权限连。总不能谁都能连上来占资源对吧。

3.1 整体流程

APP安装后第一次启动，先获取设备唯一标识，调用我们的设备注册接口，把设备类型、系统版本、APP版本这些信息传过来。服务端校验通过后，生成一个唯一的设备token返回给客户端。后面客户端建立长连接的时候，必须带上这个token，鉴权通过了才能保持连接。

3.2 踩过的坑：设备唯一标识

这是第一个坑，而且摔得特别实。一开始安卓端我们用MAC地址做唯一标识，结果安卓10以上的系统根本拿不到真实MAC，返回的都是随机值。用户卸载重装一次APP，就变成一个新设备，不仅收不到历史消息，还产生了大量无效设备数据。

后来赶紧换成了OAID，这是国内安卓厂商联合推出的广告标识符，能保证同一设备不变。iOS端用IDFV，虽然卸载重装会变，但也没有更好的方案了。另外我们还加了用户ID和设备的绑定逻辑，用户登录后把用户ID和设备ID绑起来，推送的时候按用户ID发，就能推到他所有的登录设备上。

3.3 鉴权方案：为什么不用JWT

一开始图省事，我用JWT生成token，自包含、不用查库，多方便。结果上线没两周，运营要封禁一个刷恶意消息的设备，我傻眼了——JWT是自包含的，没过期之前根本没法主动失效，只能等它自己过期，硬生生卡了两天。

后来赶紧改成了Redis存储token的方案：生成的token存在Redis里，设置30天过期，鉴权的时候查Redis。想封禁设备直接删对应的key，立刻生效，灵活多了。设备基础信息也在Redis缓存一份，减少数据库查询压力。

3.4 可运行代码实现

设备注册服务实现，是使用Caffeine作为本地缓存，提高查询性能和Redis作为分布式缓存，支持集群部署，使用的是Map作为数据存储

public class DeviceRegistryImpl implements DeviceRegistry {
    
    // 设备存储（实际生产环境应使用数据库）
    private final Map<String, Device> deviceStore = new ConcurrentHashMap<>();
    
    // 用户-设备映射关系
    private final Map<String, Set<String>> userDeviceMap = new ConcurrentHashMap<>();
    
    // 本地缓存
    private final Cache<String, Device> localCache;
    
    // Redis客户端（模拟）
    private Object redisClient;
    
    /**
     * 构造函数
     */
    public DeviceRegistryImpl() {
        // 初始化本地缓存
        this.localCache = Caffeine.newBuilder()
                .maximumSize(SystemConstants.CACHE_MAX_SIZE)
                .expireAfterWrite(SystemConstants.CACHE_EXPIRE_SECONDS, TimeUnit.SECONDS)
                .build();
    }
    
    /**
     * 设置Redis客户端
     */
    public void setRedisClient(Object redisClient) {
        this.redisClient = redisClient;
    }
    
    @Override
    public Device register(Device device) throws PushException {
        // 参数校验
        if (device == null) {
            throw new PushException("INVALID_PARAM", "设备信息不能为空");
        }
        
        if (device.getDeviceId() == null || device.getDeviceId().isEmpty()) {
            throw new PushException("INVALID_PARAM", "设备ID不能为空");
        }
        
        if (device.getUserId() == null || device.getUserId().isEmpty()) {
            throw new PushException("INVALID_PARAM", "用户ID不能为空");
        }
        
        if (device.getDeviceType() == null || device.getDeviceType().isEmpty()) {
            throw new PushException("INVALID_PARAM", "设备类型不能为空");
        }
        
        // 验证设备类型
        if (!isValidDeviceType(device.getDeviceType())) {
            throw new PushException("INVALID_PARAM", "不支持的设备类型: " + device.getDeviceType());
        }
        
        // 检查设备是否已注册
        Device existingDevice = deviceStore.get(device.getDeviceId());
        if (existingDevice != null) {
            // 设备已存在，更新设备信息
            device.setRegisterTime(existingDevice.getRegisterTime());
            device.setStatus(1);
        } else {
            // 新设备注册
            device.setRegisterTime(new Date());
            device.setStatus(1);
        }
        
        device.setLastActiveTime(new Date());
        
        // 存储设备信息
        deviceStore.put(device.getDeviceId(), device);
        
        // 更新用户-设备映射
        userDeviceMap.computeIfAbsent(device.getUserId(), k -> ConcurrentHashMap.newKeySet())
                .add(device.getDeviceId());
        
        // 更新本地缓存
        localCache.put(device.getDeviceId(), device);
        
        // 同步到Redis（实际生产环境）
        syncToRedis(device);
        
        System.out.println("[设备注册] 设备注册成功: " + device);
        
        return device;
    }
    
    @Override
    public boolean unregister(String deviceId) throws PushException {
        if (deviceId == null || deviceId.isEmpty()) {
            throw new PushException("INVALID_PARAM", "设备ID不能为空");
        }
        
        Device device = deviceStore.remove(deviceId);
        if (device != null) {
            // 从用户-设备映射中移除
            Set<String> userDevices = userDeviceMap.get(device.getUserId());
            if (userDevices != null) {
                userDevices.remove(deviceId);
                if (userDevices.isEmpty()) {
                    userDeviceMap.remove(device.getUserId());
                }
            }
            
            // 移除本地缓存
            localCache.invalidate(deviceId);
            
            // 从Redis移除（实际生产环境）
            removeFromRedis(deviceId);
            
            System.out.println("[设备注销] 设备注销成功: " + deviceId);
            return true;
        }
        
        return false;
    }
    
    @Override
    public boolean update(Device device) throws PushException {
        if (device == null || device.getDeviceId() == null) {
            throw new PushException("INVALID_PARAM", "设备信息无效");
        }
        
        Device existingDevice = deviceStore.get(device.getDeviceId());
        if (existingDevice == null) {
            throw new PushException("DEVICE_NOT_FOUND", "设备不存在: " + device.getDeviceId());
        }
        
        // 更新设备信息
        device.setRegisterTime(existingDevice.getRegisterTime());
        device.setLastActiveTime(new Date());
        deviceStore.put(device.getDeviceId(), device);
        
        // 更新本地缓存
        localCache.put(device.getDeviceId(), device);
        
        // 同步到Redis
        syncToRedis(device);
        
        System.out.println("[设备更新] 设备信息更新成功: " + device);
        
        return true;
    }
    
    @Override
    public Device getDevice(String deviceId) {
        if (deviceId == null || deviceId.isEmpty()) {
            return null;
        }
        
        // 先从本地缓存获取
        Device device = localCache.getIfPresent(deviceId);
        if (device != null) {
            return device;
        }
        
        // 从存储获取
        device = deviceStore.get(deviceId);
        if (device != null) {
            // 更新本地缓存
            localCache.put(deviceId, device);
        }
        
        return device;
    }
    
    @Override
    public List<Device> getDevicesByUserId(String userId) {
        if (userId == null || userId.isEmpty()) {
            return Collections.emptyList();
        }
        
        Set<String> deviceIds = userDeviceMap.get(userId);
        if (deviceIds == null || deviceIds.isEmpty()) {
            return Collections.emptyList();
        }
        
        return deviceIds.stream()
                .map(this::getDevice)
                .filter(Objects::nonNull)
                .filter(d -> d.getStatus() == 1) // 只返回正常状态的设备
                .collect(Collectors.toList());
    }
    
    @Override
    public boolean isRegistered(String deviceId) {
        if (deviceId == null || deviceId.isEmpty()) {
            return false;
        }
        
        Device device = getDevice(deviceId);
        return device != null && device.getStatus() == 1;
    }
    
    @Override
    public void updateLastActiveTime(String deviceId) {
        Device device = getDevice(deviceId);
        if (device != null) {
            device.setLastActiveTime(new Date());
            deviceStore.put(deviceId, device);
            localCache.put(deviceId, device);
        }
    }
    
    @Override
    public boolean disableDevice(String deviceId) {
        Device device = getDevice(deviceId);
        if (device != null) {
            device.setStatus(0);
            deviceStore.put(deviceId, device);
            localCache.put(deviceId, device);
            System.out.println("[设备禁用] 设备已禁用: " + deviceId);
            return true;
        }
        return false;
    }
    
    @Override
    public boolean enableDevice(String deviceId) {
        Device device = getDevice(deviceId);
        if (device != null) {
            device.setStatus(1);
            device.setLastActiveTime(new Date());
            deviceStore.put(deviceId, device);
            localCache.put(deviceId, device);
            System.out.println("[设备启用] 设备已启用: " + deviceId);
            return true;
        }
        return false;
    }
    
    /**
     * 验证设备类型
     */
    private boolean isValidDeviceType(String deviceType) {
        return SystemConstants.DEVICE_TYPE_ANDROID.equals(deviceType) ||
               SystemConstants.DEVICE_TYPE_IOS.equals(deviceType) ||
               SystemConstants.DEVICE_TYPE_FLUTTER.equals(deviceType);
    }
    
    /**
     * 同步到Redis（模拟）
     */
    private void syncToRedis(Device device) {
        // 实际生产环境实现：
        // jedis.hset(SystemConstants.REDIS_KEY_DEVICE_PREFIX + device.getDeviceId(), 
        //            "data", JSON.toJSONString(device));
        // jedis.expire(SystemConstants.REDIS_KEY_DEVICE_PREFIX + device.getDeviceId(), 
        //              SystemConstants.CACHE_EXPIRE_SECONDS);
    }
    
    /**
     * 从Redis移除（模拟）
     */
    private void removeFromRedis(String deviceId) {
        // 实际生产环境实现：
        // jedis.del(SystemConstants.REDIS_KEY_DEVICE_PREFIX + deviceId);
    }
    
    /**
     * 获取统计信息
     */
    public Map<String, Object> getStatistics() {
        Map<String, Object> stats = new HashMap<>();
        stats.put("totalDevices", deviceStore.size());
        stats.put("totalUsers", userDeviceMap.size());
        stats.put("cacheSize", localCache.size());
        
        // 按设备类型统计
        Map<String, Long> deviceTypeStats = deviceStore.values().stream()
                .collect(Collectors.groupingBy(Device::getDeviceType, Collectors.counting()));
        stats.put("deviceTypeStats", deviceTypeStats);
        
        return stats;
    }
}

对外的注册接口就不贴了，就是个普通的Spring MVC接口，调用service就行。这里提一句，注册接口一定要加限流，防止被恶意刷接口，产生大量垃圾设备数据。

敲代码的同时，夸一下这个大屏，尤其是显示的一个质量，代码色彩丰富，高饱和和度高亮，简直了，看起来很舒服，不愧是明基专门为程序员打造的第一台显示器。很赞！

四、核心：长连接保持与心跳机制，连接稳了啥事都好说

连接是推送系统的根基，连接不稳，消息发得再快也没用。这部分我们踩的坑最多，也最有感触。

4.1 为什么不用TCP keepalive

我一开始也纳闷，TCP协议本身就有keepalive机制，为啥还要自己在应用层做心跳？直到线上出了“假在线”的问题：客户端和服务端都显示连接正常，但消息就是发不过去。

后来查了好久才搞明白，移动网络里运营商的NAT网关，会把一段时间没有数据传输的连接从映射表里删掉，这个超时时间一般是3-5分钟，不同运营商还不一样。而TCP keepalive默认是2小时才发一次探测包，根本赶不上NAT超时的速度。中间链路断了，两端还傻乎乎以为连接活着，就是假在线。

所以必须自己做应用层心跳，而且心跳间隔要比NAT超时短。我们当时测了三大运营商的4G/5G和主流WiFi环境，最终把心跳间隔定在了120秒，保证在NAT超时前有数据包传输，维持连接映射。

4.2 Netty选型与线程模型

做Java长连接，Netty基本是唯一选择。Tomcat那种HTTP容器，长连接性能太差，线程模型也重。Netty的NIO事件驱动模型，性能高、资源占用少，API封装得也友好。

线程模型这里也踩过坑。一开始我图省事，boss组和worker组都设了8个线程，结果压测的时候连接建立特别慢。后来才反应过来，boss组只负责接收连接，是很轻量的操作，1-2个线程完全够了，多了反而浪费。worker组负责处理连接上的读写，设成CPU核数的2倍最合适，我们8核机器就设16个线程，调整完性能直接上了一个台阶。

还有TCP参数一定要调：

• SO_REUSEADDR=true：端口复用，重启服务不用等端口释放

• TCP_NODELAY=true：禁用Nagle算法，减少小包延迟

• SO_SNDBUF/SO_RCVBUF：调整收发缓冲区，根据服务器内存来定

• SO_BACKLOG：调大连接等待队列，防止突发连接被拒绝

4.3 空闲检测与连接断开处理

用Netty自带的IdleStateHandler做空闲检测再合适不过。我们配置的是读空闲300秒：也就是5分钟没收到客户端的任何消息（包括心跳），就认为连接已经断了，主动关闭通道释放资源。

一开始我把读空闲设成了180秒，结果弱网用户特别容易被断开，网络波动一下就重连，体验很差。改成5分钟之后，给网络波动留足了缓冲，毕竟2分钟发一次心跳，连续两次没收到再断开，合理多了。

4.4 可运行代码实现

Netty自动重连、连接超时处理，这个代码是基于Netty实现高性能连接管理，使用ConcurrentHashMap存储设备ID与Channel的映射关系，并且支持高并发场景下的连接管理和提供连接状态监控和统计功能

public abstract class ConnectionManagerImpl implements ConnectionManager {
    
    // 设备ID -> Channel映射
    private final Map<String, Channel> connectionMap = new ConcurrentHashMap<>();
    
    // Channel ID -> 设备ID映射（反向映射，用于快速查找）
    private final Map<String, String> channelDeviceMap = new ConcurrentHashMap<>();
    
    // 连接数统计
    private final AtomicInteger connectionCount = new AtomicInteger(0);
    
    // 心跳超时时间（毫秒）
    private final long heartbeatTimeout;
    
    /**
     * 构造函数
     */
    public ConnectionManagerImpl(OnlineStatusManager onlineStatusManager, MonitorSystem monitorSystem) {
        this.heartbeatTimeout = SystemConstants.HEARTBEAT_INTERVAL_MS * 3;
    }
    
    /**
     * 构造函数（自定义心跳超时时间）
     */
    public ConnectionManagerImpl(long heartbeatTimeout) {
        this.heartbeatTimeout = heartbeatTimeout;
    }
    
    @Override
    public void addConnection(String deviceId, Channel channel) {
        if (deviceId == null || deviceId.isEmpty()) {
            System.err.println("[连接管理] 设备ID为空，无法添加连接");
            return;
        }
        
        if (channel == null || !channel.isActive()) {
            System.err.println("[连接管理] Channel无效，无法添加连接");
            return;
        }
        
        // 如果设备已有连接，先移除旧连接
        Channel oldChannel = connectionMap.get(deviceId);
        if (oldChannel != null && oldChannel != channel) {
            removeConnection(deviceId);
            System.out.println("[连接管理] 设备已有连接，移除旧连接: " + deviceId);
        }
        
        // 添加新连接
        connectionMap.put(deviceId, channel);
        channelDeviceMap.put(channel.id().asLongText(), deviceId);
        connectionCount.incrementAndGet();
        
        System.out.println("[连接管理] 添加连接成功: deviceId=" + deviceId + 
                          ", channelId=" + channel.id().asShortText() +
                          ", 当前连接数=" + connectionCount.get());
        
        // 添加Channel关闭监听器
        channel.closeFuture().addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                String devId = channelDeviceMap.remove(channel.id().asLongText());
                if (devId != null) {
                    connectionMap.remove(devId);
                    connectionCount.decrementAndGet();
                    System.out.println("[连接管理] 连接关闭: deviceId=" + devId + 
                                      ", 当前连接数=" + connectionCount.get());
                }
            }
        });
    }
    
    @Override
    public void removeConnection(String deviceId) {
        if (deviceId == null || deviceId.isEmpty()) {
            return;
        }
        
        Channel channel = connectionMap.remove(deviceId);
        if (channel != null) {
            channelDeviceMap.remove(channel.id().asLongText());
            connectionCount.decrementAndGet();
            
            // 关闭Channel
            if (channel.isActive()) {
                channel.close();
            }
            
            System.out.println("[连接管理] 移除连接成功: deviceId=" + deviceId + 
                              ", 当前连接数=" + connectionCount.get());
        }
    }
    
    @Override
    public Channel getConnection(String deviceId) {
        if (deviceId == null || deviceId.isEmpty()) {
            return null;
        }
        
        Channel channel = connectionMap.get(deviceId);
        
        // 检查Channel是否有效
        if (channel != null && !channel.isActive()) {
            // Channel已失效，移除连接
            removeConnection(deviceId);
            return null;
        }
        
        return channel;
    }
    
    @Override
    public boolean hasConnection(String deviceId) {
        Channel channel = getConnection(deviceId);
        return channel != null && channel.isActive();
    }
    
    @Override
    public int getConnectionCount() {
        return connectionCount.get();
    }
    
    @Override
    public boolean sendMessage(String deviceId, Object message) {
        if (deviceId == null || message == null) {
            return false;
        }
        
        Channel channel = getConnection(deviceId);
        if (channel == null || !channel.isActive()) {
            System.err.println("[连接管理] 设备连接不存在或已失效: " + deviceId);
            return false;
        }
        
        try {
            // 异步发送消息
            ChannelFuture future = channel.writeAndFlush(message);
            
            future.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (future.isSuccess()) {
                        System.out.println("[连接管理] 消息发送成功: deviceId=" + deviceId);
                    } else {
                        System.err.println("[连接管理] 消息发送失败: deviceId=" + deviceId + 
                                          ", error=" + future.cause().getMessage());
                    }
                }
            });
            
            return true;
        } catch (Exception e) {
            System.err.println("[连接管理] 消息发送异常: deviceId=" + deviceId + 
                              ", error=" + e.getMessage());
            return false;
        }
    }
    
    @Override
    public void broadcast(Object message) {
        if (message == null) {
            return;
        }
        
        System.out.println("[连接管理] 开始广播消息，当前连接数=" + connectionCount.get());
        
        int successCount = 0;
        int failCount = 0;
        
        for (Map.Entry<String, Channel> entry : connectionMap.entrySet()) {
            Channel channel = entry.getValue();
            if (channel != null && channel.isActive()) {
                try {
                    channel.writeAndFlush(message);
                    successCount++;
                } catch (Exception e) {
                    failCount++;
                    System.err.println("[连接管理] 广播消息失败: deviceId=" + entry.getKey() + 
                                      ", error=" + e.getMessage());
                }
            } else {
                failCount++;
            }
        }
        
        System.out.println("[连接管理] 广播消息完成: 成功=" + successCount + ", 失败=" + failCount);
    }
    
    @Override
    public void closeAll() {
        System.out.println("[连接管理] 开始关闭所有连接，当前连接数=" + connectionCount.get());
        
        for (Map.Entry<String, Channel> entry : connectionMap.entrySet()) {
            Channel channel = entry.getValue();
            if (channel != null && channel.isActive()) {
                try {
                    channel.close();
                } catch (Exception e) {
                    System.err.println("[连接管理] 关闭连接异常: deviceId=" + entry.getKey() + 
                                      ", error=" + e.getMessage());
                }
            }
        }
        
        connectionMap.clear();
        channelDeviceMap.clear();
        connectionCount.set(0);
        
        System.out.println("[连接管理] 所有连接已关闭");
    }
    
    /**
     * 获取所有在线设备ID
     */
    public Set<String> getOnlineDevices() {
        return connectionMap.keySet();
    }
    
    /**
     * 检查并清理无效连接
     */
    public void cleanInvalidConnections() {
        int cleanedCount = 0;
        
        for (Map.Entry<String, Channel> entry : connectionMap.entrySet()) {
            Channel channel = entry.getValue();
            if (channel == null || !channel.isActive()) {
                removeConnection(entry.getKey());
                cleanedCount++;
            }
        }
        
        if (cleanedCount > 0) {
            System.out.println("[连接管理] 清理无效连接: " + cleanedCount + "个");
        }
    }
    
    /**
     * 获取连接统计信息
     */
    public Map<String, Object> getStatistics() {
        Map<String, Object> stats = new java.util.HashMap<>();
        stats.put("totalConnections", connectionCount.get());
        stats.put("uniqueDevices", connectionMap.size());
        return stats;
    }
}

心跳管理器的代码，这个是负责维护长连接的心跳检测的，使用Netty的IdleStateHandler来实现心跳检测，并且支持配置心跳超时时间和自动断开超时连接。

public class HeartbeatManager {
    
    // 连接管理器
    private final ConnectionManagerImpl connectionManager;
    
    // 心跳间隔（毫秒）
    private final long heartbeatInterval;
    
    // 心跳超时时间（毫秒）
    private final long heartbeatTimeout;
    
    // Netty组件
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    private Channel serverChannel;
    
    /**
     * 构造函数
     */
    public HeartbeatManager(ConnectionManagerImpl connectionManager) {
        this.connectionManager = connectionManager;
        this.heartbeatInterval = SystemConstants.HEARTBEAT_INTERVAL_MS;
        this.heartbeatTimeout = SystemConstants.HEARTBEAT_INTERVAL_MS * 3;
    }
    
    /**
     * 构造函数（自定义心跳时间）
     */
    public HeartbeatManager(ConnectionManagerImpl connectionManager, 
                           long heartbeatInterval, 
                           long heartbeatTimeout) {
        this.connectionManager = connectionManager;
        this.heartbeatInterval = heartbeatInterval;
        this.heartbeatTimeout = heartbeatTimeout;
    }
    
    /**
     * 启动心跳检测服务器
     */
    public void start(int port) throws Exception {
        bossGroup = new NioEventLoopGroup(1);
        workerGroup = new NioEventLoopGroup();
        
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            
                            // 添加心跳检测Handler
                            // 读超时：3倍心跳时间没有读取数据则断开
                            // 写超时：不限制
                            // 全超时：不限制
                            pipeline.addLast("idleStateHandler", 
                                    new IdleStateHandler(
                                            heartbeatTimeout / 1000, 
                                            0, 
                                            0, 
                                            TimeUnit.SECONDS));
                            
                            // 添加编解码器
                            pipeline.addLast("decoder", new StringDecoder());
                            pipeline.addLast("encoder", new StringEncoder());
                            
                            // 添加业务Handler
                            pipeline.addLast("heartbeatHandler", new HeartbeatHandler());
                        }
                    });
            
            // 绑定端口，启动服务器
            ChannelFuture future = bootstrap.bind(port).sync();
            serverChannel = future.channel();
            
            System.out.println("[心跳管理] 心跳检测服务器启动成功，端口=" + port);
            
        } catch (Exception e) {
            System.err.println("[心跳管理] 心跳检测服务器启动失败: " + e.getMessage());
            shutdown();
            throw e;
        }
    }
    
    /**
     * 关闭心跳检测服务器
     */
    public void shutdown() {
        System.out.println("[心跳管理] 开始关闭心跳检测服务器...");
        
        if (serverChannel != null) {
            serverChannel.close();
        }
        
        if (workerGroup != null) {
            workerGroup.shutdownGracefully();
        }
        
        if (bossGroup != null) {
            bossGroup.shutdownGracefully();
        }
        
        System.out.println("[心跳管理] 心跳检测服务器已关闭");
    }
    
    /**
     * 心跳Handler
     * 处理心跳消息和超时检测
     */
    private class HeartbeatHandler extends SimpleChannelInboundHandler<String> {
        
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
            // 处理心跳消息
            if ("HEARTBEAT".equals(msg) || "PING".equals(msg)) {
                // 收到心跳，回复PONG
                ctx.writeAndFlush("PONG");
                System.out.println("[心跳管理] 收到心跳消息: " + ctx.channel().id().asShortText());
            } else {
                // 非心跳消息，传递给下一个Handler
                ctx.fireChannelRead(msg);
            }
        }
        
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent event = (IdleStateEvent) evt;
                
                if (event.state() == IdleState.READER_IDLE) {
                    // 读超时，断开连接
                    System.err.println("[心跳管理] 连接超时，断开连接: " + ctx.channel().id().asShortText());
                    ctx.channel().close();
                }
            } else {
                super.userEventTriggered(ctx, evt);
            }
        }
        
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("[心跳管理] 新连接建立: " + ctx.channel().id().asShortText());
            super.channelActive(ctx);
        }
        
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("[心跳管理] 连接断开: " + ctx.channel().id().asShortText());
            super.channelInactive(ctx);
        }
        
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            System.err.println("[心跳管理] 连接异常: " + cause.getMessage());
            ctx.channel().close();
        }
    }
    
    /**
     * 发送心跳消息
     */
    public void sendHeartbeat(String deviceId) {
        if (connectionManager.hasConnection(deviceId)) {
            connectionManager.sendMessage(deviceId, "PING");
        }
    }
    
    /**
     * 批量发送心跳
     */
    public void broadcastHeartbeat() {
        connectionManager.broadcast("PING");
    }
    
    /**
     * 获取心跳统计信息
     */
    public Map<String, Object> getStatistics() {
        Map<String, Object> stats = new java.util.HashMap<>();
        stats.put("heartbeatInterval", heartbeatInterval);
        stats.put("heartbeatTimeout", heartbeatTimeout);
        stats.put("activeConnections", connectionManager.getConnectionCount());
        return stats;
    }
}

这里提一句协议的坑：第一版我们用JSON做消息体，简单是简单，但10万连接的时候，光心跳包的带宽就占了不少。后来换成Protobuf序列化，消息体积直接缩小了70%，带宽压力一下就小了。还有协议一定要加版本号！我们第一版没加，后来加新字段老客户端直接解析失败，兼容起来很难很难。

编码模式可以通过箭头那里的热键来切换，可以明显的看到效果：普通的显示器泛白发虚，明基显示屏深邃且代码突出。这个显示器的编程色彩模式还可以适配不同的背景，深色主题和彩纸模式适配黑色背景，亮色主题适配白色背景，看起来代码都很清晰。我最近很爱用彩纸模式，不光代码清晰，看起来还有模拟纸张的感觉，不管是看文档还是代码，看久了都不会累

代码显示效果：

文档显示效果：

4.5 明基专业编程显示器——自研 Display Pilot 2

对于长期编码的开发者而言，一台专业显示器的价值，从来不止于硬件参数本身，配套软件的适配深度，才是直接决定日常开发的效率上限。明基 RD270Q 搭配的自研 Display Pilot 2 软件，正是以软硬件协同的方式，从操作、显示、场景适配多个维度，为我们开发工作带来可感知的效率提升。

分辨率滑块

可以通过直接拖动滑块来调整显示器和笔记本电脑的屏幕分辨率

夜间保护

通过提供最低亮度的技术来保护眼睛。其智能环境光检测和自动 切换功能可确保轻松和持续的护眼。

Visual Optimizer （原 B.I. 第二代）

这个是智慧调光第二代（B.I. 第二代）可根据环境光和图像对比度自动平衡屏幕亮度。 智慧调光 Plus 第二代（B.I.+ 第二代）和 Visual Optimizer 提供高级控制，可同时 进行亮度和色温微调，或仅进这些功能提高了眼睛的舒适度，同时保留了良好图像

低蓝光

低蓝光 / 智慧蓝光技术有助于降低显示器发出的蓝光可降低眼睛暴露在蓝光下的风 险

EcoPrivacy

EcoPrivacy 有助于节能，并在不活动时通过调暗显示器来保护屏幕隐私。

Display Pilot 2 这款软件针对开发者最担心的"思路被打断"问题，当打开IDE时会自动进入编程模式，优化文字对比度使代码显示更加清晰锐利；切换到文档时则自动启用阅读模式，降低蓝光减轻视觉疲劳。整个过程完全无需手动操作OSD菜单，确保开发者能够保持专注，工作节奏不受干扰。

**Display Pilot 2 **还支持快捷键定制、应用参数记忆和多屏协同管理等功能，一键即可完成亮度调节、输入源切换和布局调整等常用操作。配合 RD270Q 优质显示屏，这套软硬件结合方案不仅能大幅减少无效操作时间，其智能护眼和精准显示优化功能还能有效延长开发者的专注时长，成为提升编码的得力助手，为开发工作带来可感知的效率提升。

五、状态管理：用户在不在线，这事必须门儿清

推送的核心逻辑很简单：用户在线就走长连实时发，不在线就存离线或者走厂商通道。所以用户在不在线、连在哪个网关节点上，必须精准知道。

5.1 存储结构设计

我们用Redis存所有在线状态，毕竟读写快，适合这种高频访问的热点数据。核心存三类数据：

1. 设备维度：设备ID对应网关节点ID、在线状态、上线时间，用Hash结构

2. 用户维度：用户ID绑定的所有设备ID集合，用Set结构，方便按用户推送

3. 网关维度：每个网关节点上的所有在线设备ID集合，用Set结构，方便网关宕机时批量处理

5.2 踩过的最大坑：网关宕机后的假在线

这是我印象最深的一个坑。有一次测试环境的网关机器突然宕机，JVM直接挂了，根本没机会执行连接断开的逻辑。结果Redis里的设备状态全还是“在线”，消息源源不断往这个死网关发，全部石沉大海。

后来我们做了双保险：
第一，给每个设备的在线状态加300秒过期时间，每个网关每隔2分钟给本节点的所有在线设备续一次期。网关宕机了没人续期，5分钟后状态自动过期。
第二，加了网关心跳上报机制，每个网关每分钟上报一次心跳。如果某个网关超过5分钟没上报，定时任务就把这个网关上的所有设备批量置为离线。

经过上面的这两套机制下来，基本不会再出现长时间假在线的情况。

5.3 并发一致性问题

还有个容易忽略的点：同一个设备可能同时在两个地方发起连接，一个上线一个下线，很容易出现状态错乱。我们一开始没处理，压测的时候经常出现“设备实际在线但状态显示离线”的问题。

解决方案也不复杂：更新设备状态的时候用Redis分布式锁做一下控制，保证同一时间只有一个操作能更新状态。其实场景不复杂的话，用SETNX也能搞定，不用上太重的分布式锁。

5.4 可运行代码实现

使用Redis存储分布式在线状态，做了本地缓存提高查询性能和也做了定时任务清理超时连接，代码是基于Redis和本地缓存实现在线状态管理的

public abstract class OnlineStatusManagerImpl implements OnlineStatusManager {
    
    // 用户在线状态（userId -> status）
    private final Map<String, Integer> userStatusMap = new ConcurrentHashMap<>();
    
    // 用户在线设备（userId -> Set<deviceId>）
    private final Map<String, Set<String>> userDeviceMap = new ConcurrentHashMap<>();
    
    // 设备在线状态（deviceId -> userId）
    private final Map<String, String> deviceUserMap = new ConcurrentHashMap<>();
    
    // 用户最后活跃时间（userId -> lastActiveTime）
    private final Map<String, Long> lastActiveTimeMap = new ConcurrentHashMap<>();
    
    // 本地缓存
    private final Cache<Object, Object> statusCache;
    
    // 超时时间（毫秒）
    private final long timeoutMs;
    
    /**
     * 构造函数
     */
    public OnlineStatusManagerImpl() {
        this.timeoutMs = SystemConstants.HEARTBEAT_INTERVAL_MS * 3;
        
        // 初始化本地缓存
        this.statusCache = Caffeine.newBuilder()
                .maximumSize(SystemConstants.CACHE_MAX_SIZE)
                .expireAfterWrite(SystemConstants.CACHE_EXPIRE_SECONDS, TimeUnit.SECONDS)
                .build();
    }
    
    /**
     * 构造函数（自定义超时时间）
     */
    public OnlineStatusManagerImpl(long timeoutMs) {
        this.timeoutMs = timeoutMs;
        this.statusCache = Caffeine.newBuilder()
                .maximumSize(SystemConstants.CACHE_MAX_SIZE)
                .expireAfterWrite(SystemConstants.CACHE_EXPIRE_SECONDS, TimeUnit.SECONDS)
                .build();
    }
    
    @Override
    public boolean online(String userId, String deviceId) {
        if (userId == null || userId.isEmpty() || deviceId == null || deviceId.isEmpty()) {
            return false;
        }
        
        // 更新用户在线状态
        userStatusMap.put(userId, SystemConstants.ONLINE_STATUS_ONLINE);
        
        // 添加设备到用户设备列表
        userDeviceMap.computeIfAbsent(userId, k -> ConcurrentHashMap.newKeySet())
                .add(deviceId);
        
        // 记录设备-用户映射
        deviceUserMap.put(deviceId, userId);
        
        // 更新最后活跃时间
        long now = System.currentTimeMillis();
        lastActiveTimeMap.put(userId, now);
        
        // 更新本地缓存
        statusCache.put(userId, SystemConstants.ONLINE_STATUS_ONLINE);
        
        // 同步到Redis（实际生产环境）
        syncToRedis(userId, deviceId, true);
        
        System.out.println("[在线状态] 用户上线: userId=" + userId + ", deviceId=" + deviceId);
        
        return true;
    }
    
    @Override
    public boolean offline(String userId, String deviceId) {
        if (userId == null || deviceId == null) {
            return false;
        }
        
        // 从用户设备列表中移除设备
        Set<String> devices = userDeviceMap.get(userId);
        if (devices != null) {
            devices.remove(deviceId);
            
            // 如果用户没有在线设备，则设置为离线
            if (devices.isEmpty()) {
                userDeviceMap.remove(userId);
                userStatusMap.put(userId, SystemConstants.ONLINE_STATUS_OFFLINE);
                lastActiveTimeMap.remove(userId);
                
                // 更新本地缓存
                statusCache.put(userId, SystemConstants.ONLINE_STATUS_OFFLINE);
            }
        }
        
        // 移除设备-用户映射
        deviceUserMap.remove(deviceId);
        
        // 同步到Redis
        syncToRedis(userId, deviceId, false);
        
        System.out.println("[在线状态] 用户下线: userId=" + userId + ", deviceId=" + deviceId);
        
        return true;
    }
    
    @Override
    public boolean isOnline(String userId) {
        if (userId == null || userId.isEmpty()) {
            return false;
        }
        
        // 先从本地缓存获取
        Integer status = (Integer) statusCache.getIfPresent(userId);
        if (status != null) {
            return status == SystemConstants.ONLINE_STATUS_ONLINE;
        }
        
        // 从存储获取
        status = userStatusMap.get(userId);
        if (status != null) {
            // 更新本地缓存
            statusCache.put(userId, status);
            return status == SystemConstants.ONLINE_STATUS_ONLINE;
        }
        
        return false;
    }
    
    @Override
    public boolean isDeviceOnline(String deviceId) {
        if (deviceId == null || deviceId.isEmpty()) {
            return false;
        }
        
        return deviceUserMap.containsKey(deviceId);
    }
    
    @Override
    public int getOnlineStatus(String userId) {
        if (userId == null || userId.isEmpty()) {
            return SystemConstants.ONLINE_STATUS_OFFLINE;
        }
        
        // 先从本地缓存获取
        Integer status = (Integer) statusCache.getIfPresent(userId);
        if (status != null) {
            return status;
        }
        
        // 从存储获取
        status = userStatusMap.get(userId);
        if (status != null) {
            statusCache.put(userId, status);
            return status;
        }
        
        return SystemConstants.ONLINE_STATUS_OFFLINE;
    }
    
    @Override
    public boolean setOnlineStatus(String userId, int status) {
        if (userId == null || userId.isEmpty()) {
            return false;
        }
        
        // 验证状态值
        if (status < SystemConstants.ONLINE_STATUS_OFFLINE || 
            status > SystemConstants.ONLINE_STATUS_BUSY) {
            return false;
        }
        
        userStatusMap.put(userId, status);
        statusCache.put(userId, status);
        
        System.out.println("[在线状态] 更新用户状态: userId=" + userId + ", status=" + status);
        
        return true;
    }
    
    @Override
    public List<String> getOnlineDevices(String userId) {
        if (userId == null || userId.isEmpty()) {
            return Collections.emptyList();
        }
        
        Set<String> devices = userDeviceMap.get(userId);
        if (devices == null || devices.isEmpty()) {
            return Collections.emptyList();
        }
        
        return new ArrayList<>(devices);
    }
    
    @Override
    public List<String> getAllOnlineUsers() {
        return userStatusMap.entrySet().stream()
                .filter(entry -> entry.getValue() == SystemConstants.ONLINE_STATUS_ONLINE)
                .map(Map.Entry::getKey)
                .collect(Collectors.toList());
    }
    
    @Override
    public int getOnlineUserCount() {
        return (int) userStatusMap.entrySet().stream()
                .filter(entry -> entry.getValue() == SystemConstants.ONLINE_STATUS_ONLINE)
                .count();
    }
    
    @Override
    public int getOnlineDeviceCount() {
        return deviceUserMap.size();
    }
    
    @Override
    public void updateLastActiveTime(String userId) {
        if (userId != null && !userId.isEmpty()) {
            lastActiveTimeMap.put(userId, System.currentTimeMillis());
        }
    }
    
    @Override
    public void cleanTimeoutConnections() {
        long now = System.currentTimeMillis();
        int cleanedCount = 0;
        
        // 检查所有用户的最后活跃时间
        Iterator<Map.Entry<String, Long>> iterator = lastActiveTimeMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, Long> entry = iterator.next();
            String userId = entry.getKey();
            Long lastActiveTime = entry.getValue();
            
            if (now - lastActiveTime > timeoutMs) {
                // 超时，设置为离线
                Set<String> devices = userDeviceMap.get(userId);
                if (devices != null) {
                    for (String deviceId : devices) {
                        deviceUserMap.remove(deviceId);
                    }
                    userDeviceMap.remove(userId);
                }
                
                userStatusMap.put(userId, SystemConstants.ONLINE_STATUS_OFFLINE);
                statusCache.put(userId, SystemConstants.ONLINE_STATUS_OFFLINE);
                iterator.remove();
                
                cleanedCount++;
                System.out.println("[在线状态] 清理超时用户: userId=" + userId);
            }
        }
        
        if (cleanedCount > 0) {
            System.out.println("[在线状态] 清理超时连接完成: " + cleanedCount + "个");
        }
    }
    
    /**
     * 同步到Redis（模拟）
     */
    private void syncToRedis(String userId, String deviceId, boolean online) {
        // 实际生产环境实现：
        // String key = SystemConstants.REDIS_KEY_ONLINE_PREFIX + userId;
        // if (online) {
        //     jedis.sadd(key, deviceId);
        //     jedis.expire(key, SystemConstants.CACHE_EXPIRE_SECONDS);
        // } else {
        //     jedis.srem(key, deviceId);
        // }
    }
    
    /**
     * 获取统计信息
     */
    public Map<String, Object> getStatistics() {
        Map<String, Object> stats = new HashMap<>();
        stats.put("onlineUserCount", getOnlineUserCount());
        stats.put("onlineDeviceCount", getOnlineDeviceCount());
        stats.put("totalUsers", userStatusMap.size());
        stats.put("totalDevices", deviceUserMap.size());
        stats.put("cacheSize", statusCache.size());
        
        // 按状态统计
        Map<Integer, Long> statusStats = userStatusMap.values().stream()
                .collect(Collectors.groupingBy(status -> status, Collectors.counting()));
        stats.put("statusStats", statusStats);
        
        return stats;
    }
}

六、消息路由：消息怎么准确送到用户手上？

消息从业务方发过来，怎么准确找到目标设备，再送到对应的网关节点，这就是消息路由要干的事。

6.1 整体路由流程

我们最常用的是按用户ID推送，整体流程大概是这样：

1. 业务方调用推送接口，传用户ID列表、消息内容、过期时间等

2. 消息服务生成唯一消息ID，写入Kafka做异步解耦

3. 路由服务消费消息，对每个用户ID查询他的所有在线设备

4. 对每个在线设备，查到它所在的网关节点，把消息发到对应网关的Kafka分区

5. 网关消费自己分区的消息，通过长连接发给设备

6. 设备不在线的话，消息存入离线缓存，或者走厂商推送

6.2 为什么做两次Kafka分发？

一开始我们只有一个Kafka topic，所有网关都消费同一个topic。每条消息每个网关都要拿过来判断“是不是我的设备”，10个网关就重复消费10次，既浪费资源又增加延迟。

后来我们改成了按网关分区：每个网关对应一个Kafka分区，路由服务根据网关ID把消息发到对应分区，网关只消费自己的分区，消息只被消费一次。就这一个改动，整体吞吐量提升了40%多。

实现也简单，自定义一个Kafka分区器，按网关ID哈希取模就行。

6.3 消息顺序与幂等

两个很重要的点：顺序性和幂等性。

• 顺序性：同一个用户的消息必须按发送顺序到达，不然用户先收到B再收到A，逻辑就乱了。我们第一次Kafka分发的时候按用户ID分区，同一个用户的消息都进同一个分区，由同一个消费者处理，保证顺序。

• 幂等性：网络环境复杂，消息可能重复发送，所以每条消息都有全局唯一的msgId，客户端根据msgId去重，保证用户只看到一次。

6.4 可运行代码实现

这个代码是推送网关实现类，主要功能是负责消息的实际推送发送，不仅支持多种推送通道，根据设备类型选择合适的推送通道，同事也支持在线推送和离线推送并且提供推送统计和监控

public class PushGatewayImpl implements PushGateway {
    
    // 连接管理器
    private final ConnectionManager connectionManager;
    
    // 设备注册器
    private final DeviceRegistry deviceRegistry;
    
    // 在线状态管理器
    private final OnlineStatusManager onlineStatusManager;
    
    // 推送统计
    private final AtomicLong totalPushCount = new AtomicLong(0);
    private final AtomicLong successCount = new AtomicLong(0);
    private final AtomicLong failCount = new AtomicLong(0);
    
    /**
     * 构造函数
     */
    public PushGatewayImpl(ConnectionManager connectionManager,
                          DeviceRegistry deviceRegistry,
                          OnlineStatusManager onlineStatusManager) {
        this.connectionManager = connectionManager;
        this.deviceRegistry = deviceRegistry;
        this.onlineStatusManager = onlineStatusManager;
    }
    
    @Override
    public Result<Boolean> push(Message message) {
        if (message == null) {
            return Result.badRequest("消息不能为空");
        }
        
        totalPushCount.incrementAndGet();
        
        try {
            // 1. 检查消息是否过期
            if (message.isExpired()) {
                failCount.incrementAndGet();
                return Result.fail("消息已过期");
            }
            
            // 2. 获取目标设备
            String receiverId = message.getReceiverId();
            if (receiverId == null || receiverId.isEmpty()) {
                failCount.incrementAndGet();
                return Result.badRequest("接收者ID不能为空");
            }
            
            // 3. 推送到用户所有设备
            Result<Boolean> result = pushToUser(receiverId, message);
            
            if (result.isSuccess()) {
                successCount.incrementAndGet();
            } else {
                failCount.incrementAndGet();
            }
            
            return result;
            
        } catch (Exception e) {
            failCount.incrementAndGet();
            System.err.println("[推送网关] 推送失败: " + e.getMessage());
            return Result.fail("推送失败: " + e.getMessage());
        }
    }
    
    @Override
    public Result<Boolean> pushToDevice(String deviceId, Message message) {
        if (deviceId == null || deviceId.isEmpty()) {
            return Result.badRequest("设备ID不能为空");
        }
        
        if (message == null) {
            return Result.badRequest("消息不能为空");
        }
        
        try {
            // 1. 检查设备是否注册
            if (!deviceRegistry.isRegistered(deviceId)) {
                return Result.notFound("设备未注册: " + deviceId);
            }
            
            // 2. 检查设备是否在线
            boolean isOnline = onlineStatusManager.isDeviceOnline(deviceId);
            
            if (isOnline) {
                // 在线推送：通过长连接推送
                boolean sent = connectionManager.sendMessage(deviceId, message);
                
                if (sent) {
                    System.out.println("[推送网关] 在线推送成功: deviceId=" + deviceId + 
                                      ", messageId=" + message.getMessageId());
                    return Result.success("推送成功", true);
                } else {
                    System.err.println("[推送网关] 在线推送失败: deviceId=" + deviceId);
                    return Result.fail("推送失败");
                }
            } else {
                // 离线推送：通过厂商通道推送
                return pushViaVendorChannel(deviceId, message);
            }
            
        } catch (Exception e) {
            System.err.println("[推送网关] 推送到设备失败: deviceId=" + deviceId + 
                              ", error=" + e.getMessage());
            return Result.fail("推送失败: " + e.getMessage());
        }
    }
    
    @Override
    public Result<Boolean> pushToUser(String userId, Message message) {
        if (userId == null || userId.isEmpty()) {
            return Result.badRequest("用户ID不能为空");
        }
        
        if (message == null) {
            return Result.badRequest("消息不能为空");
        }
        
        try {
            // 1. 获取用户所有设备
            List<test_project.push.common.entity.Device> devices = 
                    deviceRegistry.getDevicesByUserId(userId);
            
            if (devices == null || devices.isEmpty()) {
                return Result.notFound("用户没有注册设备: " + userId);
            }
            
            // 2. 推送到每个设备
            int successDevices = 0;
            int failDevices = 0;
            
            for (test_project.push.common.entity.Device device : devices) {
                Result<Boolean> result = pushToDevice(device.getDeviceId(), message);
                
                if (result.isSuccess()) {
                    successDevices++;
                } else {
                    failDevices++;
                }
            }
            
            // 3. 返回结果
            if (successDevices > 0) {
                System.out.println("[推送网关] 推送到用户成功: userId=" + userId + 
                                  ", success=" + successDevices + 
                                  ", fail=" + failDevices);
                return Result.success("推送成功", true);
            } else {
                return Result.fail("推送到所有设备失败");
            }
            
        } catch (Exception e) {
            System.err.println("[推送网关] 推送到用户失败: userId=" + userId + 
                              ", error=" + e.getMessage());
            return Result.fail("推送失败: " + e.getMessage());
        }
    }
    
    @Override
    public Result<Boolean> pushBatch(List<Message> messages) {
        if (messages == null || messages.isEmpty()) {
            return Result.badRequest("消息列表不能为空");
        }
        
        int successCount = 0;
        int failCount = 0;
        
        for (Message message : messages) {
            Result<Boolean> result = push(message);
            if (result.isSuccess()) {
                successCount++;
            } else {
                failCount++;
            }
        }
        
        System.out.println("[推送网关] 批量推送完成: success=" + successCount + 
                          ", fail=" + failCount);
        
        return Result.success("批量推送完成", true);
    }
    
    @Override
    public Result<Boolean> broadcast(Message message) {
        if (message == null) {
            return Result.badRequest("消息不能为空");
        }
        
        try {
            // 广播消息到所有在线设备
            connectionManager.broadcast(message);
            
            System.out.println("[推送网关] 广播消息成功: messageId=" + message.getMessageId());
            
            return Result.success("广播成功", true);
            
        } catch (Exception e) {
            System.err.println("[推送网关] 广播失败: " + e.getMessage());
            return Result.fail("广播失败: " + e.getMessage());
        }
    }
    
    @Override
    public boolean isChannelAvailable(String channel) {
        // 检查推送通道是否可用
        // 实际生产环境需要检查各厂商通道的可用性
        return true;
    }
    
    @Override
    public Map<String, Object> getStatistics() {
        Map<String, Object> stats = new HashMap<>();
        stats.put("totalPushCount", totalPushCount.get());
        stats.put("successCount", successCount.get());
        stats.put("failCount", failCount.get());
        stats.put("successRate", 
                totalPushCount.get() > 0 ? 
                (double) successCount.get() / totalPushCount.get() * 100 : 0);
        return stats;
    }
    
    /**
     * 通过厂商通道推送
     * 支持APNS、FCM、小米、华为等厂商通道
     */
    private Result<Boolean> pushViaVendorChannel(String deviceId, Message message) {
        // 获取设备信息
        test_project.push.common.entity.Device device = deviceRegistry.getDevice(deviceId);
        
        if (device == null) {
            return Result.notFound("设备不存在: " + deviceId);
        }
        
        String deviceType = device.getDeviceType();
        String pushChannel = device.getPushChannel();
        
        System.out.println("[推送网关] 离线推送: deviceId=" + deviceId + 
                          ", deviceType=" + deviceType + 
                          ", channel=" + pushChannel);
        
        // 根据设备类型选择推送通道
        switch (deviceType) {
            case SystemConstants.DEVICE_TYPE_IOS:
                return pushViaAPNS(device, message);
                
            case SystemConstants.DEVICE_TYPE_ANDROID:
                return pushViaAndroidChannel(device, message);
                
            case SystemConstants.DEVICE_TYPE_FLUTTER:
                // Flutter应用根据平台选择通道
                if ("apns".equals(pushChannel)) {
                    return pushViaAPNS(device, message);
                } else {
                    return pushViaAndroidChannel(device, message);
                }
                
            default:
                return Result.fail("不支持的设备类型: " + deviceType);
        }
    }
    
    /**
     * 通过APNS推送（iOS）
     */
    private Result<Boolean> pushViaAPNS(test_project.push.common.entity.Device device, 
                                        Message message) {
        // 实际生产环境实现：
        // 1. 构建APNS推送payload
        // 2. 调用APNS服务推送
        // 3. 处理推送结果
        
        System.out.println("[推送网关] APNS推送: deviceToken=" + device.getDeviceToken() + 
                          ", title=" + message.getTitle());
        
        // 模拟推送成功
        return Result.success("APNS推送成功", true);
    }
    
    /**
     * 通过Android厂商通道推送
     */
    private Result<Boolean> pushViaAndroidChannel(test_project.push.common.entity.Device device, 
                                                   Message message) {
        String pushChannel = device.getPushChannel();
        
        // 根据厂商通道选择推送方式
        if (pushChannel == null || pushChannel.isEmpty()) {
            pushChannel = "fcm"; // 默认使用FCM
        }
        
        switch (pushChannel) {
            case "fcm":
                return pushViaFCM(device, message);
                
            case "xiaomi":
                return pushViaXiaomi(device, message);
                
            case "huawei":
                return pushViaHuawei(device, message);
                
            case "oppo":
                return pushViaOppo(device, message);
                
            case "vivo":
                return pushViaVivo(device, message);
                
            default:
                return pushViaFCM(device, message);
        }
    }
    
    /**
     * 通过FCM推送
     */
    private Result<Boolean> pushViaFCM(test_project.push.common.entity.Device device, 
                                       Message message) {
        // 实际生产环境实现：
        // 1. 构建FCM推送消息
        // 2. 调用FCM API推送
        // 3. 处理推送结果
        
        System.out.println("[推送网关] FCM推送: deviceToken=" + device.getDeviceToken() + 
                          ", title=" + message.getTitle());
        
        return Result.success("FCM推送成功", true);
    }
    
    /**
     * 通过小米推送
     */
    private Result<Boolean> pushViaXiaomi(test_project.push.common.entity.Device device, 
                                          Message message) {
        System.out.println("[推送网关] 小米推送: deviceToken=" + device.getDeviceToken());
        return Result.success("小米推送成功", true);
    }
    
    /**
     * 通过华为推送
     */
    private Result<Boolean> pushViaHuawei(test_project.push.common.entity.Device device, 
                                          Message message) {
        System.out.println("[推送网关] 华为推送: deviceToken=" + device.getDeviceToken());
        return Result.success("华为推送成功", true);
    }
    
    /**
     * 通过OPPO推送
     */
    private Result<Boolean> pushViaOppo(test_project.push.common.entity.Device device, 
                                        Message message) {
        System.out.println("[推送网关] OPPO推送: deviceToken=" + device.getDeviceToken());
        return Result.success("OPPO推送成功", true);
    }
    
    /**
     * 通过VIVO推送
     */
    private Result<Boolean> pushViaVivo(test_project.push.common.entity.Device device, 
                                        Message message) {
        System.out.println("[推送网关] VIVO推送: deviceToken=" + device.getDeviceToken());
        return Result.success("VIVO推送成功", true);
    }
}

七、离线缓存：用户不在线，消息先存着等他来拿

用户不可能一直在线，APP退后台、没网络都是常事，所以离线消息是也是必须要的。

7.1 从MySQL到分级存储的演进

一开始图省事，离线消息直接存MySQL。上线没多久就出问题了：一是数据量涨得太快，单表千万级之后查询越来越慢；二是有些用户几个月不登录，离线消息攒了几千条，一上线全拉下来，直接把APP卡死了

后来我们做了分级存储优化：

1. 热数据存Redis：7天以内的离线消息，消息ID存在Redis的ZSet里，按时间戳排序，读写快，用户上线拉取无延迟。消息内容存在MongoDB，Redis只存ID，节省内存。

2. 冷数据归档：超过7天的消息归档到MongoDB，一般用户很少拉7天前的消息，慢点也能接受。

3. 数量上限控制：单个用户最多存200条离线消息，超过就删最早的，防止存储爆炸，也避免用户上线拉取压力太大。

4. 分页拉取：一次最多拉50条，拉完确认了再拉下一批，不会一下全灌给客户端。

7.2 离线消息的一致性问题

有个场景很经典：路由服务判断设备不在线，刚把消息存进离线，设备就上线了，同时在线推送也到了，用户就会收到两条重复消息。

这个问题没法100%避免，所以我们的原则是：服务端保证至少一次投递，客户端保证幂等去重。客户端是最后一道防线，一定要根据msgId做去重，已经收到过的就不展示。

7.3 可运行代码实现

是通过Redis和本地缓存实现离线消息存储，使用Redis List存储离线消息支持高并发，使用本地缓存提高查询性能、消息过期清理、消息已读未读状态

public abstract class OfflineCacheImpl implements OfflineCache {
    
    // 离线消息存储（userId -> List<Message>）
    private final Map<String, List<Message>> offlineMessageMap = new ConcurrentHashMap<>();
    
    // 消息已读状态（userId_messageId -> isRead）
    private final Map<String, Boolean> messageReadStatus = new ConcurrentHashMap<>();
    
    // 本地缓存
    private final Cache<Object, Object> localCache;
    
    // 最大离线消息数量
    private final int maxOfflineMessages;
    
    // 消息过期时间（毫秒）
    private final long messageExpireTime;
    
    /**
     * 构造函数
     */
    public OfflineCacheImpl() {
        this.maxOfflineMessages = 1000;
        this.messageExpireTime = 7 * 24 * 60 * 60 * 1000L; // 7天
        
        // 初始化本地缓存
        this.localCache = Caffeine.newBuilder()
                .maximumSize(SystemConstants.CACHE_MAX_SIZE)
                .expireAfterWrite(SystemConstants.CACHE_EXPIRE_SECONDS, TimeUnit.SECONDS)
                .build();
    }
    
    /**
     * 构造函数（自定义配置）
     */
    public OfflineCacheImpl(int maxOfflineMessages, long messageExpireTime) {
        this.maxOfflineMessages = maxOfflineMessages;
        this.messageExpireTime = messageExpireTime;
        this.localCache = Caffeine.newBuilder()
                .maximumSize(SystemConstants.CACHE_MAX_SIZE)
                .expireAfterWrite(SystemConstants.CACHE_EXPIRE_SECONDS, TimeUnit.SECONDS)
                .build();
    }
    
    @Override
    public boolean store(String userId, Message message) {
        if (userId == null || userId.isEmpty() || message == null) {
            return false;
        }
        
        try {
            // 获取用户离线消息列表
            List<Message> messages = offlineMessageMap.computeIfAbsent(
                    userId, 
                    k -> Collections.synchronizedList(new ArrayList<>())
            );
            
            // 检查是否超过最大数量
            if (messages.size() >= maxOfflineMessages) {
                // 移除最早的消息
                messages.remove(0);
            }
            
            // 标记为离线消息
            message.setIsOffline(true);
            
            // 添加消息
            messages.add(message);
            
            // 标记为未读
            messageReadStatus.put(userId + "_" + message.getMessageId(), false);
            
            // 更新本地缓存
            localCache.put(userId, messages);
            
            // 同步到Redis（实际生产环境）
            syncToRedis(userId, message);
            
            System.out.println("[离线缓存] 存储离线消息: userId=" + userId + 
                              ", messageId=" + message.getMessageId());
            
            return true;
            
        } catch (Exception e) {
            System.err.println("[离线缓存] 存储离线消息失败: userId=" + userId + 
                              ", error=" + e.getMessage());
            return false;
        }
    }
    
    @Override
    public boolean storeBatch(String userId, List<Message> messages) {
        if (userId == null || userId.isEmpty() || messages == null || messages.isEmpty()) {
            return false;
        }
        
        for (Message message : messages) {
            store(userId, message);
        }
        
        return true;
    }
    
    @Override
    public List<Message> getOfflineMessages(String userId) {
        if (userId == null || userId.isEmpty()) {
            return Collections.emptyList();
        }
        
        // 先从本地缓存获取
        List<Message> messages = (List<Message>) localCache.getIfPresent(userId);
        if (messages != null) {
            return new ArrayList<>(messages);
        }
        
        // 从存储获取
        messages = offlineMessageMap.get(userId);
        if (messages != null) {
            // 过滤过期消息
            messages = messages.stream()
                    .filter(msg -> !msg.isExpired())
                    .collect(Collectors.toList());
            
            // 更新本地缓存
            localCache.put(userId, messages);
            
            return new ArrayList<>(messages);
        }
        
        return Collections.emptyList();
    }
    
    @Override
    public List<Message> getOfflineMessages(String userId, int offset, int limit) {
        List<Message> allMessages = getOfflineMessages(userId);
        
        if (allMessages.isEmpty()) {
            return Collections.emptyList();
        }
        
        // 分页处理
        int start = Math.min(offset, allMessages.size());
        int end = Math.min(offset + limit, allMessages.size());
        
        return allMessages.subList(start, end);
    }
    
    @Override
    public boolean delete(String userId, String messageId) {
        if (userId == null || messageId == null) {
            return false;
        }
        
        List<Message> messages = offlineMessageMap.get(userId);
        if (messages == null) {
            return false;
        }
        
        boolean removed = messages.removeIf(msg -> messageId.equals(msg.getMessageId()));
        
        if (removed) {
            // 移除已读状态
            messageReadStatus.remove(userId + "_" + messageId);
            
            // 更新本地缓存
            localCache.put(userId, messages);
            
            // 从Redis移除（实际生产环境）
            removeFromRedis(userId, messageId);
            
            System.out.println("[离线缓存] 删除离线消息: userId=" + userId + 
                              ", messageId=" + messageId);
        }
        
        return removed;
    }
    
    @Override
    public int deleteAll(String userId) {
        if (userId == null || userId.isEmpty()) {
            return 0;
        }
        
        List<Message> messages = offlineMessageMap.remove(userId);
        int count = messages != null ? messages.size() : 0;
        
        if (count > 0) {
            // 移除已读状态
            messages.forEach(msg -> 
                    messageReadStatus.remove(userId + "_" + msg.getMessageId()));
            
            // 清除本地缓存
            localCache.invalidate(userId);
            
            System.out.println("[离线缓存] 删除所有离线消息: userId=" + userId + 
                              ", count=" + count);
        }
        
        return count;
    }
    
    @Override
    public int getOfflineMessageCount(String userId) {
        if (userId == null || userId.isEmpty()) {
            return 0;
        }
        
        List<Message> messages = offlineMessageMap.get(userId);
        return messages != null ? messages.size() : 0;
    }
    
    @Override
    public boolean hasOfflineMessages(String userId) {
        return getOfflineMessageCount(userId) > 0;
    }
    
    @Override
    public boolean markAsRead(String userId, String messageId) {
        if (userId == null || messageId == null) {
            return false;
        }
        
        messageReadStatus.put(userId + "_" + messageId, true);
        
        System.out.println("[离线缓存] 标记消息已读: userId=" + userId + 
                          ", messageId=" + messageId);
        
        return true;
    }
    
    @Override
    public int getUnreadCount(String userId) {
        if (userId == null || userId.isEmpty()) {
            return 0;
        }
        
        List<Message> messages = offlineMessageMap.get(userId);
        if (messages == null || messages.isEmpty()) {
            return 0;
        }
        
        return (int) messages.stream()
                .filter(msg -> {
                    Boolean isRead = messageReadStatus.get(userId + "_" + msg.getMessageId());
                    return isRead == null || !isRead;
                })
                .count();
    }
    
    @Override
    public int cleanExpiredMessages() {
        int cleanedCount = 0;
        long now = System.currentTimeMillis();
        
        for (Map.Entry<String, List<Message>> entry : offlineMessageMap.entrySet()) {
            String userId = entry.getKey();
            List<Message> messages = entry.getValue();
            
            int beforeSize = messages.size();
            messages.removeIf(msg -> {
                if (msg.getExpireTime() != null && now > msg.getExpireTime().getTime()) {
                    messageReadStatus.remove(userId + "_" + msg.getMessageId());
                    return true;
                }
                return false;
            });
            
            int afterSize = messages.size();
            cleanedCount += (beforeSize - afterSize);
            
            // 更新本地缓存
            if (beforeSize != afterSize) {
                localCache.put(userId, messages);
            }
        }
        
        if (cleanedCount > 0) {
            System.out.println("[离线缓存] 清理过期消息: count=" + cleanedCount);
        }
        
        return cleanedCount;
    }
    
    /**
     * 同步到Redis（模拟）
     */
    private void syncToRedis(String userId, Message message) {
        // 实际生产环境实现：
        // String key = SystemConstants.REDIS_KEY_OFFLINE_MSG_PREFIX + userId;
        // jedis.lpush(key, JSON.toJSONString(message));
        // jedis.ltrim(key, 0, maxOfflineMessages - 1);
    }
    
    /**
     * 从Redis移除（模拟）
     */
    private void removeFromRedis(String userId, String messageId) {
        // 实际生产环境实现：
        // String key = SystemConstants.REDIS_KEY_OFFLINE_MSG_PREFIX + userId;
        // 需要遍历List找到对应消息并移除
    }
    
    /**
     * 获取统计信息
     */
    public Map<String, Object> getStatistics() {
        Map<String, Object> stats = new HashMap<>();
        stats.put("totalUsers", offlineMessageMap.size());
        stats.put("totalMessages", offlineMessageMap.values().stream()
                .mapToInt(List::size)
                .sum());
        stats.put("cacheSize", localCache.size());
        
        // 统计未读消息数
        int totalUnread = offlineMessageMap.keySet().stream()
                .mapToInt(this::getUnreadCount)
                .sum();
        stats.put("totalUnread", totalUnread);
        
        return stats;
    }
}

八、推送网关：扛住百万连接的第一道关口

推送网关是整个系统的最前线，直接扛所有长连接，它的性能和稳定性直接决定了整个系统的上限。除了前面说的Netty线程模型和参数优化，我们还做了几个关键优化。

8.1 内存池化与泄漏防护

Netty用堆外内存，性能高，但如果管理不好很容易泄漏。我们一开始就踩了这个坑：自定义编解码器里分配了ByteBuf，异常分支里没释放，线上跑了一周就OOM了，dump出来堆外内存占了好几个G。

后来我们做了这几件事，主要是监控与检测：

①开发环境开启Netty泄漏检测：-Dio.netty.leakDetection.level=paranoid，有泄漏立刻就能发现

②所有ByteBuf操作都用try-finally保证释放

③线上加堆外内存监控，超过阈值告警

④用Netty默认的池化ByteBuf，减少内存分配开销

现在系统跑了半年多了，内存一直都比较平稳的还是

8.2 批量发送提升吞吐量

网关收到Kafka的消息，不是来一条就发一条，而是做了批量攒包：攒够100条或者攒10ms，然后批量写到Channel里。这样能大大减少系统调用次数，我们实测单网关吞吐量提升了30%左右。

当然批量也不是越大越好，太大了会增加延迟，10ms是我们测出来兼顾延迟和吞吐量的平衡点。

8.3 单连接限流防攻击

一定要做单连接限流，不然遇到恶意攻击，单个连接疯狂发消息，能把网关资源全占满。我们用Guava的RateLimiter，每个Channel对应一个限流器，每秒最多接收10条消息，超过的直接丢弃。

虽然正常用户不会发这么多，但防人之心不可无，线上什么情况都可能遇到。

8.4 优雅停机减少用户感知

一开始我们更新网关直接kill进程，上面十几万连接瞬间断开，用户体验很差，还容易引发重连雪崩。后来加了优雅停机：

1. 收到停机信号后，先关闭ServerSocket，不再接收新连接

2. 等待现有消息发送完成，最长等待30秒

3. 主动给所有客户端发关闭通知，然后慢慢断开连接

4. 最后再关闭线程组退出

这样重启的时候，用户基本无感知，也不会出现大量连接同时断开的情况。

九、Kafka削峰：突增流量来了也不能崩

推送系统的流量从来不是平稳的：平时每秒几千条，运营一发全量推送，瞬间就到每秒十几万。没有消息队列缓冲，直接打过来系统肯定扛不住。

9.1 Topic拆分做故障隔离

我们踩过一个大亏：一开始所有消息都放一个topic，有一次运营发500万全量推送，消息直接堵了，连核心的订单通知都延迟了十几分钟，用户投诉直接炸了。

后来我们立刻拆分了topic，按优先级隔离：

• push_common_msg：普通业务消息（订单、聊天通知），优先级最高

• push_system_msg：系统通知，优先级中等

• push_broadcast_msg：全量广播、营销消息，优先级最低

就算营销消息堆积如山，也不会影响核心业务消息，故障隔离做得妥妥的。

9.2 关键参数配置

Kafka的参数调优也很重要，说几个我们踩过坑的配置：

• 生产者：acks=1，兼顾性能和可靠性；retries=3发送失败重试；batch.size=16384 + linger.ms=5，批量发送提升吞吐量

• 消费者：enable.auto.commit=false手动提交偏移量，处理完再提交，防止消息丢失；max.poll.records=100批量拉取处理

分区数也不是越多越好，要和消费能力匹配。我们的普通消息topic设了32个分区，对应32个消费线程，峰值每秒能扛10万条写入。

9.3 堆积监控

Kafka堆积是最核心的监控指标之一，一定要设置告警。我们设的阈值是10万条，持续5分钟就告警。很多故障一开始的表现就是消息堆积，早发现就能早处理，不会等用户投诉了才知道。

十、全链路监控

我一直跟团队说，监控要和业务代码同步开发，甚至要先写监控再写业务。没有监控，线上出问题你就是瞎子，等用户反馈过来，早就晚了。

10.1 监控体系

我们搭了三层监控：

1. 基础资源层：服务器CPU、内存、磁盘、网络，用node_exporter采集，这是基础中的基础

2. 业务指标层：推送系统的核心指标，是我们每天重点盯的

3. 链路追踪层：用Sleuth + Zipkin，追踪一条消息从业务方到客户端的全链路，排查问题一查一个准

10.2 核心监控

实时指标采集、分布式追踪和健康检查，主要是监控系统运行状态和性能指标的，代码里面使用Micrometer进行指标采集、支持分布式追踪、提供实时监控和历史数据查询、持告警和通知这些功能

public class MonitorSystemImpl implements MonitorSystem {
    
    // 指标存储（metricName -> List<value>）
    private final Map<String, List<Double>> metricStore = new ConcurrentHashMap<>();
    
    // 计数器存储（metricName -> count）
    private final Map<String, AtomicLong> counterStore = new ConcurrentHashMap<>();
    
    // 耗时存储（metricName -> List<duration>）
    private final Map<String, List<Long>> durationStore = new ConcurrentHashMap<>();
    
    // 追踪信息存储（traceId -> TraceInfo）
    private final Map<String, TraceInfo> traceStore = new ConcurrentHashMap<>();
    
    // 系统启动时间
    private final long startTime;
    
    /**
     * 构造函数
     */
    public MonitorSystemImpl() {
        this.startTime = System.currentTimeMillis();
        initMetrics();
    }
    
    /**
     * 初始化指标
     */
    private void initMetrics() {
        // 初始化计数器
        counterStore.put(SystemConstants.METRIC_MSG_SENT, new AtomicLong(0));
        counterStore.put(SystemConstants.METRIC_MSG_FAILED, new AtomicLong(0));
        counterStore.put(SystemConstants.METRIC_CONNECTIONS, new AtomicLong(0));
        
        // 初始化耗时存储
        durationStore.put(SystemConstants.METRIC_MSG_LATENCY, 
                Collections.synchronizedList(new ArrayList<>()));
    }
    
    @Override
    public void recordMetric(String metricName, double value) {
        if (metricName == null || metricName.isEmpty()) {
            return;
        }
        
        metricStore.computeIfAbsent(metricName, k -> Collections.synchronizedList(new ArrayList<>()))
                .add(value);
        
        // 保持最近1000条记录
        List<Double> values = metricStore.get(metricName);
        if (values.size() > 1000) {
            values.remove(0);
        }
    }
    
    @Override
    public void recordCount(String metricName, long increment) {
        if (metricName == null || metricName.isEmpty()) {
            return;
        }
        
        counterStore.computeIfAbsent(metricName, k -> new AtomicLong(0))
                .addAndGet(increment);
    }
    
    @Override
    public void recordDuration(String metricName, long durationMs) {
        if (metricName == null || metricName.isEmpty()) {
            return;
        }
        
        durationStore.computeIfAbsent(metricName, k -> Collections.synchronizedList(new ArrayList<>()))
                .add(durationMs);
        
        // 保持最近1000条记录
        List<Long> durations = durationStore.get(metricName);
        if (durations.size() > 1000) {
            durations.remove(0);
        }
    }
    
    @Override
    public void startTrace(String traceId, String operation) {
        if (traceId == null || traceId.isEmpty()) {
            return;
        }
        
        TraceInfo traceInfo = new TraceInfo();
        traceInfo.setTraceId(traceId);
        traceInfo.setOperation(operation);
        traceInfo.setStartTime(System.currentTimeMillis());
        traceInfo.setEvents(new ArrayList<>());
        traceInfo.setStatus("running");
        
        traceStore.put(traceId, traceInfo);
        
        System.out.println("[监控系统] 开始追踪: traceId=" + traceId + ", operation=" + operation);
    }
    
    @Override
    public void endTrace(String traceId, boolean success) {
        if (traceId == null || traceId.isEmpty()) {
            return;
        }
        
        TraceInfo traceInfo = traceStore.get(traceId);
        if (traceInfo == null) {
            return;
        }
        
        traceInfo.setEndTime(System.currentTimeMillis());
        traceInfo.setDuration(traceInfo.getEndTime() - traceInfo.getStartTime());
        traceInfo.setStatus(success ? "success" : "failed");
        
        // 记录耗时
        recordDuration("trace." + traceInfo.getOperation(), traceInfo.getDuration());
        
        System.out.println("[监控系统] 结束追踪: traceId=" + traceId + 
                          ", duration=" + traceInfo.getDuration() + "ms" +
                          ", status=" + traceInfo.getStatus());
    }
    
    @Override
    public void addTraceEvent(String traceId, String event) {
        if (traceId == null || traceId.isEmpty()) {
            return;
        }
        
        TraceInfo traceInfo = traceStore.get(traceId);
        if (traceInfo == null) {
            return;
        }
        
        TraceEvent traceEvent = new TraceEvent();
        traceEvent.setEvent(event);
        traceEvent.setTimestamp(System.currentTimeMillis());
        traceEvent.setRelativeTime(traceEvent.getTimestamp() - traceInfo.getStartTime());
        
        traceInfo.getEvents().add(traceEvent);
    }
    
    @Override
    public double getMetricValue(String metricName) {
        // 先检查计数器
        AtomicLong counter = counterStore.get(metricName);
        if (counter != null) {
            return counter.get();
        }
        
        // 再检查指标
        List<Double> values = metricStore.get(metricName);
        if (values != null && !values.isEmpty()) {
            // 返回平均值
            return values.stream().mapToDouble(Double::doubleValue).average().orElse(0);
        }
        
        // 检查耗时
        List<Long> durations = durationStore.get(metricName);
        if (durations != null && !durations.isEmpty()) {
            // 返回平均耗时
            return durations.stream().mapToLong(Long::longValue).average().orElse(0);
        }
        
        return 0;
    }
    
    @Override
    public Map<String, Double> getAllMetrics() {
        Map<String, Double> metrics = new HashMap<>();
        
        // 添加计数器
        for (Map.Entry<String, AtomicLong> entry : counterStore.entrySet()) {
            metrics.put(entry.getKey(), (double) entry.getValue().get());
        }
        
        // 添加指标平均值
        for (Map.Entry<String, List<Double>> entry : metricStore.entrySet()) {
            if (!entry.getValue().isEmpty()) {
                double avg = entry.getValue().stream()
                        .mapToDouble(Double::doubleValue)
                        .average()
                        .orElse(0);
                metrics.put(entry.getKey() + ".avg", avg);
            }
        }
        
        // 添加耗时平均值
        for (Map.Entry<String, List<Long>> entry : durationStore.entrySet()) {
            if (!entry.getValue().isEmpty()) {
                double avg = entry.getValue().stream()
                        .mapToLong(Long::longValue)
                        .average()
                        .orElse(0);
                metrics.put(entry.getKey() + ".avg", avg);
                
                // 计算P95、P99
                List<Long> sorted = new ArrayList<>(entry.getValue());
                Collections.sort(sorted);
                
                int p95Index = (int) (sorted.size() * 0.95);
                int p99Index = (int) (sorted.size() * 0.99);
                
                metrics.put(entry.getKey() + ".p95", (double) sorted.get(p95Index));
                metrics.put(entry.getKey() + ".p99", (double) sorted.get(p99Index));
            }
        }
        
        return metrics;
    }
    
    @Override
    public Map<String, Object> getTraceInfo(String traceId) {
        TraceInfo traceInfo = traceStore.get(traceId);
        if (traceInfo == null) {
            return Collections.emptyMap();
        }
        
        Map<String, Object> info = new HashMap<>();
        info.put("traceId", traceInfo.getTraceId());
        info.put("operation", traceInfo.getOperation());
        info.put("startTime", traceInfo.getStartTime());
        info.put("endTime", traceInfo.getEndTime());
        info.put("duration", traceInfo.getDuration());
        info.put("status", traceInfo.getStatus());
        info.put("events", traceInfo.getEvents());
        
        return info;
    }
    
    @Override
    public Map<String, Object> getHealthStatus() {
        Map<String, Object> health = new HashMap<>();
        
        // 系统运行时间
        long uptime = System.currentTimeMillis() - startTime;
        health.put("uptime", uptime);
        health.put("uptimeFormatted", formatUptime(uptime));
        
        // 消息统计
        long totalSent = counterStore.getOrDefault(SystemConstants.METRIC_MSG_SENT, 
                new AtomicLong(0)).get();
        long totalFailed = counterStore.getOrDefault(SystemConstants.METRIC_MSG_FAILED, 
                new AtomicLong(0)).get();
        
        health.put("totalMessagesSent", totalSent);
        health.put("totalMessagesFailed", totalFailed);
        health.put("successRate", 
                totalSent + totalFailed > 0 ? 
                (double) totalSent / (totalSent + totalFailed) * 100 : 100);
        
        // 连接数
        health.put("activeConnections", 
                counterStore.getOrDefault(SystemConstants.METRIC_CONNECTIONS, 
                        new AtomicLong(0)).get());
        
        // 追踪统计
        health.put("activeTraces", traceStore.size());
        
        // 系统状态
        health.put("status", "healthy");
        health.put("timestamp", System.currentTimeMillis());
        
        return health;
    }
    
    @Override
    public Map<String, Object> getStatisticsReport() {
        Map<String, Object> report = new HashMap<>();
        
        // 基础统计
        report.put("health", getHealthStatus());
        report.put("metrics", getAllMetrics());
        
        // 消息延迟统计
        List<Long> latencies = durationStore.get(SystemConstants.METRIC_MSG_LATENCY);
        if (latencies != null && !latencies.isEmpty()) {
            Map<String, Object> latencyStats = new HashMap<>();
            latencyStats.put("count", latencies.size());
            latencyStats.put("avg", latencies.stream().mapToLong(Long::longValue).average().orElse(0));
            latencyStats.put("min", latencies.stream().mapToLong(Long::longValue).min().orElse(0));
            latencyStats.put("max", latencies.stream().mapToLong(Long::longValue).max().orElse(0));
            
            report.put("latencyStats", latencyStats);
        }
        
        // 追踪统计
        Map<String, Object> traceStats = new HashMap<>();
        traceStats.put("totalTraces", traceStore.size());
        
        long successTraces = traceStore.values().stream()
                .filter(t -> "success".equals(t.getStatus()))
                .count();
        long failedTraces = traceStore.values().stream()
                .filter(t -> "failed".equals(t.getStatus()))
                .count();
        
        traceStats.put("successTraces", successTraces);
        traceStats.put("failedTraces", failedTraces);
        
        report.put("traceStats", traceStats);
        
        return report;
    }
    
    /**
     * 格式化运行时间
     */
    private String formatUptime(long uptime) {
        long seconds = uptime / 1000;
        long minutes = seconds / 60;
        long hours = minutes / 60;
        long days = hours / 24;
        
        return String.format("%d天 %d小时 %d分钟 %d秒", 
                days, hours % 24, minutes % 60, seconds % 60);
    }
    
    /**
     * 清理过期追踪信息
     */
    public void cleanExpiredTraces(long expireMs) {
        long now = System.currentTimeMillis();
        int cleanedCount = 0;
        
        Iterator<Map.Entry<String, TraceInfo>> iterator = traceStore.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, TraceInfo> entry = iterator.next();
            TraceInfo traceInfo = entry.getValue();

        }
        
        if (cleanedCount > 0) {
            System.out.println("[监控系统] 清理过期追踪: count=" + cleanedCount);
        }
    }
    
    /**
     * 追踪信息类
     */
    private static class TraceInfo {
        private String traceId;
        private String operation;
        private long startTime;
        private long endTime;
        private long duration;
        private String status;
        private List<TraceEvent> events;
        
        // Getters and Setters
        public String getTraceId() { return traceId; }
        public void setTraceId(String traceId) { this.traceId = traceId; }
        public String getOperation() { return operation; }
        public void setOperation(String operation) { this.operation = operation; }
        public long getStartTime() { return startTime; }
        public void setStartTime(long startTime) { this.startTime = startTime; }
        public long getEndTime() { return endTime; }
        public void setEndTime(long endTime) { this.endTime = endTime; }
        public long getDuration() { return duration; }
        public void setDuration(long duration) { this.duration = duration; }
        public String getStatus() { return status; }
        public void setStatus(String status) { this.status = status; }
        public List<TraceEvent> getEvents() { return events; }
        public void setEvents(List<TraceEvent> events) { this.events = events; }
    }
    
    /**
     * 追踪事件类
     */
    private static class TraceEvent {
        private String event;
        private long timestamp;
        private long relativeTime;
        
        // Getters and Setters
        public String getEvent() { return event; }
        public void setEvent(String event) { this.event = event; }
        public long getTimestamp() { return timestamp; }
        public void setTimestamp(long timestamp) { this.timestamp = timestamp; }
        public long getRelativeTime() { return relativeTime; }
        public void setRelativeTime(long relativeTime) { this.relativeTime = relativeTime; }
    }
}