java

7.3 连接器：自定义 Source 与 Sink

Flink 自定义 Source 与 Sink 的实现：SourceFunction 的 run/cancel 与 checkpoint 锁、RichSourceFunction 的 offset 状态、RichSinkFunction 的 open/invoke/close 生命周期、批量写入的数据丢失窗口、幂等与两阶段提交的取舍，以及这些接口在 1.20 的废弃状态。

Leeting Yan 2026-11-01 4 分钟阅读 1855 字

自定义 Source

先说结论：SourceFunction 这一族（含 RichSourceFunction、ParallelSourceFunction）在 Flink 1.20 已废弃，新代码应实现 Source 接口并经 env.fromSource(...) 接入。但存量项目里它们仍然大量存在，而且 getCheckpointLock() 这类概念是理解 Flink Source 的基础，所以本节仍按这一族讲，每处标出替代方式。

SourceFunction（非并行）

public class EventGenerator implements SourceFunction<Event> {
    private volatile boolean isRunning = true;

    @Override
    public void run(SourceContext<Event> ctx) throws Exception {
        while (isRunning) {
            Event event = new Event(
                UUID.randomUUID().toString(),
                "ORDER",
                ThreadLocalRandom.current().nextDouble(100, 10000),
                System.currentTimeMillis()
            );

            synchronized (ctx.getCheckpointLock()) {
                ctx.collect(event);
            }

            Thread.sleep(100);
        }
    }

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

getCheckpointLock() 确保 collect 和 checkpoint 的原子性。

并行 Source

getRuntimeContext() 只存在于 Rich* 变体里。只实现 ParallelSourceFunction 拿不到子任务编号，必须用 RichParallelSourceFunction：

public class ParallelEventGenerator extends RichParallelSourceFunction<Event> {
    private volatile boolean isRunning = true;

    @Override
    public void run(SourceContext<Event> ctx) throws Exception {
        int subtask = getRuntimeContext().getIndexOfThisSubtask();
        int parallelism = getRuntimeContext().getNumberOfParallelSubtasks();

        // 每个实例生成不同范围的数据
        while (isRunning) {
            Event event = generateEventForSubtask(subtask, parallelism);
            synchronized (ctx.getCheckpointLock()) {
                ctx.collectWithTimestamp(event, event.getEventTime());
                // 多个子任务各自发射 Watermark，下游取最小值；
                // 某个子任务空闲不发，会拖住整条链路（用 withIdleness 兜底）
                ctx.emitWatermark(new Watermark(event.getEventTime() - 1000));
            }
        }
    }

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

RichSourceFunction（带生命周期）

public class RichEventSource extends RichSourceFunction<Event>
        implements CheckpointedFunction {
    private ListState<Long> offsetState;
    private volatile boolean isRunning = true;
    private long currentOffset = 0;

    // 1.19 起用 OpenContext，open(Configuration) 已废弃
    @Override
    public void open(OpenContext openContext) throws Exception {
        // 初始化外部连接
    }

    @Override
    public void run(SourceContext<Event> ctx) throws Exception {
        while (isRunning) {
            Event event = readFromExternal(currentOffset++);
            synchronized (ctx.getCheckpointLock()) {
                ctx.collect(event);
            }
        }
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        offsetState.clear();
        offsetState.add(currentOffset);
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        ListStateDescriptor<Long> descriptor = new ListStateDescriptor<>("offset", Types.LONG);
        offsetState = context.getOperatorStateStore().getUnionListState(descriptor);
        if (context.isRestored()) {
            for (Long offset : offsetState.get()) {
                currentOffset = Math.max(currentOffset, offset);
            }
        }
    }

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

    @Override
    public void close() throws Exception {
        // 关闭连接
    }
}

cancel() 是 SourceFunction 的抽象方法，漏写就编译不过；它由 Flink 在作业取消时从另一个线程调用，所以 isRunning 必须是 volatile（或用 AtomicBoolean），否则 run() 里的循环可能永远读不到新值。

自定义 Sink

RichSinkFunction

public class BatchJdbcSink extends RichSinkFunction<Metric> {
    private Connection connection;
    private PreparedStatement pstmt;
    private int batchSize = 0;

    @Override
    public void open(OpenContext openContext) throws Exception {
        connection = DriverManager.getConnection(
            "jdbc:postgresql://db:5432/eventd", "eventd", "eventd");
        connection.setAutoCommit(false);
        pstmt = connection.prepareStatement(
            "INSERT INTO metrics (window_start, symbol, count, total) VALUES (?, ?, ?, ?)" +
            "ON CONFLICT (window_start, symbol) DO UPDATE SET count = EXCLUDED.count, total = EXCLUDED.total");
    }

    @Override
    public void invoke(Metric metric, Context context) throws Exception {
        pstmt.setTimestamp(1, new Timestamp(metric.windowStart()));
        pstmt.setString(2, metric.symbol());
        pstmt.setLong(3, metric.count());
        pstmt.setDouble(4, metric.total());
        pstmt.addBatch();

        if (++batchSize >= 100) {
            pstmt.executeBatch();
            connection.commit();
            batchSize = 0;
        }
    }

    @Override
    public void close() throws Exception {
        if (batchSize > 0) {
            pstmt.executeBatch();
            connection.commit();
        }
        if (pstmt != null) pstmt.close();
        if (connection != null) connection.close();
    }
}

这个 Sink 的语义是至少一次，而且有丢数据窗口：只在攒够 100 条或 close() 时提交。作业异常退出时 close() 不保证被调用，最近一批（最多 99 条）就丢了。要补齐这一点，得让提交与 Checkpoint 绑定（实现 CheckpointedFunction，在 snapshotState 里 flush），或者干脆实现下面的事务型 Sink。

幂等性保障

自定义 Sink 的幂等策略：

下游	幂等手段
数据库	UPSERT（INSERT ON CONFLICT UPDATE）
Redis	SETEX 覆盖
Kafka	幂等生产者（enable.idempotence）
文件	按分区写入不重叠路径

TwoPhaseCommitSinkFunction（Exactly-Once）

需要 Exactly-Once 的 Sink 实现两阶段提交。TwoPhaseCommitSinkFunction 在 Flink 1.20 已标记 @Deprecated，新连接器应基于 Sink V2 的 TwoPhaseCommittingSink 实现；这里保留它是因为存量代码和教程仍然大量使用，而且它的协议描述最直观。

public class TwoPhaseJdbcSink
        extends TwoPhaseCommitSinkFunction<Metric, String, Void> {
    private transient DataSource dataSource;
    // 事务 ID → 未提交的连接。连接对象只留在内存，不进 Checkpoint
    private transient Map<String, Connection> openTransactions;

    public TwoPhaseJdbcSink() {
        // 两个参数分别是 TXN 和 CONTEXT 的序列化器。
        // 第一个参数以前常被写成 TypeInformation.of(Metric.class)...，那是 IN 的序列化器，
        // 类型对不上；TXN 必须是可序列化的事务句柄，这里用 String 事务 ID
        super(new StringSerializer(), VoidSerializer.INSTANCE);
    }

    @Override
    public void open(OpenContext openContext) throws Exception {
        super.open(openContext);
        openTransactions = new HashMap<>();
        // dataSource = 初始化连接池
    }

    @Override
    protected String beginTransaction() throws Exception {
        String txId = UUID.randomUUID().toString();
        Connection conn = dataSource.getConnection();
        conn.setAutoCommit(false);
        openTransactions.put(txId, conn);
        return txId;
    }

    @Override
    protected void invoke(String transaction, Metric value, Context context)
            throws Exception {
        Connection conn = openTransactions.get(transaction);
        // INSERT ... ON CONFLICT (...) DO UPDATE ...
    }

    @Override
    protected void preCommit(String transaction) throws Exception {
        // JDBC 没有 prepare 阶段：在 preCommit 里直接 commit，
        // 代价是提交之后本次 Checkpoint 若失败也无法回滚（见下方说明）
        openTransactions.get(transaction).commit();
    }

    @Override
    protected void commit(String transaction) {
        // 数据已在 preCommit 提交，这里只释放连接
        closeQuietly(openTransactions.remove(transaction));
    }

    @Override
    protected void abort(String transaction) {
        Connection conn = openTransactions.remove(transaction);
        try {
            if (conn != null) conn.rollback();
        } catch (SQLException ignored) {
        } finally {
            closeQuietly(conn);
        }
    }
}

这个例子有两个容易写错的地方：

TXN 不能是 Connection。TwoPhaseCommitSinkFunction 会在 snapshotState() 时把未完成的事务写入状态，TXN 必须能被序列化；JDBC 连接是活对象，既不可序列化，恢复后也用不了。所以 TXN 用事务 ID，真正的连接放在算子内存里，靠 ID 关联。
JDBC 没有两阶段提交的准备阶段。上游 Kafka Sink 可以把 preCommit 做成 flush、把真正的提交放到 commit；JDBC 只能在 preCommit 里 commit()，此时数据已对其他会话可见。如果之后 Checkpoint 失败，Flink 会重放，而这些已经提交的行不会回滚——所以下游表还得靠唯一键 UPSERT 兜底，才能得到实际的 Exactly-Once 效果。

小结

接口	场景	方法
`SourceFunction`	非并行 Source	`run()`, `cancel()`
`ParallelSourceFunction`	并行 Source	`run()`, `cancel()`
`RichSourceFunction`	带生命周期的 Source	`open()`, `run()`, `close()`
`RichSinkFunction`	带生命周期的 Sink	`open()`, `invoke()`, `close()`
`TwoPhaseCommitSinkFunction`	EOS Sink（1.20 起废弃）	`beginTransaction()`, `preCommit()`, `commit()`, `abort()`
`CheckpointedFunction`	状态保存	`snapshotState()`, `initializeState()`

第 7 章结束。下一节 8.1 指标与延迟分析。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页