BlockingCollection - 高同步问题

Question

将消息从多个线程获取到队列并让一个单独的线程一次处理该队列的一个项目的最佳方法是什么？

当我尝试断开活动与多个线程的连接时，我经常使用这种模式。

我为此使用了 BlockingCollection，如下面的代码摘录所示：

// start this task in a static constructor
Task.Factory.StartNew(() => ProcessMultiUseQueueEntries(), TaskCreationOptions.LongRunning);


private static BlockingCollection<Tuple<XClientMsgExt, BOInfo, string, BOStatus>> _q = new BlockingCollection<Tuple<XClientMsgExt, BOInfo, string, BOStatus>>();

    /// <summary>
    /// queued - Simple mechanism that will log the fact that this user is sending an xMsg (FROM a user)
    /// </summary>
    public static void LogXMsgFromUser(XClientMsgExt xMsg)
    {
        _q.Add(new Tuple<XClientMsgExt, BOInfo, string, BOStatus>(xMsg, null, "", BOStatus.Ignore));
    }

    /// <summary>
    /// queued - Simple mechanism that will log the data being executed by this user
    /// </summary>
    public static void LogBOToUser(BOInfo boInfo)
    {
        _q.Add(new Tuple<XClientMsgExt, BOInfo, string, BOStatus>(null, boInfo, "", BOStatus.Ignore));
    }

    /// <summary>
    /// queued - Simple mechanism that will log the status of the BO being executed by this user (causes the red square to flash)
    /// </summary>
    public static void LogBOStatus(string UserID, BOStatus status)
    {
        _q.Add(new Tuple<XClientMsgExt, BOInfo, string, BOStatus>(null, null, UserID, status));
    }

    /// <summary>
    /// An endless thread that will keep checking the Queue for new entrants.
    /// NOTE - no error handling since this can't fail... :) lol etc 
    /// </summary>
    private static void ProcessMultiUseQueueEntries()
    {
        while (true)        //  eternal loop
        {
            Tuple<XClientMsgExt, BOInfo, string, BOStatus> tuple = _q.Take();

            // Do stuff

        }
    }

这工作得很好 - 所以我想 - 直到 VS2010 中的性能向导开始突出显示 _q.Take() 行作为最高争用我的代码中的一行！

注意，我还使用了带有 ManualResetEvent 组合的标准 ConcurrentQueue，每次将项目插入队列时，我都会发出重置事件信号，允许工作线程检查和处理队列，但这也具有在 . WaitOne() 方法...

是否有其他方法来解决这种常见模式，即让许多线程将对象添加到并发队列中 - 并让单个线程在自己的时间内一次处理一个项目...

谢谢！！

Answer 1

竞争最高的线路？是的，因为它是一个阻塞集合！该调用将阻塞（例如，它可能正在等待WaitHandle），直到另一个元素添加到集合中。

您确定这是一个问题吗？这听起来正是我所期望的。

如果不清楚我的意思，请考虑以下代码：

var blocker = new BlockingCollection<int>();
int nextItem = blocker.Take();

您期望上面的 Take 调用运行多长时间？我希望它会永远等待，因为blocker中没有添加任何内容。因此，如果我分析上述代码的“性能”，我会在长时间运行的方法列表的顶部看到 Take 。但这并不表示存在问题；再次强调：您想要阻止该调用。

---

在完全独立的注释中，我是否可以建议将 Tuple 替换为属性具有描述性名称的类型？ （当然，这个建议与你的问题无关；这只是一条一般性建议。）

Answer 2

_q.Take() 是争用最高的行，其本身毫无意义。如果许多线程正在等待项目，就会出现争用。最大的问题是这种争用是否会损害您的性能。您需要找到以下几个问题的答案：

1. 您是否能够足够快地处理项目以防止队列无限制增长？
2. 就 CPU 使用率而言，争用是否会造成您的损失？
3. 如果您停止向集合中添加内容，性能向导是否仍报告争用？
4. 如果停止向集合中添加内容，CPU 使用率是否会很高？

如果您能够阻止队列增长，并且 CPU 资源不会花在获取物品上，那么就没有问题。除非您从集合中获得的线程数超出了必要的阅读量。

Answer 3

您是否考虑过使用实际的 MSMQ ？可能看起来有点矫枉过正，但它确实提供了您所需要的。并不是说您的应用程序不能既是编写器又具有专用线程来读取和处理消息。