本文主要是介绍FlinkX流控实现，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
FlinkX流控实现
流量控制防止并发性能过高对源数据库造成影响。
在FlinkX中，流量控制是采用Byte为单位/s进行流量控制的。
配置参数实例：
{“speed”: {"bytes": 0}
}
当 bytes > 0 时，才会开启流量控制。
整个计算的速率是根据整个系统中的指标，按照每秒的窗口，实时计算出限流的速率的。通过对读取记录的限流，但是没有背压。
实现逻辑
首先我们看一下读取器的初始化过程，
InputFormat初始化
public void open(InputSplit inputSplit) throws IOException {checkIfCreateSplitFailed(inputSplit);if(!inited){// 初始化累加器收集器，每秒从FlinkAPI读取对应累加器的值，并设置为global值initAccumulatorCollector();// 初始化指标累加器，每次调用nextRecord时提交指标initStatisticsAccumulator();// 开启限流器openByteRateLimiter();initRestoreInfo();if(restoreConfig.isRestore()){formatState.setNumOfSubTask(indexOfSubTask);}inited = true;}openInternal(inputSplit);}
本次只看前三个。
-
初始化累加器收集器，每秒从FlinkAPI读取对应累加器的值，并设置为global值(下文中计算速率是有用到)
-
初始化指标累加器，每次调用
nextRecord时提交指标 -
开启限流器
-
初始化Restore配置（本章不讲，后续章节有用到）
我们重点详解一下前三个步骤：
在详解每一个步骤之前，首先了解下在数据同步过程中具体的指标
指标详情
| 分类 | 指标名称 | 含义 |
|---|---|---|
| 读取指标 | numRead | 累计读取数据条数 |
| byteRead | 累计读取数据字节数 | |
| readDuration | 读取数据的总时间 | |
| 写入指标 | numWrite | 累计写入数据条数 |
| byteWrite | 累计写入数据字节数 | |
| writeDuration | 写入数据的总时间 | |
| 错误指标 | nErrors | 累计错误记录数 |
| nullErrors | 累计空指针错误记录数 | |
| duplicateErrors | 累计主键冲突错误记录数 | |
| conversionErrors | 累计类型转换错误记录数 | |
| otherErrors | 累计其它错误记录数 |
全局指标实现
如何控制全局限流，很重要的一环就是收集到全局系统的关键状况，无论是微服务调用还是读取限流本质都是同一个道理。首先需要找一个全局存储提供这些指标的存储和更新，FlinkX在这里使用的Flinx的累加器。
指标初始化
private void initStatisticsAccumulator(){numReadCounter = getRuntimeContext这篇关于FlinkX流控实现的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
