【问题标题】:Tell if Concurrent Queue updated without looping判断并发队列是否在没有循环的情况下更新
【发布时间】:2015-08-07 19:21:04
【问题描述】:

感谢您的帮助。我有一个由并发队列链接的三线程进程。线程一处理信息,返回到第二个线程,第二个线程将数据放入并发队列。第三个线程只是像这样循环:

while (true) {
    if(queue.TryDequeue(out info)) {
        doStuff(info);
    } else {
        Thread.Sleep(1);
    }
}

有没有更好的方法来处理它,这样我就不会过多地遍历循环?该应用程序对性能极为敏感,目前只有 TryDequeue 占用了应用程序运行时间的约 8-9%。希望尽可能减少这种情况,但不确定我的选择是什么。

【问题讨论】:

    标签: c# multithreading concurrency


    【解决方案1】:

    您应该考虑使用 System.Collections.Concurrent.BlockingCollection 及其 Add() / Take() 方法。使用 Take() 您的第三个线程将在等待新项目时暂停。 Add() 是线程安全的,可以被第二个线程使用。

    通过这种方法,您应该能够将代码简化为类似的内容:

    while (true) {
       var info = collection.Take();
       doStuff(info);
    }
    

    【讨论】:

    • 我之前一直在使用 BlockingCollection,但发现我在 *.take() 中花费的时间比在 while(true) 循环中花费的时间更多。不确定那里的程序差异,但交换到 ConcurrentQueue 似乎显示出性能提升。我会换回来看看我是否可以再次确认/拒绝
    • 你是如何衡量那个时间的?
    • 在操作之前我保存了 Environment.TickCount,之后我从保存的 TickCount 中减去它并将其添加到计数器中
    • 似乎还可以。如果您不关心项目的顺序,您可以尝试的另一种可能性是 ConcurrentBag..
    【解决方案2】:

    您可以增加睡眠时间。我也会使用await Task.Delay 而不是睡眠。这样,您可以等待更长的时间,而无需 Thread.Sleep 使用的额外 CPU 周期,并且仍然可以通过使用 CancellationTokenSource 来取消延迟。

    另一方面,还有更好的排队工作的方法。考虑到您似乎希望同步运行这些作业,一个示例是有一个单例类,它接收您的工作项并将它们排队。因此,如果在您添加一项时队列中没有项目,它应该检测到然后开始您的工作流程。在您的工作流程结束时,检查更多工作,使用递归来完成该工作,或者如果没有更多工作则退出工作流程,当您将项目添加到空队列时,该流程将再次运行。如果我的假设是错误的,并且您可以并行运行这些作业,那么为什么要使用队列?

    您可能希望使用ObservableCollection 的线程安全实现。看看这个 SO 问题ObservableCollection and threading

    【讨论】:

    • 与上述答案相同,出于性能原因,我需要这些都异步运行。应用程序现在每秒处理 > 100k rabbitmq 消息(峰值 ~138k/s),同步运行会破坏速度。我担心如果我等到我将数据放入队列以产生一个处理线程,我将得到备份,因为即使任务创建和任务实际启动之间的几毫秒延迟也会是毁灭性的。这就是为什么我现在有一个专门的任务正在全职运行
    • 那么,为什么还要使用队列,而将第二个任务直接调用到第三个任务中呢?
    • 我有多个第一个任务异步运行,并且必须保留顺序,所以我不能让它们在完成时启动到下一个任务。当前设置是一个任务队列,因此它们异步运行,我只需弹出顶部任务,等待它完成,获取结果并将其传递给下一个任务。这一切都基于 RabbitMQ 消息,第一个任务是在接收到消息时产生并开始处理它。我有多达 3 个同时运行的流程任务,因为这是流程中完成的大部分实际工作量
    • 看来你应该考虑重写你的工作流程。您担心它们何时启动的顺序,但如果您将任务添加到线程池,则线程池负责确定接下来应该运行哪个任务。这不保证任何类型的订单。在不了解您的整个工作流程的情况下,恐怕我对您的执行方式应该如何安排没有多大帮助。也许考虑引发事件以触发下一步......这只是我在猜测混乱中的特定顺序。
    • 所以我确保顺序的方式是任务队列。随着任务的产生(因此,当消息进入并导致产生任务时),任务被放入队列中。任务本身不会影响外部数据,因此它们实际完成的顺序无关紧要。为了确保正确的顺序,我需要做的就是从队列中弹出顶部任务,如有必要,调用 task.wait(),然后获取该结果。以这种方式设置它允许处理的并行化,同时保持输出中的顺序
    【解决方案3】:

    我没有避免循环的建议,但是我建议您远离

    while (true)
    

    并考虑这个:

    MyThing thing;
    while (queue.TryDequeue(out thing))
    {
        doWork(thing);
    }
    

    将其放入每次修改队列时都会调用的方法中,这样可以确保它在需要时运行,但在不需要时结束。

    【讨论】:

    • 这似乎剥夺了 OP 想要的东西。如果TryDequeue 为假,此代码将退出,因此无法继续检查新作业。
    • 我有点不愿意这样做,因为我必须确保这个 doWork() 永远不会被备份。我担心当有工作要做时产生一个新任务可能会导致任务创建和激活之间的延迟,这对于流程来说是完全不可接受的。
    • Jacob,如果 TryDequeue 为假,则无事可做,这就是为什么我声明每次添加某些内容时都必须再次调用循环方法的原因。一旦这个过程开始,假设有多个项目排队,那么它将继续'直到完成。
    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2021-11-12
    • 1970-01-01
    • 1970-01-01
    相关资源
    最近更新 更多