using System;
using System.Runtime.CompilerServices;
using System.Threading;
namespace Common.Engine
{
/// <summary>
/// 限流器
/// 改写自:hadoop\src\hdfs\org\apache\hadoop\hdfs\server\datanode\BlockTransferThrottler.java
/// </summary>
public class Throttler
{
private readonly int _period; // period over which bw is imposed
private readonly int _periodExtension; // Max period over which bw accumulates.
private int _bytesPerPeriod; // total number of bytes can be sent in each period
private int _curPeriodStart; // current period starting time
private int _curReserve; // remaining bytes can be sent in the period
private int _bytesAlreadyUsed;
/** Constructor
* @param bandwidthPerSec bandwidth allowed in bytes per second.
*/
public Throttler(int bandwidthPerSec): this(500, bandwidthPerSec)
{ // by default throttling period is 500ms
}
/**
* Constructor
* @param period in milliseconds. Bandwidth is enforced over this
* period.
* @param bandwidthPerSec bandwidth allowed in bytes per second.
*/
public Throttler(int period, int bandwidthPerSec)
{
this._curPeriodStart = CurrentTimeMillis();
this._period = period;
this._curReserve = this._bytesPerPeriod = bandwidthPerSec * period / 1000;
this._periodExtension = period * 3;
}
/// <summary>
/// 当前的时间戳
/// </summary>
/// <returns></returns>
[MethodImpl(MethodImplOptions.Synchronized)]
public int CurrentTimeMillis()
{
return Environment.TickCount;
}
/**
* @return current throttle bandwidth in bytes per second.
*/
[MethodImpl(MethodImplOptions.Synchronized)]
public long GetBandwidth()
{
return _bytesPerPeriod * 1000 / _period;
}
/**
* Sets throttle bandwidth. This takes affect latest by the end of current
* period.
*
* @param bytesPerSecond
*/
[MethodImpl(MethodImplOptions.Synchronized)]
public void SetBandwidth(int bytesPerSecond)
{
if (bytesPerSecond <= 0)
{
throw new Exception("" + bytesPerSecond);
}
_bytesPerPeriod = bytesPerSecond * _period / 1000;
}
/** Given the numOfBytes sent/received since last time throttle was called,
* make the current thread sleep if I/O rate is too fast
* compared to the given bandwidth.
*
* @param numOfBytes
* number of bytes sent/received since last time throttle was called
*/
[MethodImpl(MethodImplOptions.Synchronized)]
public void Throttle(int numOfBytes)
{
if (numOfBytes <= 0)
{
return;
}
_curReserve -= numOfBytes;
_bytesAlreadyUsed += numOfBytes;
while (_curReserve <= 0)
{
int now = CurrentTimeMillis();
int curPeriodEnd = _curPeriodStart + _period;
if (now < curPeriodEnd)
{
// Wait for next period so that curReserve can be increased.
try
{
Thread.Sleep(curPeriodEnd - now);
}
catch (Exception) { }
}
else if (now < (_curPeriodStart + _periodExtension))
{
_curPeriodStart = curPeriodEnd;
_curReserve += _bytesPerPeriod;
}
else
{
// discard the prev period. Throttler might not have
// been used for a long time.
_curPeriodStart = now;
_curReserve = _bytesPerPeriod - _bytesAlreadyUsed;
}
}
_bytesAlreadyUsed -= numOfBytes;
}
}
}
分享到:
相关推荐
(1)向HDFS中上传任意文本文件,如果指定的文件在HDFS中已经存在,则由用户来指定是追加到原有文件末尾还是覆盖原有的文件 (2)从HDFS中下载指定文件,如果本地文件与要下载的文件名称相同,则自动对下载的文件...
7、NIFI综合应用场景-将mysql查询出的json数据转换成txt后存储至HDFS中 网址:https://blog.csdn.net/chenwewi520feng/article/details/130620392 本文是在6、NIFI综合应用场景-离线同步Mysql数据到HDFS中基础上完成...
hdfs文件的查看 hdfs fs -cat /文件名
HDFS是Hadoop分布式计算的存储基础。HDFS具有高容错性,可以部署在通用硬件设备上,适合数据密集型应用,并且提供对数据读写的高吞 吐量。HDFS能 够提供对数据的可扩展访问,通过简单地往集群里添加节点就可以解决...
动手实验: Oracle HDFS直接连接器 软件环境: 本实验主要由以下软件搭建而成: Oracle Enterprise Linux, Oracle 11g, Java SE6pdate30, Apache Hadoop, Oracle Connector for Hadoop等。 实验 1: 直接访问单个 ...
从HDFS中读取数据时,根据文件之间的相关性,对接下来用户最有可能访问的文件进行预取,减少了客户端对NameNode节点的访问次数,提高了文件命中率和处理速度。实验结果证明,该方法有效提升了Hadoop对小文件的存取...
自己实现的一个简单的HDFS查看器源码。 运行方式:将2个class文件拷贝到namenode的hadoop bin目录下,然后运行 JAVA Main 就可以打开该查看器。 注意事项: hadoop的dfs服务必须启动 必须放在hadoop的bin目录下运行...
Hadoop 分布式文件系统 (HDFS)是一个设计为用在普通硬件设备上的分布式文件系统。它与现有的分布式文件系统有很多近似的地方,但又和这些文件系统有很明显的不同。HDFS是高容错的,设计为部署在廉价硬件上的。HDFS对...
windows平台下的HDFS文件浏览器,就像windows管理器一样管理你的hdfs文件系统。现在官网已经停止更新这款软件。具体配置如下: HDFS配置页面及端口http://master:50070 配置HDFS服务器 配置WebHDFS HDFS Explorer...
hdfs上传文件过程源码分析
Hadoop框架之HDFS的shell操作Hadoop框架之HDFS的shell操作Hadoop框架之HDFS的shell操作Hadoop框架之HDFS的shell操作
java操作Hbase之从Hbase中读取数据写入hdfs中源码,附带全部所需jar包,欢迎下载学习。
HDFS是基于流数据模式访问和处理超大文件的需求而开发的,它可以运行于廉价的商用服务器上。总的来说,可以将 HDFS的主要特点概括为以下几点:(1 )处理超大文件 (2 )流式地访问数据 (3 )运行于廉价的商用...
赠送jar包:hadoop-hdfs-2.9.1.jar 赠送原API文档:hadoop-hdfs-2.9.1-javadoc.jar 赠送源代码:hadoop-hdfs-2.9.1-sources.jar 包含翻译后的API文档:hadoop-hdfs-2.9.1-javadoc-API文档-中文(简体)版.zip 对应...
(1)理解 HDFS在Hadoop体系结构中的角色。(2)熟练使用HDFS操作常用的 Shell命令。(3)熟悉HDFS操作常用的Java API。 A.2.2 实验平台 (1)操作系统:Linux(建议Ubuntu 16.04)。(2) Hadoop版本:2.7.1。 (3)JDK版本:1.7...
上传文件到HDFS的java代码实现。已经测试了,可以直接运行。
hadoop的 hdfs配置文件 对文件的每项property进行中文描述 希望的家支持 过几天还会献上其他
第1章 HDFS 1 1.1 HDFS概述 1 1.1.1 HDFS体系结构 1 1.1.2 HDFS基本概念 2 1.2 HDFS通信协议 4 1.2.1 Hadoop RPC接口 4 1.2.2 流式接口 20 1.3 HDFS主要流程 22 1.3.1 HDFS客户端读流程 22 1.3.2 ...
4.1 HDFS 写数据流程 4.1.1 剖析文件写入 4.1.2 网络拓扑概念 4.1.3 机架感知(副本节点选择) 4.2 HDFS 读数据流程
赠送jar包:hadoop-hdfs-2.7.3.jar; 赠送原API文档:hadoop-hdfs-2.7.3-javadoc.jar; 赠送源代码:hadoop-hdfs-2.7.3-sources.jar; 赠送Maven依赖信息文件:hadoop-hdfs-2.7.3.pom; 包含翻译后的API文档:hadoop...