舉例說明

甲乙丙丁四人要組織一個(gè)會(huì)議，需要確定會(huì)議時(shí)間，不妨設(shè)甲是協(xié)調(diào)者，乙丙丁是參與者。

投票階段：

（1）甲發(fā)郵件給乙丙丁，周二十點(diǎn)開會(huì)是否有時(shí)間；

（2）甲回復(fù)有時(shí)間；

（3）乙回復(fù)有時(shí)間；

（4）丙遲遲不回復(fù)，此時(shí)對(duì)于這個(gè)活動(dòng)，甲乙丙均處于阻塞狀態(tài)，算法無法繼續(xù)進(jìn)行；

（5）丙回復(fù)有時(shí)間（或者沒有時(shí)間）；

提交階段：

（1）協(xié)調(diào)者甲將收集到的結(jié)果反饋給乙丙丁（什么時(shí)候反饋，以及反饋結(jié)果如何，在此例中取決與丙的時(shí)間與決定）；

（2）乙收到；

（3）丙收到；

（4）丁收到；

?

?

來源

代碼以[4]為主，我自己修正了一些bug，增加了依賴。

集群準(zhǔn)備工作

啟動(dòng)Zookeeper集群、Kafka集群、Flink集群。

流程圖

?

?

?

Mysql準(zhǔn)備工作

create database test;

use test;

CREATE TABLE `mysqlExactlyOnce_test` (
? `id` bigint(20) NOT NULL AUTO_INCREMENT,
? `value` varchar(255) DEFAULT NULL,
? `insert_time` datetime DEFAULT NULL,
? PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

?

文件說明與聯(lián)系

文件	說明	需要設(shè)定的參數(shù)
DBConnectUtil.java	java連接數(shù)據(jù)庫	無
KafkaUtils.java(生產(chǎn)者,頂層文件)	生產(chǎn)者	broker_list
MysqlExactlyOncePOJO.java	一個(gè)pojo對(duì)象	無
MySqlTwoPhaseCommitSink.java(被調(diào)用)	插入數(shù)據(jù),制造異常	jdbc:mysql://以及sql語句
streamdemokafka2mysql.java(調(diào)用MySqlTwoPhaseCommitSink,頂層架構(gòu))	Flink source與sink	BOOTSTRAP_SERVERS_CONFIG

運(yùn)行步驟

①運(yùn)行StreamDemoKafka2Mysql.java
②運(yùn)行KafkaUtils.java

?

調(diào)試用命令

$KAFKA/bin/kafka-topics.sh --list ?--zookeeper ?Desktop:2181

$KAFKA/bin/kafka-console-consumer.sh ?--bootstrap-server Desktop:9091 --from-beginning --topic mysql-exactly-Once-4

修改的主題的話，需要同時(shí)在kafkautils.java和StreamDemoKafka2Mysql.java中修改

實(shí)驗(yàn)效果

?

?

?

tijiao最終代碼如下

?

Reference:

[1]Flink兩階段提交

[2]Flink 之 MySQL二階提交

[3]解決Flink消費(fèi)Kafka信息，結(jié)果存儲(chǔ)在Mysql的重復(fù)消費(fèi)問題

[4]Flink實(shí)現(xiàn)Kafka到Mysql的Exactly-Once

[5]Kafka常用命令

[6]一分鐘了解兩階段提交2PC（運(yùn)營(yíng)MM也懂了）

總結(jié)

以上是生活随笔為你收集整理的flink二阶提交(没有搞完)的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。