数据采集层设计：WebSocket连接管理、心跳机制、断线重连策略

行情数据采集，是整个系统的入口。说白了，如果这层没做好，后面再牛的算法、再快的计算都是白搭。我见过太多团队，花大把精力优化计算引擎，结果采集层一崩，全盘皆输。

今天咱们就聊聊数据采集层的三个核心问题：连接怎么管、心跳怎么保、断了怎么重连。嗯，这三个点，任何一个出问题，行情数据就会断流。

WebSocket连接管理：别让连接变成一团乱麻

WebSocket 连接管理，听起来简单，做起来坑不少。我个人习惯把连接管理拆成三个维度：连接池、连接状态、连接生命周期。

连接池设计

行情源通常不止一个。比如你可能同时接入沪深交易所、期货交易所、甚至加密货币的行情。每个交易所可能还有多个订阅通道。这时候，连接池就派上用场了。

我建议用 ConcurrentHashMap 来管理连接，key 是交易所+通道标识，value 是连接对象。这样查找快，也方便做隔离。

// 连接池示例
public class WebSocketConnectionPool {
    private final ConcurrentHashMap<String, WebSocketClient> connections = new ConcurrentHashMap<>();
    
    public void addConnection(String channelId, WebSocketClient client) {
        connections.put(channelId, client);
    }
    
    public WebSocketClient getConnection(String channelId) {
        return connections.get(channelId);
    }
    
    public void removeConnection(String channelId) {
        WebSocketClient client = connections.remove(channelId);
        if (client != null) {
            client.close();
        }
    }
}

连接状态机

每个连接都有生命周期。我习惯用状态机来管理：INIT → CONNECTING → CONNECTED → RECONNECTING → CLOSED。

为什么会这样设计？因为连接不是非黑即白的。比如网络抖动时，连接可能处于半断开状态。这时候状态机就能帮你准确判断该做什么。

public enum ConnectionState {
    INIT,       // 初始状态
    CONNECTING, // 正在连接
    CONNECTED,  // 已连接
    RECONNECTING, // 正在重连
    CLOSED      // 已关闭
}

心跳机制：别让连接静默死亡

WebSocket 连接建立后，如果长时间没有数据交互，网络设备（比如防火墙、负载均衡器）可能会认为连接已失效，直接给你掐断。这就是所谓的「静默断开」。

我在项目中遇到过这种情况：某次上线后，行情数据每 30 分钟准时断一次。查了半天，发现是防火墙的会话超时设置是 30 分钟。嗯，从那以后，心跳机制就成了我的标配。

心跳策略选择

策略	说明	适用场景
固定间隔心跳	每隔 N 秒发送一次心跳包	大多数行情源
自适应心跳	根据网络状况动态调整间隔	网络不稳定的环境
双向心跳	客户端和服务端都发送心跳	高可靠性要求

我个人推荐固定间隔心跳 + 双向心跳的组合。固定间隔简单可靠，双向心跳能及时发现单方面断开的情况。

// 心跳检测器
public class HeartbeatChecker {
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(4);
    private final long heartbeatInterval = 5000; // 5秒一次
    
    public void startHeartbeat(WebSocketClient client) {
        scheduler.scheduleAtFixedRate(() -> {
            if (client.isOpen()) {
                client.sendPing();
            }
        }, 0, heartbeatInterval, TimeUnit.MILLISECONDS);
    }
}

断线重连策略：从失败中优雅恢复

断线重连，是数据采集层的最后一道防线。网络不可能永远稳定，所以重连策略必须设计好。

重连策略的核心参数

• 最大重连次数：防止无限重连浪费资源
• 重连间隔：从短到长，避免频繁重连
• 重连超时：单次重连的最大等待时间
• 退避策略：指数退避还是固定间隔

我建议用指数退避 + 随机抖动。指数退避能避免同时重连造成雪崩，随机抖动能防止多个客户端同时重连。

// 指数退避重连策略
public class ExponentialBackoffReconnect {
    private final int maxRetries = 5;
    private final long baseDelay = 1000; // 初始1秒
    private int retryCount = 0;
    
    public long getNextDelay() {
        if (retryCount >= maxRetries) {
            return -1; // 停止重连
        }
        long delay = (long) (baseDelay * Math.pow(2, retryCount));
        // 加入随机抖动，避免同时重连
        delay += (long) (Math.random() * 1000);
        retryCount++;
        return delay;
    }
}

重连时的数据恢复

重连成功后，别忘了恢复订阅。我遇到过这样的情况：重连成功了，但忘了重新订阅，结果行情数据一直为空。排查了半天才发现是订阅没恢复。

我的做法是：在连接对象里维护一个订阅列表。重连成功后，自动遍历这个列表，重新发起订阅请求。

public class ReconnectHandler {
    private final Set<String> subscriptions = new CopyOnWriteArraySet<>();
    
    public void onReconnectSuccess(WebSocketClient client) {
        // 重新订阅所有频道
        for (String channel : subscriptions) {
            client.send("SUBSCRIBE " + channel);
        }
    }
}

整体架构图

下面这张图，展示了数据采集层的整体设计。你可以看到连接管理、心跳机制、断线重连是如何协同工作的。

避坑指南

最后，分享几个我踩过的坑，希望能帮你少走弯路。

• 不要在主线程做重连：重连操作会阻塞主线程，导致其他连接也受影响。一定要用独立的线程池。
• 注意内存泄漏：每次重连都会创建新的连接对象，如果旧的连接没有正确关闭，内存会慢慢涨上去。
• 日志要打全：连接断开、重连成功、重连失败，这些关键事件一定要打日志。不然出了问题，你连从哪开始查都不知道。
• 测试要模拟真实场景：我曾经在测试环境一切正常，上线后才发现断网重连有问题。因为测试环境网络太好了，根本不会断。后来我专门写了个网络模拟器，随机断开连接来测试。

数据采集层，说白了就是行情系统的「地基」。地基不稳，上面盖再高的楼也是白搭。连接管理、心跳机制、断线重连，这三块做好了，你的行情系统就成功了一半。