4. 流编排核心组件：数据源（Source）、处理节点（Processor）、数据汇（Sink）的职责与生命周期

流编排应用的本质是将数据采集、处理与输出拆解为三个可独立部署、可组合的组件。理解 Source、Processor、Sink 的职责边界与生命周期管理，是构建可靠边缘计算管线的第一步。本章从职责定义、状态转换、生命周期回调三个维度展开，并给出典型边缘场景下的组件协作示例。

4.1 三大组件的职责定义

在流编排运行时，每个组件都是一个独立的执行单元，通过连接器（Connector）与上下游通信。下表从输入、输出、核心行为三个维度对比三者的职责差异：

4.2 组件生命周期状态机

每个组件在流编排引擎内部都经历统一的状态转换。状态机确保组件在启动、运行、停止、故障恢复等阶段行为可预测。下图展示了完整的生命周期状态流转：

4.3 生命周期回调方法

每个组件必须实现以下四个核心回调方法，编排引擎在状态转换时自动调用：

// Source 示例：HTTP API 轮询数据源
public class HttpPollingSource implements Source {

    private HttpClient client;
    private String endpoint;

    @Override
    public void init(SourceContext ctx) {
        // 1. 解析配置（URL、间隔、认证）
        this.endpoint = ctx.getConfig().getString("endpoint");
        this.client = HttpClient.newBuilder()
            .connectTimeout(Duration.ofSeconds(5))
            .build();
        // 2. 预校验：测试连接是否可达
        if (!healthCheck()) {
            throw new SourceInitException("Endpoint unreachable: " + endpoint);
        }
    }

    @Override
    public void start(SourceCollector<Record> collector) {
        // 进入事件循环：定时拉取 HTTP API
        while (isRunning()) {
            HttpRequest req = HttpRequest.newBuilder()
                .uri(URI.create(endpoint))
                .GET()
                .build();
            client.sendAsync(req, BodyHandlers.ofString())
                .thenAccept(response -> {
                    List<Record> records = parseResponse(response.body());
                    records.forEach(collector::collect);
                });
            Thread.sleep(pollIntervalMs);
        }
    }

    @Override
    public void pause() {
        // 暂停拉取，保留连接池
        setPaused(true);
    }

    @Override
    public void close() {
        // 释放连接、关闭线程池
        client.close();
    }
}

4.4 组件协作流程：一次完整的流处理

下图展示了一个边缘 HTTP API 采集场景下，三个组件如何协作完成一次数据流转：

4.5 组件生命周期管理最佳实践

• 资源隔离： 每个 Source 独占一个连接池，避免一个组件故障拖垮整个引擎。
• 优雅关闭： 在 close() 中等待正在处理的记录完成（drain），设置超时防止永久阻塞。
• 状态持久化： Processor 的中间状态（如窗口聚合结果）应在 pause() 时快照到本地存储，resume() 时恢复。
• 故障恢复： 组件从 FAILED 状态只能通过 close() 进入 CLOSED，然后重新 init()；不允许直接从 FAILED 跳回 RUNNING。

4.6 小结