背景介绍

业务中有用到flink读取mysql数据,然后实时处理回写到mysql和doris供c端使用,代码的大体逻辑都是
1.创建执行环境
2.写需要使用到表的flinksql ddl语句
3.通过flink sql 写核心的业务逻辑
4.将sql处理的结果通过toRetractStream来转换成一个可撤回流
5.对这个流进行其他处理,比如通过异步IO关联一些高基维表,数据解密,脱敏,过滤脏数据等工作
6.将处理后的结果数据写入到mysql和doris
但是在之前的代码中存在了很多的冗余代码和硬编码,存在如下问题
1.每个任务环境构建的大部分代码都是一样的
2.每一个表都需要手动去写flinksql ddl语句,而且直接写死在代码中,导致代码很长
3.核心的sql处理逻辑也是直接写到了代码中,不方便阅读和修改
4.表转流后的数据类型都需要手动写一个对应的实体类,每次加字段或者修改会很麻烦
5.JdbcSink 的sql语句需要写很长,而且还有写对应的PreparedStatement
6.存在多个库多个环境,数据库的配置信息太多没有一个统一管理的地方

优化思路

1.重复的代码抽取成工具类
2.数据库配置写到配置文件中而不是代码中
3.flink sql ddl语句根据表名和数据库名自动生成
4.核心的sql语句放到文件中,用的时候读取
5.表转流后统一转换成Row类型
6.写一个通用的JdbcSink方法,传入数据库配置,表名称,写入的字段即可自动生成一个SinkFunction
7.mybatis SqlSessionFactory存在多个,写一个工厂类型,通过传入不同的配置文件来生成不同的SqlSessionFactory
8.工具类的开发,数据库连接工具类,线程池工具类,文件读取工具类,环境构建工具类
9.将代码中的一些常量抽取出来放到一个全局的常量类中

遇到的一些问题

长时间没有数据到来,数据库连接失效的问题

解决方案 :
1.jdbc url 添加参数 &autoReconnect=true&maxReconnects=3
2.用完的连接要通过close方法归还到连接池,而不是其放到一个成员变量上长时间使用

mybatis sqlSession失效

使用mybatis的sqlSession用完后应该使用close方法释放资源,这里和数据库连接池的使用是一样的道理

flinkcdc server-id 冲突的问题

不知道之前是哪位大哥在网上写的server-id必须是5400-6400之间,所以一直在使用这个范围内的server-id,经常会遇到server-id冲突的问题,直到前段时间咨询了社区大佬并且也阅读了相关的源码才发现,server-id的范围是整型数字的方位,最大2147483647,所以可以使用的范围是极大的.