Flink 中有两个 Exactly-Once 语义实现，一个是 Kafka，另一个是 StreamingFileSink。

 

参考他们实现的逻辑，来自定义实现端到端exactly-once语义。

 

分析：

 

Flink的checkpoint机制（通过Chandy-Lamport）：

 

JobManager的CheckpointCoordinator通过在stream中添加barrier，当barrier前的数据的所有operator的checkpoint都操作完成并返回CheckpointCoordinator，才代表此次checkpoint执行完成；

 

checkpoint机制可以保证不丢数据，因为每次恢复的时候都是从最后一次checkpoint成功的地方开始处理，这样可能会重复处理某些数据，实现了at-least-once，没法做到exactly-once语义；

 

flink提供了TwoPhaseCommit两阶段提交机制：pre-commit预提交和commit正式提交，其中pre-commit不是真正提交了，可以回滚的，当两次checkpoint间某operator挂了，此时sink端预提交的数据操作会被回滚，然后从最后一次checkpoint成功的地方开始处理，实现了exactly-once语义。

 

实现：

 

具体实现主要是通过继承TwoPhaseCommitSinkFunction，重写里面的方法，关闭mysql自动提交，在commit（）方法中真正提交，abort（）方法中rollback

 

主要代码如下：

 

public class TwoPhaseCommitMysqlConsumerDemo {

 

    public static void main（String[] args） throws Exception{

 

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment（）;

 

        //  设置ck属性

 

        env.setStateBackend（new FsStateBackend（"hdfs://zcx1:9000:/flink/ck"））;

 

        env.enableCheckpointing（10000）;

 

        env.getCheckpointConfig（）。setCheckpointingMode（CheckpointingMode.EXACTLY_ONCE）;

 

        env.getCheckpointConfig（）。setMinPauseBetweenCheckpoints（500）;

 

        env.getCheckpointConfig（）。setCheckpointTimeout（60000）;

 

        env.getCheckpointConfig（）。setFailOnCheckpointingErrors（false）;

 

        env.getCheckpointConfig（）。enableExternalizedCheckpoints（CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION）;

 

        env.getCheckpointConfig（）。setMaxConcurrentCheckpoints（1）;

 

        Properties properties=new Properties（）;

 

        properties.setProperty（ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"zcx4:9092,zcx5:9092,zcx6:9092"）;

 

        //设置读取已提交的数据

 

        properties.setProperty（ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed"）;

 

        properties.setProperty（ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false"）;

 

        FlinkKafkaConsumer kafka = new FlinkKafkaConsumer（"zcx2222", new SimpleStringSchema（），properties）;

 

        DataStreamSource dataStreamSource = env.addSource（kafka）;

 

        SingleOutputStreamOperator sum = dataStreamSource.flatMap（new FlatMapFunction<String, WC>（） {

 

            @Override

 

            public void flatMap（String s, Collector<WC> collector） throws Exception {

 

                String[] split = s.split（" "）;

 

                for （String ss : split） {

 

                    collector.collect（new WC（ss, 1））;

 

                }

 

            }

 

        }）。keyBy（"word"）。sum（"num"）;

 

        sum.print（）;

 

        TwoPhaseCommitSinkFunction<WC, MyConnection, Void> twoPhaseCommitSinkFunction = new TwoPhaseCommitSinkFunction<WC, MyConnection, Void>（new KryoSerializer<MyConnection>（MyConnection.class, new ExecutionConfig（））， new VoidSerializer（）） {

 

            @Override

 

            protected void invoke（MyConnection myconnection, WC wc, Context context） throws Exception {

 

                PreparedStatement preparedStatement = myconnection.connection.prepareStatement（"insert into wc values（？,?） on duplicate key update num=?"）;

 

                preparedStatement.setString（1, wc.word）;

 

                preparedStatement.setInt（2, wc.num）;

 

                preparedStatement.setInt（3, wc.num）;

 

                preparedStatement.executeUpdate（）;

 

                preparedStatement.close（）;

 

            }

 

            @Override

 

            protected MyConnection beginTransaction（） throws Exception {

 

                Class.forName（"com.mysql.jdbc.Driver"）;

 

                Connection connection = DriverManager.getConnection（"jdbc:mysql://localhost:3306/test", "root", "root"）;

 

                //关闭自动提交

 

                connection.setAutoCommit（false）;

 

                return new MyConnection（connection）;

 

            }

 

            @Override

 

            protected void preCommit（MyConnection myconnection） throws Exception {

 

            //invoke中完成

 

            }

 

            @Override

 

            protected void commit（MyConnection myconnection） {

 

                try {

 

                    myconnection.connection.commit（）;

 

                    myconnection.connection.close（）;

 

                } catch （SQLException e） {

 

                    e.printStackTrace（）;

 

                }

 

            }

 

            @Override

 

            protected void abort（MyConnection myconnection） {

 

                try {

 

                    myconnection.connection.rollback（）;

 

                    myconnection.connection.close（）;

 

                } catch （SQLException e） {

 

                    e.printStackTrace（）;

 

                }

 

            }

 

        };

 

        sum.addSink（twoPhaseCommitSinkFunction）;

 

        env.execute（）;

 

    }

 

}

 

class MyConnection{

 

 transient Connection connection;

 

    public MyConnection（Connection connection） {

 

        this.connection = connection;

 

    }

 

}