DataX tutorial (09) - how does dataX achieve speed limit?

01 introduction

Through the previous blog , We are right. DataX Have a certain in-depth understanding ：

• 《DataX course （01）- introduction 》
• 《DataX course （02）- IDEA function DataX Complete process （ Fill all the holes ）》
• 《DataX course （03）- Source code interpretation （ Super detailed Edition ）
• 《DataX course （04）- The configuration is complete 》
• 《DataX course （05）- DataX Web Project practice 》
• 《DataX course （06）- DataX tuning 》
• 《DataX course （07）- The illustration DataX Task assignment and execution process 》
• 《DataX course （08）- Monitoring and reporting 》

As for DataX Learn more , I raised a question , Exactly? DataX How to limit the speed ？ This article will explain .

02 Reverse positioning code

We know it's in core.json In the document speed Speed limit in the method DataX Of , Can pass record Number of records and byte Bytes to speed limit :

This configuration is CoreConstant Class defines :

Select constant , Right click to use IDEA Of Find Usages… You can see that this value is called in two places ：

Next , Look at these two configurations in Channel Class to implement speed limit .

03 Channel Class

From the chart , You can see in the Channel On initialization , The number of records that have initiated the speed limit (recordSpeed) And the number of bytes (byteSpeed) , Next Control+F have a look recordSpeed Where was it called .

You can see in the statPush The method uses ：

statPush Description of the whole process ：

1. Judge byteSpeed（bps） and recordSpeed（tps） Are they all greater than 0？ If not , The exit ;
2. Based on the current byteSpeed And set byteSpeed contrast , Find the sleep time （ The formula ：currentByteSpeed * interval / this.byteSpeed- interval;）
3. Based on the current recordSpeed And set recordSpeed contrast , Find the sleep time （ The formula ：currentRecordSpeed * interval / this.recordSpeed - interval;）
4. Take the maximum sleep time ;
5. Thread.sleep(sleepTime) To sleep

Stick it on the bottom statPush Complete code for ：

private void statPush(long recordSize, long byteSize) {
    
      currentCommunication.increaseCounter(CommunicationTool.READ_SUCCEED_RECORDS,
              recordSize);
      currentCommunication.increaseCounter(CommunicationTool.READ_SUCCEED_BYTES,
              byteSize);
      // Statistics while reading waitCounter that will do , Because of the writing （pull） It might be blocking , But when I read it, I can read the blocking counter Count 

      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) {
    
                  //  Calculation basis byteLimit Sleep time obtained 
                  byteLimitSleepTime = currentByteSpeed * interval / this.byteSpeed
                          - interval;
              }
          }

          if (isChannelRecordSpeedLimit) {
    
              long currentRecordSpeed = (CommunicationTool.getTotalReadRecords(currentCommunication) -
                      CommunicationTool.getTotalReadRecords(lastCommunication)) * 1000 / interval;
              if (currentRecordSpeed > this.recordSpeed) {
    
                  //  Calculation basis recordLimit Sleep time obtained 
                  recordLimitSleepTime = currentRecordSpeed * interval / this.recordSpeed
                          - interval;
              }
          }

          //  Sleep time greater 
          long sleepTime = byteLimitSleepTime < recordLimitSleepTime ?
                  recordLimitSleepTime : byteLimitSleepTime;
          if (sleepTime > 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 At the end of the article

By reading this article, you can know DataX The principle of speed limit , If there is any doubt about children's shoes , Welcome to leave a message , The end of this paper ！