vlambda博客
学习文章列表

  public class DataTransferThrottler {  private final long period;  //限流的时间窗口，一般hard-coded为500ms   private final long periodExtension; // 对一次传输最长限流的时间，一般hard-coded为3个period周期  private long bytesPerPeriod;  //为按带宽大小，每个period内传输的数据量  private long curPeriodStart;  // 该轮throttle的起始时间  private long curReserve;      // 每个period内能传输的数据量，初始值bytesPerPeriod private long bytesAlreadyUsed;  ...... public DataTransferThrottler(long period, long bandwidthPerSec) {    this.curPeriodStart = monotonicNow(); //创建throttle时初始化 this.period = period; this.curReserve = this.bytesPerPeriod = bandwidthPerSec*period/1000; this.periodExtension = period*3; }//限流核心方法public synchronized void throttle(long numOfBytes, Canceler canceler) { if ( numOfBytes <= 0 ) { return; }  //先扣除即将传输的byte数 curReserve -= numOfBytes;    //预加上 bytesAlreadyUsed += numOfBytes;//reserve小于0则进入限流逻辑 while (curReserve <= 0) { if (canceler != null && canceler.isCancelled()) { return; } long now = monotonicNow(); long curPeriodEnd = curPeriodStart + period;     //1）只有当前时间在本周期内才限流，执行wait if ( now < curPeriodEnd ) { // Wait for next period so that curReserve can be increased. try {        //等待到下一个period start时间点 wait( curPeriodEnd - now ); } catch (InterruptedException e) { // Abort throttle and reset interrupted status to make sure other // interrupt handling higher in the call stack executes. Thread.currentThread().interrupt(); break; }        //2）now超过当前周期，但还在最大周期内，更新curPeriodStart和curReserve  } else if ( now < (curPeriodStart + periodExtension)) { curPeriodStart = curPeriodEnd;        //curReserve 加上一个周期的byte额度 curReserve += bytesPerPeriod; } else {         //3)此时时间已经超过最大period，表示已经多次没有throttle了，        // 重置curPeriodStart, 进入下一个1+3 period轮回，并清零之前的bytesAlreadyUsed // discard the prev period. Throttler might not have // been used for a long time. curPeriodStart = now; curReserve = bytesPerPeriod - bytesAlreadyUsed; } } //传输成功结束，扣除已传数的byte数 bytesAlreadyUsed -= numOfBytes; }}

StandbyCheckpointer.doCheckpoint() --->TransferFsImage.uploadImageFromStorage()    -->uploadImage()    -->writeFileToPutRequest()    -->copyFileToStream()

 ImageServlet.java //静态构造throttler，根据配置的带宽决定是否开启 限流 public static DataTransferThrottler getThrottler(Configuration conf) { long transferBandwidth = conf.getLongBytes( DFSConfigKeys.DFS_IMAGE_TRANSFER_RATE_KEY,        DFSConfigKeys.DFS_IMAGE_TRANSFER_RATE_DEFAULT);//默认52428800，说明该throttler默认开启了。 DataTransferThrottler throttler = null; if (transferBandwidth > 0) { throttler = new DataTransferThrottler(transferBandwidth); } return throttler;  }

 TransferFsImage.java private static void writeFileToPutRequest(Configuration conf, HttpURLConnection connection, File imageFile, Canceler canceler) throws IOException { connection.setRequestProperty(Util.CONTENT_TYPE, "application/octet-stream"); connection.setRequestProperty(Util.CONTENT_TRANSFER_ENCODING, "binary"); OutputStream output = connection.getOutputStream(); FileInputStream input = new FileInputStream(imageFile); try {    copyFileToStream(output, imageFile, input, //获取throttler限流器 ImageServlet.getThrottler(conf), canceler); } finally { IOUtils.closeStream(input); IOUtils.closeStream(output); } }  private static void copyFileToStream(OutputStream out, File localfile, FileInputStream infile, DataTransferThrottler throttler, Canceler canceler) throws IOException { byte buf[] = new byte[IO_FILE_BUFFER_SIZE]; long total = 0; int num = 1; IOException ioe = null;... try { while (num > 0) { ... num = infile.read(buf); if (num <= 0) { break; } ... out.write(buf, 0, num); total += num; if (throttler != null) {        //对读入的byte数限流，canceler预留，表示是否跳过限流 throttler.throttle(num, canceler); } } } catch (...) { ... } finally { ... }

JournalNode.startSyncer()--->JournalNodeSyncer.start()    -->startSyncJournalsDaemon()    -->syncJournals()    -->syncWithJournalAtIndex()    -->getMissingLogSegments()    -->downloadMissingLogSegment()--->Util.doGetUrl(...throttler)    -->receiveFile(...throttler)

DataNode.startDataNode()--->BlockPoolManager.refreshNamenodes()   -->doRefreshNamenodes() //创建BPOfferService并启动   -->startAll();--->BPOfferService.start() //维护BPServiceActor列表--->BPServiceActor.start() //启动actor线程，是实际与NN心跳交互的线程    -->connectToNNAndHandshake()--->BPOfferService.verifyAndSetNamespaceInfo()--->DataNode.initBlockPool(this);    -->initStorage(nsInfo); //初始化存储，创建FsDatasetImpl--->FSDatasetImpl.addVolume() //加入磁盘--->FsVolumeList.activateVolume() // 每个盘创建一个VolumeScanner并启动--->BlockScanner.addVolumeScanner() //为每个盘创建一个VolumeScanner并启动--->VolumeScanner.start() //启动线程    -->runLoop()     -->scanBlock() //扫描可疑的block文件--->BlockSender.sendBlock(nullStream, null, throttler)//读取并发送到/dev/null -->doSendBlock() -->sendPacket()