任务并行库 - 自定义任务调度程序

我需要向在线 api 发出 Web 服务请求，我认为并行扩展非常适合我的需求。

所涉及的网络服务被设计为重复调用，但有一种机制，如果每秒的调用次数超过一定数量，就会向您收费。我显然想最大程度地减少费用，因此想知道是否有人见过可以满足以下要求的 TaskScheduler：

1. 限制每个时间跨度计划的任务数量。我想如果请求数量超过这个限制，那么它需要丢弃任务或者可能阻塞？ （停止积压的任务）
2. 检测相同的请求是否已在要执行的调度程序中但尚未执行，如果是，则不将第二个任务排队，而是返回第一个任务。

人们是否认为这些是任务调度程序应该处理的职责，或者我是不是找错了对象？如果您有其他选择，我愿意接受建议。

我同意其他人的观点，即 TPL Dataflow 听起来是一个很好的解决方案。

要限制处理，您可以创建一个TransformBlock这实际上不会以任何方式转换数据，只是如果它在前一个数据之后太早到达，它就会延迟它：

static IPropagatorBlock<T, T> CreateDelayBlock<T>(TimeSpan delay)
{
    DateTime lastItem = DateTime.MinValue;
    return new TransformBlock<T, T>(
        async x =>
                {
                    var waitTime = lastItem + delay - DateTime.UtcNow;
                    if (waitTime > TimeSpan.Zero)
                        await Task.Delay(waitTime);

                    lastItem = DateTime.UtcNow;

                    return x;
                },
        new ExecutionDataflowBlockOptions { BoundedCapacity = 1 });
}

然后创建一个生成数据的方法（例如从 0 开始的整数）：

static async Task Producer(ITargetBlock<int> target)
{
    int i = 0;
    while (await target.SendAsync(i))
        i++;
}

它是异步写入的，因此如果目标块现在无法处理这些项目，它将等待。

然后编写消费者方法：

static void Consumer(int i)
{
    Console.WriteLine(i);
}

最后，将它们链接在一起并启动它：

var delayBlock = CreateDelayBlock<int>(TimeSpan.FromMilliseconds(500));

var consumerBlock = new ActionBlock<int>(
    (Action<int>)Consumer,
    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded });

delayBlock.LinkTo(consumerBlock, new DataflowLinkOptions { PropagateCompletion = true });

Task.WaitAll(Producer(delayBlock), consumerBlock.Completion);

Here, delayBlock每 500 毫秒最多接受一个项目，并且Consumer()方法可以并行运行多次。要完成处理，请致电delayBlock.Complete().

如果你想为 #2 添加一些缓存，你可以创建另一个TransformBlock在那里完成工作并将其链接到其他块。