c# – TPL架构问题

// Setup the blocking collection somewhere when your process starts up (OnStart for a Windows service)
BlockingCollection<string> idsToProcess = new BlockingCollection<string>();

从那里我将使用BlockingCollection::GetConsumingEnumerable从BlockingCollection::GetConsumingEnumerable返回的枚举.在ForEach调用中,您将设置ParallelOptions::MaxDegreeOfParallelism在ForEach的主体内,您将执行存储过程.

现在,一旦存储过程执行完成,您就说您不想重新安排执行至少两秒钟.没问题,安排一个带有回调的System.Threading.Timer,它只会将ID添加回提供的回调中的BlockingCollection.

Parallel.ForEach(
    idsToProcess.GetConsumingEnumerable(),new ParallelOptions 
    { 
        MaxDegreeOfParallelism = 4 // read this from config
    },(id) =>
    {
       // ... execute sproc ...

       // Need to declare/assign this before the delegate so that we can dispose of it inside 
       Timer timer = null;

       timer = new Timer(
           _ =>
           {
               // Add the id back to the collection so it will be processed again
               idsToProcess.Add(id);

               // Cleanup the timer
               timer.Dispose();
           },null,// no state,id wee need is "captured" in the anonymous delegate
           2000,// probably should read this from config
           Timeout.Infinite);
    }

最后,当进程关闭时,您将调用BlockingCollection::CompleteAdding,以便使用停止阻塞和完成处理可枚举,并且Parallel :: ForEach将退出.例如,如果这是Windows服务,您将在OnStop执行此操作.

// When ready to shutdown you just signal you're done adding
idsToProcess.CompleteAdding();

更新

您在评论中提出了一个有效的问题,即您可能在任何给定点处理大量ID,并担心每个ID的计时器会产生过多的开销.我绝对同意这一点.因此,在您同时处理大量ID的情况下,我将从使用每个ID的定时器更改为使用另一个队列来保存由单个短间隔计时器监视的“休眠”ID.首先,你需要一个ConcurrentQueue来放置睡着的ID：

ConcurrentQueue<Tuple<string,DateTime>> sleepingIds = new ConcurrentQueue<Tuple<string,DateTime>>();

现在,我在这里使用两部分Tuple进行说明,但您可能希望为它创建一个更强类型的结构(或者至少使用using语句将其替换)以提高可读性.元组具有id和DateTime,表示它何时被放入队列.

现在,您还需要设置将监视此队列的计时器：

Timer wakeSleepingIdsTimer = new Timer(
   _ =>
   {
       DateTime utcNow = DateTime.UtcNow;

       // Pull all items from the sleeping queue that have been there for at least 2 seconds
       foreach(string id in sleepingIds.TakeWhile(entry => (utcNow - entry.Item2).TotalSeconds >= 2))
       {
           // Add this id back to the processing queue
           idsToProcess.Enqueue(id);
       }
   },// no state
   Timeout.Infinite,// no due time
   100 // wake up every 100ms,probably should read this from config
 );

然后你只需要改变Parallel :: ForEach来执行以下操作,而不是为每个设置一个计时器：

(id) =>
{
       // ... execute sproc ...

       sleepingIds.Enqueue(Tuple.Create(id,DateTime.UtcNow)); 
}