📚 课程目录第3章实时数据采集

3. 实时数据采集：基于 Apache Flink 的 API 数据源连接器开发

好，咱们进入第三章。这一章要解决一个很实际的问题——怎么让 Flink 实时地从 API 里拿数据。

你可能会想：「Flink 不是有现成的 Kafka 连接器、JDBC 连接器吗？直接用不就行了？」嗯，但现实是，很多业务系统的数据只通过 HTTP API 暴露出来。比如内部的管理后台、第三方天气服务、或者某个老旧的 CRM 系统。这时候，你就得自己动手写一个 API 数据源连接器。

我最早遇到这个场景是在做一个实时风控项目。业务方要求把某个第三方黑名单 API 的数据实时拉进 Flink 做关联。当时我翻遍了 Flink 官方文档，发现没有现成的 HTTP 连接器。怎么办？自己写呗。这一写，就踩了不少坑。今天我把这些经验整理出来，希望能帮你少走弯路。

3.1 两种接入模式：轮询 vs WebSocket

API 数据接入，说白了就两种路子：

轮询（Polling）：定时去请求 HTTP 接口，拿到数据后交给 Flink 处理。
WebSocket：跟服务端建立长连接，服务端主动推送数据过来。

这两种模式各有各的适用场景。我画了张图，你看一眼就明白了。

3.2 自定义 SourceFunction：轮询实现

先看轮询。Flink 提供了 SourceFunction 接口，我们实现它就能自定义数据源。核心逻辑就三步：

在 run() 方法里写一个循环。
每次循环调用 HTTP 客户端请求 API。
拿到数据后通过 SourceContext.collect() 发射出去。
循环结束后调用 cancel() 清理资源。

下面是一个完整的轮询实现。我习惯用 Apache HttpClient，因为它稳定、线程安全。

import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.json.JSONObject;

public class PollingApiSource implements SourceFunction<String> {

    private volatile boolean running = true;
    private final String apiUrl;
    private final long intervalMs;

    public PollingApiSource(String apiUrl, long intervalMs) {
        this.apiUrl = apiUrl;
        this.intervalMs = intervalMs;
    }

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        CloseableHttpClient httpClient = HttpClients.createDefault();

        while (running) {
            try {
                HttpGet request = new HttpGet(apiUrl);
                try (CloseableHttpResponse response = httpClient.execute(request)) {
                    String result = EntityUtils.toString(response.getEntity());
                    // 这里可以做数据清洗、格式转换
                    ctx.collect(result);
                }
            } catch (Exception e) {
                // 记录异常，但不要中断循环
                System.err.println("API 请求失败: " + e.getMessage());
            }

            // 控制轮询间隔
            Thread.sleep(intervalMs);
        }

        httpClient.close();
    }

    @Override
    public void cancel() {
        running = false;
    }
}

我的经验：轮询间隔别设太短。我曾经设过 100 毫秒，结果把对方的 API 打挂了。一般建议 1 秒以上，如果对实时性要求高，可以考虑 WebSocket。

3.3 自定义 SourceFunction：WebSocket 接入

WebSocket 的实现稍微复杂一点。你需要一个 WebSocket 客户端库，比如 Java-WebSocket。核心逻辑是：

在 run() 里建立 WebSocket 连接。
在 onMessage() 回调里调用 ctx.collect()。
在 cancel() 里关闭连接。

这里有个坑：onMessage() 是在 WebSocket 客户端的线程里执行的，而 ctx.collect() 必须在 Flink 的 checkpoint 锁保护下调用。否则会出现数据乱序甚至丢失。

正确的做法是用 SourceContext.getCheckpointLock() 做同步。

import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import java.net.URI;

public class WebSocketSource implements SourceFunction<String> {

    private volatile boolean running = true;
    private final String wsUrl;
    private transient WebSocketClient wsClient;

    public WebSocketSource(String wsUrl) {
        this.wsUrl = wsUrl;
    }

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        wsClient = new WebSocketClient(new URI(wsUrl)) {
            @Override
            public void onOpen(ServerHandshake handshakedata) {
                System.out.println("WebSocket 连接已建立");
            }

            @Override
            public void onMessage(String message) {
                synchronized (ctx.getCheckpointLock()) {
                    ctx.collect(message);
                }
            }

            @Override
            public void onClose(int code, String reason, boolean remote) {
                System.out.println("WebSocket 连接关闭: " + reason);
            }

            @Override
            public void onError(Exception ex) {
                System.err.println("WebSocket 错误: " + ex.getMessage());
            }
        };

        wsClient.connect();

        // 等待连接建立
        while (!wsClient.isOpen() && running) {
            Thread.sleep(100);
        }

        // 保持线程存活
        while (running) {
            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {
        running = false;
        if (wsClient != null) {
            wsClient.close();
        }
    }
}

注意：WebSocket 连接可能会因为网络波动断开。你需要在 onClose() 和 onError() 里实现重连逻辑。我一般会加一个指数退避的重试策略，避免频繁重连把服务端打爆。

3.4 两种方案的对比与选型

我把两种方案的核心差异整理成了表格，方便你选型时参考。

对比维度	轮询模式	WebSocket 模式
实时性	取决于轮询间隔，通常秒级	毫秒级，服务端主动推送
资源消耗	即使无数据也在请求，浪费带宽	有数据才传输，节省资源
实现复杂度	低，几行代码搞定	中，需要处理连接、重连、心跳
服务端要求	只需普通 HTTP 接口	必须支持 WebSocket 协议
数据完整性	可能漏掉两次请求之间的数据	只要连接不断，数据不丢
适用场景	低频数据、内部系统、快速原型	高频实时数据、金融行情、IoT

3.5 避坑指南与最佳实践

这部分是我用真金白银换来的教训，你仔细看看。

避坑 1：别忘了 checkpoint 锁

在 WebSocket 的 onMessage() 里直接调用 ctx.collect() 是不安全的。Flink 的 checkpoint 机制需要保证数据的一致性，而 onMessage() 是在 WebSocket 库的线程里执行的。不加锁的话，checkpoint 时可能把未处理完的数据也保存了，恢复时就会重复或丢失。

避坑 2：轮询要处理空响应

我曾经遇到过一个 API，在没有新数据时返回空 JSON {}。如果不做过滤，Flink 下游会收到大量无意义的空数据，白白浪费计算资源。建议在 run() 里加一个判断：如果响应体为空或没有有效字段，直接跳过，不调用 collect()。

避坑 3：连接断开要自动重连

WebSocket 连接不会永远稳定。网络抖动、服务端重启都可能导致断开。我建议在 onClose() 里启动一个重连线程，用指数退避策略：第一次等 1 秒，第二次等 2 秒，第三次等 4 秒……最多等 60 秒。这样既不会把服务端打爆，也能在网络恢复后尽快重新接入。

3.6 在 Flink 作业中使用自定义 Source

写好了 Source，怎么用？很简单，在 StreamExecutionEnvironment 里加上就行。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 轮询模式：每 5 秒请求一次
DataStream<String> pollingStream = env
    .addSource(new PollingApiSource("https://api.example.com/data", 5000))
    .name("Polling API Source");

// WebSocket 模式：实时接入
DataStream<String> wsStream = env
    .addSource(new WebSocketSource("wss://api.example.com/stream"))
    .name("WebSocket Source");

// 后续处理
wsStream
    .map(new JsonToPojoMapper())
    .keyBy(event -> event.getUserId())
    .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
    .process(new MyWindowFunction())
    .print();

env.execute("API Data Ingestion Job");

你看，代码量不大。但这里有个细节：Source 的并行度。默认情况下，addSource() 的并行度是 1。如果你的 API 支持并发请求，可以手动设置并行度，比如 .setParallelism(4)。不过要注意，WebSocket 连接通常是一对一的，设高了反而浪费资源。