使用 Parallel.ForEach 限制每秒启动的 API 请求数

Question

我正在努力改进我的一些代码以提高效率。在原始代码中，我将允许的线程数限制为 5，如果我已经有 5 个活动线程，我会等到一个完成后再开始另一个。现在我想修改此代码以允许任意数量的线程，但我希望能够确保每秒只有 5 个线程启动。例如：

• 第二个 0 - 5 个新线程
• 第二个1-5个新线程
• 第二个 2 - 5 个新线程 ...

原始代码（cleanseDictionary 通常包含数千个项目）：

        ConcurrentDictionary<long, APIResponse> cleanseDictionary = new ConcurrentDictionary<long, APIResponse>();
        ConcurrentBag<int> itemsinsec = new ConcurrentBag<int>();
        ConcurrentDictionary<long, string> resourceDictionary = new ConcurrentDictionary<long, string>();
        DateTime start = DateTime.Now;

        Parallel.ForEach(resourceDictionary, new ParallelOptions { MaxDegreeOfParallelism = 5 }, row =>
        {
            lock (itemsinsec)
            {
                ThrottleAPIRequests(itemsinsec, start);

                itemsinsec.Add(1);
            }

            cleanseDictionary.TryAdd(row.Key, _helper.MakeAPIRequest(string.Format("/endpoint?{0}", row.Value)));
        });


    private static void ThrottleAPIRequests(ConcurrentBag<int> itemsinsec, DateTime start)
    {
        if ((start - DateTime.Now).Milliseconds < 10001 && itemsinsec.Count > 4)
        {
            System.Threading.Thread.Sleep(1000 - (start - DateTime.Now).Milliseconds);
            start = DateTime.Now;
            itemsinsec = new ConcurrentBag<int>();
        }
    }

我的第一个想法是将MaxDegreeofParallelism 增加到更高的值，然后有一个辅助方法，该方法将在一秒钟内仅限制 5 个线程，但我不确定这是否是最好的方法，是否是，我可能需要一个lock 在这一步附近？

提前致谢！

编辑 我实际上正在寻找一种限制 API 请求而不是实际线程的方法。我以为他们是同一个。

编辑 2：我的要求是每秒发送超过 5 个 API 请求

Answer 1

MS 网站上的“Parallel.ForEach”

可以并行运行

如果您想对线程的管理方式进行任何程度的精细控制，这不是办法。
如何创建自己的帮助类，您可以在其中使用组 id 对作业进行排队，允许您等待组 id X 的所有作业完成，并在需要时产生额外的线程？

Answer 2

对我来说最好的解决方案是：

using System;
using System.Collections.Concurrent;
using System.Threading.Tasks;

namespace SomeNamespace
{
    public class RequestLimiter : IRequestLimiter
    {
        private readonly ConcurrentQueue<DateTime> _requestTimes;
        private readonly TimeSpan _timeSpan;

        private readonly object _locker = new object();

        public RequestLimiter()
        {
            _timeSpan = TimeSpan.FromSeconds(1);
            _requestTimes = new ConcurrentQueue<DateTime>();
        }

        public TResult Run<TResult>(int requestsOnSecond, Func<TResult> function)
        {
            WaitUntilRequestCanBeMade(requestsOnSecond).Wait();
            return function();
        }

        private Task WaitUntilRequestCanBeMade(int requestsOnSecond)
        {
            return Task.Factory.StartNew(() =>
            {
                while (!TryEnqueueRequest(requestsOnSecond).Result) ;
            });
        }

        private Task SynchronizeQueue()
        {
            return Task.Factory.StartNew(() =>
            {
                _requestTimes.TryPeek(out var first);

                while (_requestTimes.Count > 0 && (first.Add(_timeSpan) < DateTime.UtcNow))
                    _requestTimes.TryDequeue(out _);
            });
        }

        private Task<bool> TryEnqueueRequest(int requestsOnSecond)
        {
            lock (_locker)
            {
                SynchronizeQueue().Wait();
                if (_requestTimes.Count < requestsOnSecond)
                {
                    _requestTimes.Enqueue(DateTime.UtcNow);
                    return Task.FromResult(true);
                }
                return Task.FromResult(false);
            }
        }
    }
}

Answer 3

我希望能够每秒发送超过 5 个 API 请求

这真的很简单：

while (true) {
 await Task.Delay(TimeSpan.FromSeconds(1));
 await Task.WhenAll(Enumerable.Range(0, 5).Select(_ => RunRequestAsync()));
}

也许不是最好的方法，因为会有大量请求。这不是连续的。

此外，还有时间偏差。一次迭代需要超过 1 秒。这可以通过几行时间逻辑来解决。