DataX教程（09）- DataX是如何做到限速的？

01 引言

通过前面的博文，我们对DataX有了一定的深入了解了：

• 《DataX教程（01）- 入门》
• 《DataX教程（02）- IDEA运行DataX完整流程（填完所有的坑）》
• 《DataX教程（03）- 源码解读（超详细版）
• 《DataX教程（04）- 配置完整解读》
• 《DataX教程（05）- DataX Web项目实践》
• 《DataX教程（06）- DataX调优》
• 《DataX教程（07）- 图解DataX任务分配及执行流程》
• 《DataX教程（08）- 监控与汇报》

随着对DataX深入学习，我提出了一个疑问，究竟DataX是如何做到限速的？本文来讲解下。

02 逆向定位代码

我们知道是在core.json文件里面的speed方法里面限速DataX的，可以通过record记录数和byte字节数来限速: 这个配置在CoreConstant类里面定义了: 选中常量，右键使用IDEA的Find Usages…可以看到有两个地方调用了这个值： 接下来，看看这两个配置在Channel类如何实现限速的。

03 Channel类里实现限速

从下图，可以看到在Channel初始化时，顺带初始化了限速的记录数(recordSpeed)以及字节数(byteSpeed) ，接下来Control+F看看recordSpeed在哪里调用了。

可以看到在statPush方法里面用到了： statPush整个流程的描述：

1. 判断byteSpeed（bps）和recordSpeed（tps）是否都大于0？如果不是，则退出；
2. 根据当前的byteSpeed和设定的byteSpeed对比，求出睡眠时间（公式：currentByteSpeed * interval / this.byteSpeed- interval;）
3. 根据当前的recordSpeed和设定的recordSpeed对比，求出睡眠时间（公式：currentRecordSpeed * interval / this.recordSpeed - interval;）
4. 取休眠时间最大值；
5. Thread.sleep(sleepTime)来休眠

下面贴上statPush的完整代码：

private void statPush(long recordSize, long byteSize) {
      currentCommunication.increaseCounter(CommunicationTool.READ_SUCCEED_RECORDS,
              recordSize);
      currentCommunication.increaseCounter(CommunicationTool.READ_SUCCEED_BYTES,
              byteSize);
      //在读的时候进行统计waitCounter即可，因为写（pull）的时候可能正在阻塞，但读的时候已经能读到这个阻塞的counter数

      currentCommunication.setLongCounter(CommunicationTool.WAIT_READER_TIME, waitReaderTime);
      currentCommunication.setLongCounter(CommunicationTool.WAIT_WRITER_TIME, waitWriterTime);

      boolean isChannelByteSpeedLimit = (this.byteSpeed > 0);
      boolean isChannelRecordSpeedLimit = (this.recordSpeed > 0);
      if (!isChannelByteSpeedLimit && !isChannelRecordSpeedLimit) {
          return;
      }

      long lastTimestamp = lastCommunication.getTimestamp();
      long nowTimestamp = System.currentTimeMillis();
      long interval = nowTimestamp - lastTimestamp;
      if (interval - this.flowControlInterval >= 0) {
          long byteLimitSleepTime = 0;
          long recordLimitSleepTime = 0;
          if (isChannelByteSpeedLimit) {
              long currentByteSpeed = (CommunicationTool.getTotalReadBytes(currentCommunication) -
                      CommunicationTool.getTotalReadBytes(lastCommunication)) * 1000 / interval;
              if (currentByteSpeed > this.byteSpeed) {
                  // 计算根据byteLimit得到的休眠时间
                  byteLimitSleepTime = currentByteSpeed * interval / this.byteSpeed
                          - interval;
              }
          }

          if (isChannelRecordSpeedLimit) {
              long currentRecordSpeed = (CommunicationTool.getTotalReadRecords(currentCommunication) -
                      CommunicationTool.getTotalReadRecords(lastCommunication)) * 1000 / interval;
              if (currentRecordSpeed > this.recordSpeed) {
                  // 计算根据recordLimit得到的休眠时间
                  recordLimitSleepTime = currentRecordSpeed * interval / this.recordSpeed
                          - interval;
              }
          }

          // 休眠时间取较大值
          long sleepTime = byteLimitSleepTime  0) {
              try {
                  Thread.sleep(sleepTime);
              } catch (InterruptedException e) {
                  Thread.currentThread().interrupt();
              }
          }

          lastCommunication.setLongCounter(CommunicationTool.READ_SUCCEED_BYTES,
                  currentCommunication.getLongCounter(CommunicationTool.READ_SUCCEED_BYTES));
          lastCommunication.setLongCounter(CommunicationTool.READ_FAILED_BYTES,
                  currentCommunication.getLongCounter(CommunicationTool.READ_FAILED_BYTES));
          lastCommunication.setLongCounter(CommunicationTool.READ_SUCCEED_RECORDS,
                  currentCommunication.getLongCounter(CommunicationTool.READ_SUCCEED_RECORDS));
          lastCommunication.setLongCounter(CommunicationTool.READ_FAILED_RECORDS,
                  currentCommunication.getLongCounter(CommunicationTool.READ_FAILED_RECORDS));
          lastCommunication.setTimestamp(nowTimestamp);
      }
  }

04 文末

通过阅读本文可以知道DataX的限速原理了，如有疑问的童鞋，欢迎留言，本文完！