我正在开发一个程序，它不断地在后台发送数据流，我希望允许用户为上传和下载限制设置上限。C＃中的带宽限制＃
我对token bucket和leaky bucket alghorhithms念起来，貌似后者似乎适合的描述，因为这不是最大化网络带宽，而是被尽可能不引人注目的问题。
但是，我有点不确定我会如何实现这一点。一种自然的方法是扩展抽象的Stream类，以便扩展现有的流量，但是这会不会需要额外的线程来发送数据，同时接收（泄漏的桶）呢？任何暗示其他实现相同的提示将不胜感激。
此外，尽管我可以修改程序接收的数据量，但是C＃级别的带宽限制工作有多好？计算机是否仍会收到数据并简单地保存数据，从而有效地取消节流效果，还是等到我要求接收更多数据？
编辑：我对节流传入和传出数据，其中I具有通过该流的相对端没有控制权。
请参阅this article about ThrottledStream class.它应该适合您的需求。
我想出了arul提到的ThrottledStream类的不同实现。我的版本使用的WaitHandle和定时器用1秒间隔：
public ThrottledStream(Stream parentStream, int maxBytesPerSecond=int.MaxValue)
{
MaxBytesPerSecond = maxBytesPerSecond;
parent = parentStream;
processed = 0;
resettimer = new System.Timers.Timer();
resettimer.Interval = 1000;
resettimer.Elapsed += resettimer_Elapsed;
resettimer.Start();
}
protected void Throttle(int bytes)
{
try
{
processed += bytes;
if (processed >= maxBytesPerSecond)
wh.WaitOne();
}
catch
{
}
}
private void resettimer_Elapsed(object sender, ElapsedEventArgs e)
{
processed = 0;
wh.Set();
}
每当带宽限制超过线程将休眠，直到下一秒开始。无需计算最佳睡眠时间。
全面实施：
public class ThrottledStream : Stream
{
#region Properties
private int maxBytesPerSecond;
/// <summary>
/// Number of Bytes that are allowed per second
/// </summary>
public int MaxBytesPerSecond
{
get { return maxBytesPerSecond; }
set
{
if (value < 1)
throw new ArgumentException("MaxBytesPerSecond has to be >0");
maxBytesPerSecond = value;
}
}
#endregion
#region Private Members
private int processed;
System.Timers.Timer resettimer;
AutoResetEvent wh = new AutoResetEvent(true);
private Stream parent;
#endregion
/// <summary>
/// Creates a new Stream with Databandwith cap
/// </summary>
/// <param name="parentStream"></param>
/// <param name="maxBytesPerSecond"></param>
public ThrottledStream(Stream parentStream, int maxBytesPerSecond=int.MaxValue)
{
MaxBytesPerSecond = maxBytesPerSecond;
parent = parentStream;
processed = 0;
resettimer = new System.Timers.Timer();
resettimer.Interval = 1000;
resettimer.Elapsed += resettimer_Elapsed;
resettimer.Start();
}
protected void Throttle(int bytes)
{
try
{
processed += bytes;
if (processed >= maxBytesPerSecond)
wh.WaitOne();
}
catch
{
}
}
private void resettimer_Elapsed(object sender, ElapsedEventArgs e)
{
processed = 0;
wh.Set();
}
#region Stream-Overrides
public override void Close()
{
resettimer.Stop();
resettimer.Close();
base.Close();
}
protected override void Dispose(bool disposing)
{
resettimer.Dispose();
base.Dispose(disposing);
}
public override bool CanRead
{
get { return parent.CanRead; }
}
public override bool CanSeek
{
get { return parent.CanSeek; }
}
public override bool CanWrite
{
get { return parent.CanWrite; }
}
public override void Flush()
{
parent.Flush();
}
public override long Length
{
get { return parent.Length; }
}
public override long Position
{
get
{
return parent.Position;
}
set
{
parent.Position = value;
}
}
public override int Read(byte[] buffer, int offset, int count)
{
Throttle(count);
return parent.Read(buffer, offset, count);
}
public override long Seek(long offset, SeekOrigin origin)
{
return parent.Seek(offset, origin);
}
public override void SetLength(long value)
{
parent.SetLength(value);
}
public override void Write(byte[] buffer, int offset, int count)
{
Throttle(count);
parent.Write(buffer, offset, count);
}
#endregion
}
基于@ 0xDEADBEEF的解决方案，我创建了以下（可测试）解决方案基础上的Rx调度：
public class ThrottledStream : Stream
{
private readonly Stream parent;
private readonly int maxBytesPerSecond;
private readonly IScheduler scheduler;
private readonly IStopwatch stopwatch;
private long processed;
public ThrottledStream(Stream parent, int maxBytesPerSecond, IScheduler scheduler)
{
this.maxBytesPerSecond = maxBytesPerSecond;
this.parent = parent;
this.scheduler = scheduler;
stopwatch = scheduler.StartStopwatch();
processed = 0;
}
public ThrottledStream(Stream parent, int maxBytesPerSecond)
: this (parent, maxBytesPerSecond, Scheduler.Immediate)
{
}
protected void Throttle(int bytes)
{
processed += bytes;
var targetTime = TimeSpan.FromSeconds((double)processed/maxBytesPerSecond);
var actualTime = stopwatch.Elapsed;
var sleep = targetTime - actualTime;
if (sleep > TimeSpan.Zero)
{
using (var waitHandle = new AutoResetEvent(initialState: false))
{
scheduler.Sleep(sleep).GetAwaiter().OnCompleted(() => waitHandle.Set());
waitHandle.WaitOne();
}
}
}
public override bool CanRead
{
get { return parent.CanRead; }
}
public override bool CanSeek
{
get { return parent.CanSeek; }
}
public override bool CanWrite
{
get { return parent.CanWrite; }
}
public override void Flush()
{
parent.Flush();
}
public override long Length
{
get { return parent.Length; }
}
public override long Position
{
get
{
return parent.Position;
}
set
{
parent.Position = value;
}
}
public override int Read(byte[] buffer, int offset, int count)
{
var read = parent.Read(buffer, offset, count);
Throttle(read);
return read;
}
public override long Seek(long offset, SeekOrigin origin)
{
return parent.Seek(offset, origin);
}
public override void SetLength(long value)
{
parent.SetLength(value);
}
public override void Write(byte[] buffer, int offset, int count)
{
Throttle(count);
parent.Write(buffer, offset, count);
}
}
和一些测试，只需要几毫秒：
[TestMethod]
public void ShouldThrottleReading()
{
var content = Enumerable
.Range(0, 1024 * 1024)
.Select(_ => (byte)'a')
.ToArray();
var scheduler = new TestScheduler();
var source = new ThrottledStream(new MemoryStream(content), content.Length/8, scheduler);
var target = new MemoryStream();
var t = source.CopyToAsync(target);
t.Wait(10).Should().BeFalse();
scheduler.AdvanceTo(TimeSpan.FromSeconds(4).Ticks);
t.Wait(10).Should().BeFalse();
scheduler.AdvanceTo(TimeSpan.FromSeconds(8).Ticks - 1);
t.Wait(10).Should().BeFalse();
scheduler.AdvanceTo(TimeSpan.FromSeconds(8).Ticks);
t.Wait(10).Should().BeTrue();
}
[TestMethod]
public void ShouldThrottleWriting()
{
var content = Enumerable
.Range(0, 1024 * 1024)
.Select(_ => (byte)'a')
.ToArray();
var scheduler = new TestScheduler();
var source = new MemoryStream(content);
var target = new ThrottledStream(new MemoryStream(), content.Length/8, scheduler);
var t = source.CopyToAsync(target);
t.Wait(10).Should().BeFalse();
scheduler.AdvanceTo(TimeSpan.FromSeconds(4).Ticks);
t.Wait(10).Should().BeFalse();
scheduler.AdvanceTo(TimeSpan.FromSeconds(8).Ticks - 1);
t.Wait(10).Should().BeFalse();
scheduler.AdvanceTo(TimeSpan.FromSeconds(8).Ticks);
t.Wait(10).Should().BeTrue();
}
如果当计时器滴答你不设置`processed`到`0`将变得更加准确，但从中减去`maxBytesPerSecond`。 – 2017-02-09 06:25:27