c# - 如何在给定时间处理(动态添加的)项目？

Question

我有一个 (concurrent) priority queue 以时间戳(将来)作为键，以到达时间时应调用的函数(/应处理的项目)作为值。我不想为每个项目附加一个计时器，因为有很多。我宁愿使用调度程序线程/任务。

这样做的好策略是什么？

---

一个线程运行一个调度器...(后面是伪代码)

// scheduler
readonly object _threadLock = new object();
while (true)
{
    if(queue.Empty) 
    {
        Monitor.Wait(_threadLock);
    } 
    else 
    {
        var time = GetWaitingTimeForNextElement();

        if(time > 0)
          Monitor.Wait(_threadLock, time);
        else
          // dequeue and process element
    }  
}

...并在添加元素时发出脉冲(添加到空队列或添加新的第一个元素)？

// element enqueued 
Monitor.Pulse(_threadLock);

---

或者使用 Task.Delay(int, CancellationToken ) 以某种方式链接(Task.ContinueWith(...))任务？这将需要一些逻辑来在 新的第一个 元素入队时中止等待，或者在没有人运行时创建新任务。感觉好像有一个我现在没有得到的更简单的解决方案。 :)

---

或者使用计时器(非常伪代码，只是为了理解这个想法)...

System.Timers.Timer x = new System.Timers.Timer().Start();

x.Elapsed += (sender, args) =>
{
    // dequeue and process item(s)

    x.Interval = GetWaitingTimeForNextElement(); // does this reset the timer anyway?
}

...并在添加元素时更新间隔(如上)。

// element enqueued 
x.Interval = updatedTime;

---

我还关心等待方法/计时器的精度:毫秒相当粗糙(尽管它可能有效)有没有更好的选择？

因此...

这又是一堆问题/想法 - 对此感到抱歉 - 但有太多的选择和担忧，很难得到一个概述。总结一下:为动态传入项目实现(精确)时间调度系统的最佳方法是什么？。

我感谢所有提示和答案!非常感谢。

Answer 1

我建议这样做:

1. 创建一个名为 TimedItemsConcurrentPriorityQueue<TKey, TValue> 的类继承自 ConcurrentPriorityQueue<TKey, TValue> .

2. 在您的 TimedItemsConcurrentPriorityQueue<TKey, TValue> 中实现一个名为 ItemReady 的事件每当项目根据时间戳准备好(用于处理)时被触发的类。您可以使用单个计时器并根据需要通过隐藏 Enqueue 来更新计时器, Insert , Remove和其他需要的方法(或者通过修改 ConcurrentPriorityQueue<TKey, TValue> 的源代码并使这些方法成为虚拟的，以便您可以覆盖它们)。

3. 实例化 TimedItemsConcurrentPriorityQueue<TKey, TValue> 的单个实例，我们称该变量为 itemsWaitingToBecomeReady。

4. 实例化 BlockingCollection<T> 的单个对象，我们称该变量为 itemsReady。使用 constructor that takes an IProducerConsumerCollection<T> 并传递给它一个 ConcurrentPriorityQueue<TKey, TValue> 的新实例(继承IProducerConsumerCollection<KeyValuePair<TKey,TValue>>)

5. 每当在 itemsWaitingToBecomeReady 中触发事件 ItemReady 时，您就会将该项目从队列中取出并将其排入 itemsReady。

6. 使用 BlockingCollection<T>.GetConsumingEnumerable 处理 itemsReady 中的项目使用这样的新任务的方法:

.

Task.Factory.StartNew(() =>
{
  foreach (var item in itemsReady.GetConsumingEnumerable())
  {
    ...
  }
}