先拋出代碼

package com.dpf.flink;import com.dpf.flink.sink.MysqlSink;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.util.Properties;public class MysqlTwoPhaseCommit {//topicprivate static final String topic_ExactlyOnce = "TwoPhaseCommit";public static void main(String[] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//設置并行度,為了方便測試，查看消息的順序，這里設置為1，可以更改為多并行度env.setParallelism(1);//checkpoint的設置//每隔10s進行啟動一個檢查點【設置checkpoint的周期】env.enableCheckpointing(30000);//設置模式為：exactly_one，僅一次語義env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);//確保檢查點之間有1s的時間間隔【checkpoint最小間隔】env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000);//檢查點必須在10s之內完成，或者被丟棄【checkpoint超時時間】env.getCheckpointConfig().setCheckpointTimeout(10000);//同一時間只允許進行一次檢查點env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);//表示一旦Flink程序被cancel后，會保留checkpoint數據，以便根據實際需要恢復到指定的checkpointenv.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);//設置statebackend,將檢查點保存在hdfs上面，默認保存在內存中。這里先保存到本地env.setStateBackend(new FsStateBackend("file:///Users/david.dong/tmp/flink/checkpoint"));//設置kafka消費參數Properties properties = new Properties();properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, topic_ExactlyOnce);/*SimpleStringSchema可以獲取到kafka消息，JSONKeyValueDeserializationSchema可以獲取都消息的key,value，metadata:topic,partition，offset等信息*/FlinkKafkaConsumer<String> kafkaConsumer011 = new FlinkKafkaConsumer<>(topic_ExactlyOnce,new SimpleStringSchema(),properties);//加入kafka數據源DataStreamSource<String> streamSource = env.addSource(kafkaConsumer011);SingleOutputStreamOperator<Tuple2<String, Integer>> tupleStream = streamSource.map(str -> Tuple2.of(str, 1)).returns(Types.TUPLE(Types.STRING, Types.INT));tupleStream.print();//數據傳輸到下游tupleStream.addSink(new MysqlSink()).name("MySqlTwoPhaseCommitSink");//觸發執行env.execute("StreamDemoKafka2Mysql");}
}

package com.dpf.flink.sink;import com.dpf.flink.utils.DBConnectUtil;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.base.VoidSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.Connection;
import java.sql.PreparedStatement;public class MysqlSink extends TwoPhaseCommitSinkFunction<Tuple2<String,Integer>, Connection,Void> {private static final Logger log = LoggerFactory.getLogger(MysqlSink.class);public MysqlSink() {super(new KryoSerializer<>(Connection.class,new ExecutionConfig()), VoidSerializer.INSTANCE);}/*** 執行數據庫入庫操作  task初始化的時候調用* @param connection* @param tuple* @param context* @throws Exception*/@Overrideprotected void invoke(Connection connection, Tuple2<String, Integer> tuple, Context context) throws Exception {log.info("start invoke...");String value = tuple.f0;Integer total = tuple.f1;String sql = "update student set name = 'aaa' where id = 1";log.info("====執行SQL:{}===",sql);PreparedStatement ps = connection.prepareStatement(sql);ps.setString(1, value);ps.setInt(2, total);ps.setLong(3, System.currentTimeMillis());log.info("要插入的數據:{}----{}",value,total);if (ps != null) {String sqlStr = ps.toString().substring(ps.toString().indexOf(":")+2);log.error("執行的SQL語句:{}",sqlStr);}//執行insert語句ps.execute();}/*** 獲取連接，開啟手動提交事物（getConnection方法中）* @return* @throws Exception*/@Overrideprotected Connection beginTransaction() throws Exception {log.info("start beginTransaction.......");String url = "jdbc:mysql://localhost:3306/bigdata?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&useSSL=false&autoReconnect=true";Connection connection = DBConnectUtil.getConnection(url, "root", "12345678");return connection;}/***預提交，這里預提交的邏輯在invoke方法中* @param connection* @throws Exception*/@Overrideprotected void preCommit(Connection connection) throws Exception {log.info("start preCommit...");}/*** 如果invoke方法執行正常，則提交事務* @param connection*/@Overrideprotected void commit(Connection connection) {log.info("start commit...");DBConnectUtil.commit(connection);}/*** 如果invoke執行異常則回滾事物，下一次的checkpoint操作也不會執行* @param connection*/@Overrideprotected void abort(Connection connection) {log.info("start abort rollback...");DBConnectUtil.rollback(connection);}
}

package com.dpf.flink.utils;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;/*** 數據庫連接工具類*/
public class DBConnectUtil {private static final Logger log = LoggerFactory.getLogger(DBConnectUtil.class);/*** 獲取連接** @param url* @param user* @param password* @return* @throws SQLException*/public static Connection getConnection(String url, String user, String password) throws SQLException {Connection conn = null;try {Class.forName("com.mysql.jdbc.Driver");} catch (ClassNotFoundException e) {log.error("獲取mysql.jdbc.Driver失敗");e.printStackTrace();}try {conn = DriverManager.getConnection(url, user, password);log.info("獲取連接:{" + conn + "} 成功...");} catch (Exception e) {log.error("獲取連接失敗，url:" + url + ",user:" + user);}//設置手動提交conn.setAutoCommit(false);return conn;}/*** 提交事務*/public static void commit(Connection conn) {if (conn != null) {try {conn.commit();} catch (SQLException e) {log.error("提交事務失敗,Connection:" + conn);e.printStackTrace();} finally {close(conn);}}}/*** 事務回滾** @param conn*/public static void rollback(Connection conn) {if (conn != null) {try {conn.rollback();} catch (SQLException e) {log.error("事務回滾失敗,Connection:" + conn);e.printStackTrace();} finally {close(conn);}}}/*** 關閉連接** @param conn*/public static void close(Connection conn) {if (conn != null) {try {conn.close();} catch (SQLException e) {log.error("關閉連接失敗,Connection:" + conn);e.printStackTrace();}}}
}

這部分代碼網上抄的，但是大致不差

前提：

1.source消息密集，全據并行度設置1

2.sink就執行update操作，并且就update同一條數據，為了更好驗證問題

結果：

這邊我嘗試了很多次，中間有時候能順利執行，但是有時候程序在sink這里卡住了，過一段時間就報錯socket interrupt異常。

我的分析：

1.首先設置ck的間隔是10秒一次，那么當ck barrier到達sink算子的時候，就會進行預提交，并且立刻開啟一個新事物用來處理后續的消息。那么這里就會出現多事務同時存在的情況，比如預提交的事務A，和新開啟的事務B

2.新事物B開啟后立刻就可以繼續處理后續到來的消息。

3.那么此時如果事務A預提交后，他需要等待來自JobManager的complete指令，到代碼層面也就是調用notifyComplete方法來進行commit。那么加入在這個期間，就是還沒有收到complete指令的時候，事務B已經執行到了? ps.execute();這里，此時事務B就會卡住，因為他們都是操作同一條數據，那么問題來了，此時程序已經卡住了，也就是線程卡住了，那么此時就算JobManager發送complete指令了，然后調用notifyComplete方法，但是，此時沒有線程執行這個方法！（因為主線程卡在了（ps.execute();這里）所以整個程序就徹底卡住了。

然后flnik dag上看到的是sink紅了，前面的節點都黑了，就是背壓瞬間就很嚴重了

以上是我對flink兩階段提交存在的問題的分析，我的source消息大概100多萬，我分析是這個原因，如果分析的不對，還請大佬幫我分析下我哪里理解不對？或者為什么會卡住？