FlinkX流控实现

淡淡的烟草味﹌ 2023-02-11 15:27 79阅读 0赞

# FlinkX流控实现 #

流量控制防止并发性能过高对源数据库造成影响。

在FlinkX中，流量控制是采用Byte为`单位/s`进行流量控制的。

配置参数实例：

{ 
        “speed”: { 
            "bytes": 0
        }
    }

当 bytes > 0 时，才会开启流量控制。

**整个计算的速率是根据整个系统中的指标，按照每秒的窗口，实时计算出限流的速率的。通过对读取记录的限流，但是没有背压。**

--------------------

![image-20200524100353781][]

限流原理

## 实现逻辑 ##

首先我们看一下读取器的初始化过程，

### 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);
        }

本次只看前三个。

1.  初始化累加器收集器，每秒从FlinkAPI读取对应累加器的值，并设置为global值(下文中计算速率是有用到)
2.  初始化指标累加器，每次调用`nextRecord`时提交指标
3.  开启限流器
4.  初始化Restore配置（本章不讲，后续章节有用到）

我们重点详解一下前三个步骤：

在详解每一个步骤之前，首先了解下在数据同步过程中具体的指标

### 指标详情 ###

<table> 
 <thead> 
  <tr> 
   <th>分类</th> 
   <th>指标名称</th> 
   <th>含义</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>读取指标</td> 
   <td>numRead</td> 
   <td>累计读取数据条数</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>byteRead</td> 
   <td>累计读取数据字节数</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>readDuration</td> 
   <td>读取数据的总时间</td> 
  </tr> 
  <tr> 
   <td>写入指标</td> 
   <td>numWrite</td> 
   <td>累计写入数据条数</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>byteWrite</td> 
   <td>累计写入数据字节数</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>writeDuration</td> 
   <td>写入数据的总时间</td> 
  </tr> 
  <tr> 
   <td>错误指标</td> 
   <td>nErrors</td> 
   <td>累计错误记录数</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>nullErrors</td> 
   <td>累计空指针错误记录数</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>duplicateErrors</td> 
   <td>累计主键冲突错误记录数</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>conversionErrors</td> 
   <td>累计类型转换错误记录数</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>otherErrors</td> 
   <td>累计其它错误记录数</td> 
  </tr> 
 </tbody> 
</table>

### 全局指标实现 ###

如何控制全局限流，很重要的一环就是收集到全局系统的关键状况，无论是微服务调用还是读取限流本质都是同一个道理。首先需要找一个全局存储提供这些指标的存储和更新，FlinkX在这里使用的Flinx的`累加器`。

![image-20200524101615078][]

#### 指标初始化 ####

private void initStatisticsAccumulator(){ 
            numReadCounter = getRuntimeContext().getLongCounter(Metrics.NUM_READS);
            bytesReadCounter = getRuntimeContext().getLongCounter(Metrics.READ_BYTES);
            durationCounter = getRuntimeContext().getLongCounter(Metrics.READ_DURATION);
    
            inputMetric = new BaseMetric(getRuntimeContext());
            inputMetric.addMetric(Metrics.NUM_READS, numReadCounter, true);
            inputMetric.addMetric(Metrics.READ_BYTES, bytesReadCounter, true);
            inputMetric.addMetric(Metrics.READ_DURATION, durationCounter);
        }

#### 指标上报 ####

@Override
        public Row nextRecord(Row row) throws IOException {
            // 获取本地信号量
            if(byteRateLimiter != null) {
                byteRateLimiter.acquire();
            }
            Row internalRow = nextRecordInternal(row);
            if(internalRow != null){
                internalRow = setChannelInformation(internalRow);
    
                updateDuration();
                // 更新累加器
                if(numReadCounter !=null ){
                    numReadCounter.add(1);
                }
                if(bytesReadCounter!=null){
                    bytesReadCounter.add(internalRow.toString().length());
                }
            }
    
            if (testConfig.errorTest() && testConfig.getFailedPerRecord() > 0) {
                numReadeForTest++;
                if (numReadeForTest > testConfig.getFailedPerRecord()) {
                    throw new RuntimeException(testConfig.getErrorMsg());
                }
            }
    
            return internalRow;
        }

#### 本地更新全局指标 ####

`AccumulatorCollector#collectAccumulatorWithApi`

private void collectAccumulatorWithApi(){ 
            for (String monitorUrl : monitorUrls) { 
                try { 
                    String response = UrlUtil.get(httpClient, monitorUrl);
                    Map<String,Object> map = gson.fromJson(response, Map.class);
                    List<LinkedTreeMap> userTaskAccumulators = (List<LinkedTreeMap>) map.get(KEY_ACCUMULATORS);
                    for(LinkedTreeMap accumulator : userTaskAccumulators) { 
                        String name = (String) accumulator.get(KEY_NAME);
                        if(name != null && !"tableCol".equalsIgnoreCase(name)) { 
                            String accValue = (String) accumulator.get(KEY_VALUE);
                            if(!"null".equals(accValue)){ 
                                long value = Double.valueOf(accValue).longValue();
                                ValueAccumulator valueAccumulator = valueAccumulatorMap.get(name);
                                if(valueAccumulator != null){ 
                                    valueAccumulator.setGlobal(value);
                                }
                            }
                        }
                    }
                } catch (Exception e){ 
                    checkErrorTimes();
                    LOG.error("Update data error,url:[{}],error info:", monitorUrl, e);
                }
                break;
            }
        }

### 限流器逻辑 ###

限流器采用的是guava的RateLimit实现，全局与本地速率根据上述指标实际得来。

#### Guava RateLimit ####

guava的RateLimiter使用的是`令牌桶算法`，也就是以固定的频率向桶中放入令牌。

在FlinkX中，是以秒为单位，计算全局的令牌个数，然后根据各个处理器的处理占比获取对应的全局令牌。

详情参考 http://ifeve.com/guava-ratelimiter/

*  **acquire()** ： 从RateLimiter获取一个许可，该方法会被阻塞直到获取到请求。如果存在等待的情况的话，告诉调用者获取到该请求所需要的睡眠时间。该方法等同于acquire(1)。
 *  **setRate()**: 更新RateLimite的稳定速率，参数permitsPerSecond 由构造RateLimiter的工厂方法提供。调用该方法后，当前限制线程不会被唤醒，因此他们不会注意到最新的速率；只有接下来的请求才会。需要注意的是，由于每次请求偿还了（通过等待，如果需要的话）上一次请求的开销，这意味着紧紧跟着的下一个请求不会被最新的速率影响到，在调用了setRate 之后；它会偿还上一次请求的开销，这个开销依赖于之前的速率。RateLimiter的行为在任何方式下都不会被改变，比如如果 RateLimiter 有20秒的预热期配置，在此方法被调用后它还是会进行20秒的预热。
 *  **create(double permitsPerSecond)**: 根据指定的稳定吞吐率创建RateLimiter，这里的吞吐率是指每秒多少许可数（通常是指QPS，每秒多少查询）。

#### 限流器创建 ####

private void openByteRateLimiter(){ 
            if (this.bytes > 0) { 
                this.byteRateLimiter = new ByteRateLimiter(accumulatorCollector, this.bytes);
                this.byteRateLimiter.start();
            }
        }
    
    
     public ByteRateLimiter(AccumulatorCollector accumulatorCollector, double expectedBytePerSecond) { 
            double initialRate = 1000.0;
            this.rateLimiter = RateLimiter.create(initialRate);
            this.expectedBytePerSecond = expectedBytePerSecond;
            this.accumulatorCollector = accumulatorCollector;
    
            ThreadFactory threadFactory = new BasicThreadFactory
                    .Builder()
                    .namingPattern("ByteRateCheckerThread-%d")
                    .daemon(true)
                    .build();
            scheduledExecutorService = new ScheduledThreadPoolExecutor(1, threadFactory);
        }

*  从上我们可以知道，bytes配置必须大于0。限流器才会开启
 *  ByteRateLimiter 的默认控制速率是1000row/s，**逗**

#### 令牌获取 ####

@Override
        public Row nextRecord(Row row) throws IOException { 
            if(byteRateLimiter != null) { 
                byteRateLimiter.acquire();
            }
            // ...
    
            return internalRow;
        }

每次调用nextRecord 读取的时候，判断是否还有有效的令牌。

#### 流速计算 ####

`ByteRateLimiter#updateRate`

private void updateRate(){ 
            // 读取累加器里面记录的值。
            long totalBytes = accumulatorCollector.getAccumulatorValue(Metrics.READ_BYTES); // 全局读取bytes，从累加器中读取
            long thisRecords = accumulatorCollector.getLocalAccumulatorValue(Metrics.NUM_READS); // 本地读取记录数
            long totalRecords = accumulatorCollector.getAccumulatorValue(Metrics.NUM_READS); // 全局读取记录数
    
            double thisWriteRatio = (totalRecords == 0 ? 0 : thisRecords / (double)totalRecords); // 计算写入速率
    
            if (totalRecords > MIN_RECORD_NUMBER_UPDATE_RATE && totalBytes != 0 && thisWriteRatio != 0) { 
                double bpr = totalBytes / totalRecords; // 根据全局计算每个记录的平均bytes
                double permitsPerSecond = expectedBytePerSecond / bpr * thisWriteRatio; // 根据计算的平均值，计算全局的每秒的放行记录数，乘上本地占比，就是本地应该的速率，
                rateLimiter.setRate(permitsPerSecond);
            }
        }

1.  读取本地累加器的值以及全局读取累加器的值
2.  计算本地读取记录对全局读取记录的占比
3.  根据计算的平均值，计算全局的下一秒秒的放行记录数，乘上本地记录占比，就是本地应该限流的速率。

#### 定时更新令牌桶 ####

public void start(){ 
            scheduledExecutorService.scheduleAtFixedRate(this::updateRate,0, 1000L, TimeUnit.MILLISECONDS);
        }

每秒更新令牌。

## 总结 ##

本篇针对于整个Flinx的限流实现进行了深入的分析，从中我们应该可以了解到全局限流的实现方式以及限流器信号量的原理，并且可以旁敲侧击的了解一下整个InputeFormat的初始化过程和指标。

[image-20200524100353781]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93dWh1bGFsYS5vc3MtY24tc2hhbmdoYWkuYWxpeXVuY3MuY29tL2Jsb2cvaW1hZ2UtMjAyMDA1MjQxMDAzNTM3ODEucG5n?x-oss-process=image/format,png
[image-20200524101615078]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93dWh1bGFsYS5vc3MtY24tc2hhbmdoYWkuYWxpeXVuY3MuY29tL2Jsb2cvaW1hZ2UtMjAyMDA1MjQxMDE2MTUwNzgucG5n?x-oss-process=image/format,png