`
liudunxu2
  • 浏览: 30641 次
  • 性别: Icon_minigender_1
  • 来自: 青岛
文章分类
社区版块
存档分类
最新评论

hdfs中限流器BlockTransferThrottler的c#改写

 
阅读更多
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;
        }
    }
}

分享到:
评论

相关推荐

    实验2常用的HDFS操作.doc

    (1)向HDFS中上传任意文本文件,如果指定的文件在HDFS中已经存在,则由用户来指定是追加到原有文件末尾还是覆盖原有的文件 (2)从HDFS中下载指定文件,如果本地文件与要下载的文件名称相同,则自动对下载的文件...

    7、NIFI综合应用场景-将mysql查询出的json数据转换成txt后存储至HDFS中

    7、NIFI综合应用场景-将mysql查询出的json数据转换成txt后存储至HDFS中 网址:https://blog.csdn.net/chenwewi520feng/article/details/130620392 本文是在6、NIFI综合应用场景-离线同步Mysql数据到HDFS中基础上完成...

    HDFS文件的查看

    hdfs文件的查看 hdfs fs -cat /文件名

    HDFS Comics HDFS 漫画

    HDFS是Hadoop分布式计算的存储基础。HDFS具有高容错性,可以部署在通用硬件设备上,适合数据密集型应用,并且提供对数据读写的高吞 吐量。HDFS能 够提供对数据的可扩展访问,通过简单地往集群里添加节点就可以解决...

    动手实验 Hadoop Oracle HDFS直接连接器

    动手实验: Oracle HDFS直接连接器 软件环境: 本实验主要由以下软件搭建而成: Oracle Enterprise Linux, Oracle 11g, Java SE6pdate30, Apache Hadoop, Oracle Connector for Hadoop等。 实验 1: 直接访问单个 ...

    论文研究-一种面向HDFS中海量小文件的存取优化方法.pdf

    从HDFS中读取数据时,根据文件之间的相关性,对接下来用户最有可能访问的文件进行预取,减少了客户端对NameNode节点的访问次数,提高了文件命中率和处理速度。实验结果证明,该方法有效提升了Hadoop对小文件的存取...

    HDFS本地查看器

    自己实现的一个简单的HDFS查看器源码。 运行方式:将2个class文件拷贝到namenode的hadoop bin目录下,然后运行 JAVA Main 就可以打开该查看器。 注意事项: hadoop的dfs服务必须启动 必须放在hadoop的bin目录下运行...

    hdfs官方设计文档(中文)

    Hadoop 分布式文件系统 (HDFS)是一个设计为用在普通硬件设备上的分布式文件系统。它与现有的分布式文件系统有很多近似的地方,但又和这些文件系统有很明显的不同。HDFS是高容错的,设计为部署在廉价硬件上的。HDFS对...

    HDFS管理工具HDFS Explorer下载地址、使用方法.docx

    windows平台下的HDFS文件浏览器,就像windows管理器一样管理你的hdfs文件系统。现在官网已经停止更新这款软件。具体配置如下: HDFS配置页面及端口http://master:50070 配置HDFS服务器 配置WebHDFS HDFS Explorer...

    Hadoop源码分析HDFS数据流

    hdfs上传文件过程源码分析

    Hadoop框架之HDFS的shell操作

    Hadoop框架之HDFS的shell操作Hadoop框架之HDFS的shell操作Hadoop框架之HDFS的shell操作Hadoop框架之HDFS的shell操作

    java操作Hbase之从Hbase中读取数据写入hdfs中源码

    java操作Hbase之从Hbase中读取数据写入hdfs中源码,附带全部所需jar包,欢迎下载学习。

    HDFS分布式文件系统(1).pdf

    HDFS是基于流数据模式访问和处理超大文件的需求而开发的,它可以运行于廉价的商用服务器上。总的来说,可以将 HDFS的主要特点概括为以下几点:(1 )处理超大文件 (2 )流式地访问数据 (3 )运行于廉价的商用...

    hadoop-hdfs-2.9.1-API文档-中文版.zip

    赠送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 对应...

    实验二:熟悉常用的HDFS操作

    (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

    上传文件到HDFS的java代码实现。已经测试了,可以直接运行。

    hadoop hdfs配置文件详解

    hadoop的 hdfs配置文件 对文件的每项property进行中文描述 希望的家支持 过几天还会献上其他

    hdfs源码.zip

    第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 ...

    10_尚硅谷大数据之HDFS的数据流1

    4.1 HDFS 写数据流程 4.1.1 剖析文件写入 4.1.2 网络拓扑概念 4.1.3 机架感知(副本节点选择) 4.2 HDFS 读数据流程

    hadoop-hdfs-2.7.3-API文档-中文版.zip

    赠送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...

Global site tag (gtag.js) - Google Analytics