Apache Flink无同步原语的SinkFunction是否有风险及线程安全优化方案

一、风险判定与理解正确性 #

你的理解完全正确，该实现确实存在线程安全风险：
默认场景下Flink单个Sink子任务的invoke方法由框架单线程调用，不存在并发问题，但如果你的实现中引入了自定义多线程逻辑（比如单独的定时线程执行批量flush、异步写操作），多线程会同时操作非线程安全的bufferedRecords列表，会触发以下典型竞态问题：

• 多线程同时调用add可能导致列表内部结构损坏、元素丢失
• size判断、writeRecords、empty三个操作不具备原子性，可能出现多个线程同时判断达到批量阈值，重复写入同一份数据；也可能出现清空操作执行时刚好有新元素插入，导致未写入的元素被直接清空丢失。

二、更优的线程安全方案 #

直接给整个invoke方法加锁可以解决问题，但锁粒度太大，高吞吐场景下会严重影响性能，可根据是否需要数据容错选择更优方案：

方案1：无容错需求场景用并发阻塞队列 #

使用JDK自带的LinkedBlockingQueue替代普通List，利用其内置的原子操作方法避免手动加锁，读写锁分离的设计也能保留更高的并发度：

private LinkedBlockingQueue<Record> bufferedRecords;
@Override
public void invoke(Point point, Context context) throws Exception {
    bufferedRecords.put(point);
    // 达到阈值触发批量写
    if (bufferedRecords.size() >= batchSize) {
        List<Record> batch = new ArrayList<>(batchSize);
        // 原子拉取最多batchSize个元素
        bufferedRecords.drainTo(batch, batchSize);
        writeRecords(batch);
    }
}
// 定时flush逻辑也可直接调用drainTo，无需额外加锁
private void scheduledFlush() {
    List<Record> batch = new ArrayList<>();
    bufferedRecords.drainTo(batch);
    if (!batch.isEmpty()) {
        writeRecords(batch);
    }
}

方案2：需要容错场景用Flink状态存储缓冲区 #

如果要求故障恢复时不丢失缓冲区数据，更推荐使用Flink原生的ListState存储缓冲数据，框架本身会保证状态操作的线程安全，同时缓冲数据会被持久化到Checkpoint中，避免进程崩溃导致的缓冲数据丢失：

private transient ListState<Record> bufferedState;
@Override
public void open(Configuration parameters) throws Exception {
    ListStateDescriptor<Record> descriptor = new ListStateDescriptor<>(
        "buffered-records",
        TypeInformation.of(Record.class)
    );
    bufferedState = getRuntimeContext().getListState(descriptor);
}

@Override
public void invoke(Point point, Context context) throws Exception {
    bufferedState.add(point);
    int size = Iterables.size(bufferedState.get());
    if (size >= batchSize) {
        List<Record> batch = Lists.newArrayList(bufferedState.get());
        writeRecords(batch);
        bufferedState.clear();
    }
}

// 配合Checkpoint触发flush可实现端到端恰好一次语义
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
    if (!Iterables.isEmpty(bufferedState.get())) {
        List<Record> batch = Lists.newArrayList(bufferedState.get());
        writeRecords(batch);
        bufferedState.clear();
    }
}

内容的提问来源于stack exchange，提问作者Neo