我正在开发一个在后台持续发送数据流的程序，我希望允许用户设置上传和下载限制的上限。

我已经阅读了 token bucket和 leaky bucket算法，并且看起来后者似乎符合描述，因为这不是最大化网络带宽的问题，而是尽可能不引人注目的问题。

不过，我有点不确定我将如何实现它。一种自然的方法是扩展抽象 Stream 类，以简化现有流量的扩展，但这是否需要额外线程的参与来发送数据，同时接收数据(漏桶)？任何有关执行相同操作的其他实现的提示都将不胜感激。

此外，虽然我可以修改程序接收的数据量，但带宽限制在 C# 级别的效果如何？计算机是否仍会接收数据并简单地保存它，从而有效地取消 throttle 效果，或者它会等到我要求接收更多数据时再接收吗？

编辑:我对限制传入和传出数据感兴趣，我无法控制流的另一端。

最佳答案

基于@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();
}

关于c# - C# 中的带宽限制，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/371032/