使用 Parallel.For 时的限制

Question

当使用单线程循环时，我可以轻松地通过将线程置于睡眠状态（即 Thread.Sleep(1000/MessagesPerSecond)）来限制每秒发送的消息，这很容易......但现在我已经扩展到并行线程，这不再正常工作。

有人建议如何限制使用并行线程时发送的消息吗？

Parallel.For(0, NumberOfMessages, delegate(int i) {

   // Code here

   if (MessagesPerSecond != 0)
      Thread.Sleep(1000/MessagesPerSecond);
});

Answer 1

使用 AutoResetEvent 和计时器。每当计时器触发时，就让它设置 AutoResetEvent。

然后让您的进程在发送之前立即在 AutoResetEvent 上发送消息 WaitOne。

    private static readonly AutoResetEvent _Next = new AutoResetEvent(true);
    private static Timer _NextTimer;

    private static void SendMessages(IEnumerable<Message> messages)
    {
        if (_NextTimer == null)
            InitializeTimer();

        Parallel.ForEach(
            messages,
            m =>
            {
                _Next.WaitOne();
                // Do something
            }
            );
    }

    private static void SetNext(object state)
    {
        _Next.Set();
    }

Answer 2

您可以考虑使用共享的 ConcurrentQueue，您的并行循环将填充它与准备好的消息。使用 System.Threading.Timer 拉取消息按照您想要的时间间隔从队列中取出并发送它们。请注意，只有当创建要发送的消息成本很高时，这种设计才有意义。如果消息的实际发送是昂贵的部分，则没有理由并行运行循环。

如果您需要在发送消息后停止计时器，则必须执行一些额外的工作，但这种设计非常适合必须处理异步消息队列的受限制的消息发送方。另一个需要考虑的边界情况是“消息堆积”，即消息排队的速度比处理消息的速度快。您可能需要考虑在这种情况下生成错误（因为它可能表示错误）或使用 阻止集合。