c# - 听取使用 IEnumerable<T> 时用于 BlockingCollection 的 Queue.Peek() 的模拟

Question

我正在使用 Pipelines pattern将消息消费者与生产者分离的实现，以避免消费者缓慢的问题。

消息处理阶段出现异常[1]它将丢失并且不会被分派(dispatch)到其他服务/层 [2] .我该如何处理 [3] 中的此类问题所以消息不会丢失，重要的是!消息的顺序不会混淆，因此上层服务/层将按照消息进来的顺序获取消息。我有一个想法涉及另一个中间层 Queue但它似乎很复杂？遗憾BlockingCollection<T>不暴露 Queue.Peek() 的任何类似物方法，这样我就可以阅读下一条可用消息，如果处理成功，请执行 Dequeue()

private BlockingCollection<IMessage> messagesQueue;    

// TPL Task does following:
// Listen to new messages and as soon as any comes in - process it
foreach (var cachedMessage in 
             messagesQueue.GetConsumingEnumerable(cancellation))
{    
    const int maxRetries = 3;
    int retriesCounter = 0;
    bool isSent = false;

    // On this point a message already is removed from messagesQueue
    while (!isSent && retriesCounter++ <= maxRetries)
    {
        try
        {
           // [1] Preprocess a message
           // [2] Dispatch to an other service/layer    
           clientProxyCallback.SendMessage(cachedMessage);
           isSent = true;
        }                                
        catch(Exception exception)
        {
           // [3]   
           // logging
           if (!isSent && retriesCounter < maxRetries)
           {
              Thread.Sleep(NSeconds);
           }
        }            
    
        if (!isSent && retriesCounter == maxRetries)
        {
           // just log, message is lost on this stage!
        }
    }
}

编辑:忘了说这是 IIS 托管的 WCF 服务，它通过客户端回调协定将消息分派(dispatch)回 Silverlight 客户端 WCF 代理。

EDIT2: 下面是我将如何使用 Peek() 执行此操作, 我错过了什么吗？

bool successfullySent = true;
try
{
   var item = queue.Peek();
   PreProcessItem(item);
   SendItem(item);       
}
catch(Exception exception)
{
   successfullySent = false;
}
finally
{
   if (successfullySent)
   {
       // just remove already sent item from the queue
       queue.Dequeue();
   }
}

EDIT3: 当然，我可以使用 while 循环、bool 标志、Queue 等旧式方法和 AutoResetEvent , 但我只是想知道是否可以使用 BlockingCollection和 GetConsumingEnumerable()我认为像 Peek 这样的设施将是 与使用可枚举一起使用时非常有用，因为否则所有管道模式实现都会示例新的东西，如 BlockingCollection和 GetConsumingEnumerable()看起来不耐用，我必须回到旧方法。

Answer 1

你应该考虑中间队列。

BlockingCollection<T>由于其性质，不能“窥视”项目 - 可以有多个消费者。其中一个可以偷看一个项目，另一个可以拿它 - 因此，第一个将尝试拿取已经被拿走的项目。