如何确保Apache Flink流处理的数据一致性和可靠性

import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.checkpoint.Checkpointed;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
public class FlinkConsistencyExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 配置检查点
        env.enableCheckpointing(10000); // 每10秒进行一次检查点
        // 添加状态的source函数
        env.addSource(new SourceFunctionWithState()).setParallelism(1);
        // 启动执行
        env.execute("Flink Consistency and Reliability Example");
    }
    public static class SourceFunctionWithState
            extends RichParallelSourceFunction<String>
            implements Checkpointed<Long> {
        private final Object lock = new Object();
        private long state = 0;
        @Override
        public void run(SourceContext<String> ctx) throws Exception {
            while (true) {
                synchronized (lock) {
                    // 业务逻辑处理
                    state++;
                }
                // 发出数据
                ctx.collect("Event " + state);
                Thread.sleep(1000); // 模拟处理时间
            }
        }
        @Override
        public void cancel() {}
        @Override
        public Long getState() {
            synchronized (lock) {
                return state;
            }
        }
        @Override
        public void restore(Long state) {
            synchronized (lock) {
                this.state = state;
            }
        }
    }
}