生产环境FlinkCDC实时数据同步高可用，值得收藏

你好呀，我是小羊。

1.简介

上篇文章我讲了一下如何基于FlinkCDC来实时同步数据库的数据到ES。说明在流程上是可行的，技术实现没什么问题。

2.高可用的思考

当然，这仅仅只是一个demo,只能说明在流程上是可行的，在我们的实际生产环境中，可能会遇到各种突发情况，比如进程挂了，节点挂了，网络异常，虚拟机重启等等各种问题。而实际生产环境是很难容忍服务中断的问题的，试想如果你在京东下了一个单，钱已经付了，但是因为数据同步的问题，导致你一直查不到这个订单。你会不会非常生气。那就是会投诉走起。到时候就得用程序员祭天了。

那么如何保证从数据库到我们的ES如何保证高可用呢？flink的同步任务的正常运行，才能整体的数据同步正常。

所以，至少要做到以下几点：

flink 运行正常。
flink 任务保证正常运行，如果挂了，需要有重启和故障转移的能力
任务异常重启，需要从异常的地方恢复，保证数据不丢失。

3.高可用的实现方案

1. flink 集群部署

Flink的核心是一个分布式数据流引擎，它可以在大规模集群上运行。它采用了一种基于事件时间的处理模型，能够处理乱序事件和延迟事件。同时，它还提供了高可用性、容错性和动态伸缩性等特性，可以保证处理过程的稳定性和可靠性，flink要做到高可用，可以采用集群部署的方案，这样能够在一个节点挂了的时候，不会影响到整体任务的运行。flink 提供了多种集群方案，比如standalone，yarn,k8s 等多种集群部署方式。我这边采用standalone集群方式。

我们准备三个节点：

节点	任务
10.82.64.67	jobManager taskmanager
10.82.64.68	jobManager taskmanager
10.82.64.69	taskmanager

前期准备：

三台节点配置好免密登录
安装好jdk环境
安装好flink

准备集群配置将flink每个节点的 conf/flink-conf.yml 修改jobmanager如下：

jobmanager.rpc.address: 10.82.64.67
jobmanager.rpc.port: 6123

卡槽数可以配置多一点，以便运行更多的任务。

taskmanager.numberOfTaskSlots: 12

masters 配置主节点

10.82.64.67:8081

workers配置工作节点

10.82.64.67
10.82.64.68
10.82.64.69

然后在 10.82.64.67 节点启动:

bin/start-cluster.sh

启动成功可以看到三个节点正常运行

flink中jobmanager负责协调整个集群的任务运行，taskmanager 负责具体的任务执行。如果只是taskmanager 挂了可以把任务转移到其他节点运行，但是jobmanager 挂了整个集群都不能正常工作了，这对于生产业务来讲也是不可以接受的，所以jobmanager 也需要配置多个。flink 提供了HA 模式，可以保证有多个jobmanager 同时运行，一个jobmanager 挂了，可以迅速切换成另外一个备用的jobmanager。提升整个flink 集群的可用性。

HA 是依赖于zookeeper 来做分布式协调的，正好flink 也提供了zookeeper，我这边就使用flink自带的。

运行zookeeper

/bin/start-zookeeper-quorum.sh

conf/flink-conf.yml 配置HA高可用

high-availability: zookeeper
high-availability.storageDir: file:///app/flink/ha/
high-availability.zookeeper.quorum: 10.82.64.67:2181

high-availability.storageDir 是高可用的保存目录，集群中可以使用hdfs 或者 nfs 等等，只要能保证flink集群中每个节点都能访问到即可。
high-availability.zookeeper.quorum是zookeeper的访问地址

masters 配置多个主节点

10.82.64.67
10.82.64.68

重启整个flink集群

./stop-cluster.sh && ./start-cluster.sh

可以发现 67 68 节点分别都有一个jobmanager

将其中一个jobmanager kill 掉

kill -9 79524

整个集群也是能正常工作的。

2. 配置任务自动恢复

上面配置了整个flink 集群的高可用，但是任务挂了不能保证它还能恢复，因为flink默认的配置是不会重启的，我们可以配置一下任务的重启策略：

conf/flink-conf.yml 配置重启策略

restart-strategy: fixed-delay 
restart-strategy.fixed-delay.attempts: 3 
restart-strategy.fixed-delay.delay: 10 s

这个是固定的重启策略，如果任务失败，就执行重启，尝试三次，每次间隔10s。

除了固定重启策略外。flink 还提供了其他的重启策略

Flink提供了多种任务重启策略，可以根据不同的需求进行选择和配置。以下是常见的几种任务重启策略：

失败率重启策略(FailureRateRestartStrategy)：在一段时间内，如果任务失败率超过一定阈值，则触发重启。可以配置重启尝试的次数、时间间隔和失败率的阈值。
无限重启策略(InfiniteDelayRestartStrategy)：在任务失败后，无限重启任务，直到手动停止任务或者达到一定的重启次数。
随机延迟重启策略(RandomDelayRestartStrategy)：在任务失败后，等待一段随机时间后重启任务。可以配置重启尝试的次数和重启的时间间隔的范围。
外部重启策略(ExternalizedRestartStrategy)：将任务状态保存在外部存储中，在任务失败后，可以从外部存储中恢复任务状态并重启任务。

以上是常见的几种任务重启策略，Flink还提供了自定义重启策略的接口，可以根据具体需求实现自定义的重启策略。

3. 配置checkpoint定时保存任务状态

如果任务重启，还需要保证是从故障处恢复的，flink提供了checkpoint机制，定时把任务状态保存下来。这样，如果一个任务挂了并且重启成功，也能从最后的任务状态恢复，保证数据不丢失。

flinksql 配置：

set execution.checkpointing.interval=1000;
set state.checkpoints.dir=file:///app/flink/checkpoint/;
set execution.checkpointing.externalized-checkpoint-retention=RETAIN_ON_CANCELLATION;

set execution.checkpointing.interval 是自动checkpoint的间隔时间

set state.checkpoints.dir 是自动checkpoint保存的路径,同样可以使用nfs 或者 hdfs。

尝试kill 一个正在执行任务的 taskmanager。发现几秒钟之后任务会自动重启，状态也是从重启的时候恢复的。

Flink的Checkpoint机制是一种容错机制，用于确保在发生故障时，作业能够从故障前的状态进行恢复。Checkpoint机制通过定期将作业的状态数据保存到持久化存储中，以实现作业的容错和恢复。