DataTransferThrottler类别Datanode读取和写入数据时控制传输数据速率。这个类是线程安全的，它可以由多个线程共享。
用途是构建DataTransferThrottler对象，并设置期限period和带宽bandwidthPerSec，际读写前调用DataTransferThrottler.throttle()方法。假设I/O的速率相对给定的带宽太快，则该方法会将当前线程wait。
两个构造函数
- 双參构造函数，能够设置周期period和带宽bandwidthPerSec。
public DataTransferThrottler( long period, long bandwidthPerSec)
- 单參构造函数，能够设置带宽bandwidthPerSec, 周期period默认被设置为500ms。
public DataTransferThrottler( long bandwidthPerSec)
重要属性
| period | 周期，单位毫秒 |
| periodExtension | 周期扩展时间。单位毫秒 |
| bytesPerPeriod | 一个周期内能够发送/接收的byte总数 |
| curPeriodStart | 当前周期開始时间，单位毫秒 |
| curReserve | 当前周期内还能够发送/接收的byte数 |
| bytesAlreadyUsed | 当前周期内已经使用的byte数 |
带宽bandWidthPerPerSec = bytesPerPeriod * period/1000
DataTransferThrottler.throttle()方法
该方法有两个入參：
- numOfBytes是自从上次throttle被调用到如今，须要发送/接收的byte数；
- canceler是一个外部取消器，能够用来检測是否取消流量控制。
DataTransferThrottler.throttle()方法会先将curReserve减去numOfBytes。接着运行例如以下逻辑。
- 假设curReserve大于0，则说明当前周期还有余量，throttle方法直接返回。
- 假设curReserve小于等于0，则说明当前周期已经没有余量，会运行以下逻辑。
>假设当前时间now小于当前周期结束时间curPeriodEnd，则wait到下一个周期
>假设当前时间now大于当前周期结束时间curPeriodEnd而且小于curPeriodStart+periodExtension。说明已经进入下一个周期而且throttle应该不是非常长时间没有使用，则将curReserve加上下一个周期能够传输的byte总数bytesPerPeriod，并将curPeriodStart设置到下一个周期的開始。
>假设当前时间now大于curPeriodStart+periodExtension。则可能Throttler非常长时间没有使用。则抛弃上一个周期。
DataTransferThrottler.throttle()方法源代码例如以下：
public synchronized void throttle(long numOfBytes , Canceler canceler) { if ( numOfBytes <= 0 ) { return; } //当前周期余量减去须要发送/接收的byte数numOfBytes curReserve -= numOfBytes ; bytesAlreadyUsed += numOfBytes; //假设curReserve小于等于0，则说明当前周期已经没有余量 while ( curReserve <= 0) { //假设传入了有效取消器canceler，而且取消器的取消状态isCancelled是true，则直接退出while循环 if (canceler != null && canceler.isCancelled()) { return; } long now = monotonicNow(); //计算当前周期结束时间。并存放在curPeriodEnd变量中 long curPeriodEnd = curPeriodStart + period ; if ( now < curPeriodEnd ) { //等待下一个周期，这样curReserve就能够添加 try { wait( curPeriodEnd - now ); } catch (InterruptedException e) { //终止throttle, 而且重置interrupted状态来确保在调用栈中其他interrupt处理器能够正确运行 Thread. currentThread().interrupt(); break; } } //假设当前时间now比当前结束时间curPeriodEnd晚，而且小于curPeriodStart+periodExtension(周期3倍时间)，则进入下一个周期 //并添加bytesPerPeriod到curReserve else if ( now < (curPeriodStart + periodExtension)) { curPeriodStart = curPeriodEnd; curReserve += bytesPerPeriod ; } //假设当前时间now大于curPeriodStart+periodExtension，则可能Throttler非常长时间没有使用。抛弃上一个周期 else { curPeriodStart = now; curReserve = bytesPerPeriod - bytesAlreadyUsed; } } bytesAlreadyUsed -= numOfBytes; }
总结
DataTransferThrottler类能够控制传输数据在一段时期内的平均速率。但可能在一个周期结束时瞬时速率会失控。
參考资料
hadoop-release-2.5.0源代码
转载请附上原博客地址：http://blog.csdn.net/jeff_fangji/article/details/44258827
版权声明：本文博客原创文章，博客，未经同意，不得转载。