4、数据采集层架构：采集器设计模式（生产者-消费者）、多线程与异步IO选型、采集任务调度与监控

好，咱们进入数据采集层。这一层说白了，就是行情系统的「水源地」。水源要是断了，下游再牛的分析引擎也是白搭。我在早期做量化系统时，就吃过这个亏——数据源一抖动，整个策略回测全废。所以今天咱们把采集层的几个核心问题掰开揉碎讲清楚。

4.1 采集器设计模式：生产者-消费者

先聊聊设计模式。行情采集最经典的模型，就是生产者-消费者。为什么？因为行情数据是典型的「快生产、慢消费」场景。

生产者是谁？是各个交易所的行情推送。比如上交所的L1快照，每秒可能来几百笔。消费者是谁？是下游的数据清洗、存储、分发模块。它们处理速度往往跟不上生产速度。

我习惯用阻塞队列来做缓冲。生产者只管往队列里塞数据，消费者从队列里取。这样两边解耦，互不干扰。来看一个简化版的核心代码：

public class MarketDataCollector {
    private BlockingQueue<Tick> queue = new LinkedBlockingQueue<>(10000);
    
    // 生产者：接收交易所行情
    public void onTick(Tick tick) {
        if (!queue.offer(tick)) {
            // 队列满了，说明消费太慢
            logger.warn("队列积压，丢弃tick: {}", tick);
            metrics.recordDrop();
        }
    }
    
    // 消费者：处理行情数据
    public void processLoop() {
        while (running) {
            Tick tick = queue.poll(100, TimeUnit.MILLISECONDS);
            if (tick != null) {
                handleTick(tick);
            }
        }
    }
}

这里有个坑。队列大小设多少？设太小容易丢数据，设太大又占内存。我建议根据行情峰值速率来算。比如每秒1000笔，消费者处理速度500笔/秒，那队列至少得能缓冲2秒的数据，也就是2000个tick。再留点余量，设成5000比较稳妥。

4.2 多线程与异步IO选型

好，设计模式定下来了。接下来是技术选型。多线程还是异步IO？这问题我面试时经常问，很多人答不上来。

我的经验是：IO密集用异步，计算密集用多线程。行情采集属于典型的IO密集型——大部分时间在等网络数据。所以异步IO是更优解。

但实际项目中，我往往两者混用。为什么？因为行情数据到了之后，还需要做一些轻量级处理，比如校验、解码。这部分是CPU计算。所以架构上我习惯这样分层：

• 网络层：用Netty或Java NIO做异步IO，负责收发数据
• 业务层：用固定线程池做数据解码、校验、分发
• 存储层：用独立线程写磁盘或发消息队列

来看一个Netty的接入示例：

EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup(4);

try {
    ServerBootstrap b = new ServerBootstrap();
    b.group(bossGroup, workerGroup)
     .channel(NioServerSocketChannel.class)
     .childHandler(new ChannelInitializer<SocketChannel>() {
         @Override
         public void initChannel(SocketChannel ch) {
             ch.pipeline().addLast(new MarketDataDecoder());
             ch.pipeline().addLast(new MarketDataHandler());
         }
     });
    
    ChannelFuture f = b.bind(port).sync();
    f.channel().closeFuture().sync();
} finally {
    bossGroup.shutdownGracefully();
    workerGroup.shutdownGracefully();
}

这里worker线程数设4个，是因为行情数据通常来自少数几个交易所连接。每个连接一个线程就够了。设太多反而增加上下文切换开销。

4.3 采集任务调度与监控

最后一块，也是容易被忽视的——任务调度与监控。行情采集不是启动就完事了，它需要持续稳定运行。我总结了三层监控体系：

监控层级	监控指标	告警阈值
连接层	TCP连接状态、心跳延迟	心跳超时3秒告警
数据层	每秒tick数、队列积压数、数据延迟	队列积压>80%告警
系统层	CPU、内存、网络IO	CPU>80%告警

调度方面，我习惯用Quartz或自研的定时任务框架。主要做三件事：

• 定时重连：交易所连接断了，自动重试。重试间隔指数退避，别一断就连，会把交易所打挂
• 数据补采：发现某段时间数据缺失，自动触发补采任务
• 健康检查：每隔30秒检查一次所有采集器状态，异常自动重启

下面这张图是我常用的采集器架构，你可以参考：

嗯，这里要注意。监控不是摆设，得有自动响应机制。比如队列积压超过80%，自动扩容消费者线程数。连接断了，自动触发重连。我见过不少系统，监控面板花里胡哨，但告警了没人处理，那还不如不监控。

好了，数据采集层就聊到这儿。下一节咱们进入数据清洗与标准化，看看那些「脏数据」是怎么被收拾干净的。