在某些情况下使用阻塞处理从队列中消耗元素答案

【问题标题】：Consuming elements from queue with blocking processing in some cases在某些情况下使用阻塞处理从队列中消耗元素
【发布时间】：2019-01-20 04:32:55
【问题描述】：

我有一个推送到队列的线程和一个消耗队列中元素的线程。其中一个元素的处理是异步的，但我不想在这个处理过程中处理其他元素。（假设输出流和队列是线程安全的）

我想知道实现消费线程的最佳方式是什么……我认为 while(true) 和条件不是最佳选择。

实现是否简单（process2 必须是异步的）。

#include <iostream>
#include <queue>
#include <thread>
#include <atomic>
#include <future>

std::atomic_bool isProcess2Processing{false};

void process0()
{
    std::cout << "process0" << std::endl;
}

void process1()
{
    std::cout << "process1" << std::endl;
}

void process2()
{
    std::async(std::launch::async, []() { isProcess2Processing = true; std::cout << "start process2" << std::endl; while (std::rand() > 10000) {}; std::cout << "finished proces2" << std::endl; isProcess2Processing = false; });
}

void consume(int x)
{
    if (x == 0)
    {
        process0();
    }
    else if (x == 1)
    {
        process1();
    }
    else
    {
        process2();
    }
}

int main()
{
    std::queue<int> q;

    std::thread consumingThread([&q]() {
        while (true) {
            if (!q.empty() && !isProcess2Processing) {
                consume(q.front());
                q.pop();
            }
        }
    });

    while (true)
    {
        q.push(std::rand() % 3);
    }
}

【问题讨论】：

您可以简单地在consumingThread 中运行process2（无需创建新线程）。并且没有其他东西可以同时运行。
不，process2 必须是异步的 - 我无法更改它。我的代码只是一个演示。我必须在process2 中使用异步调用。
你可以一直等。
您对process2 有什么控制权？你能让它返回一个future，然后调用者可以等待吗？还是在最后通知一个条件变量？

标签： c++ multithreading asynchronous queue c++14

【解决方案1】：

我想知道实现消费线程的最佳方式是什么……我想 while(true) 和条件不是最佳选择。

您的想法是有道理的：使用这样的 while 循环（即不涉及任何等待）的最大问题是您在浪费 CPU 时间和功率。您的辅助线程（以及根据给定主线程的代码）无缘无故地将 CPU 内核全部保留一段时间，因此其他任务无法为自己获得 CPU 时间。

进行更改的最天真的方法是添加some kind of sleep，如下所示：

std::thread consumingThread([&q]() {
    while (true) {
        if (!q.empty() && !isProcess2Processing) {
            consume(q.front());
            q.pop();
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
});

在这里，您将休眠 5 毫秒，在此期间调度程序将能够让其他任务完成它们的工作。

除此之外，您还应该确保每个循环的退出条件，并在您离开 main() 之前调用 consumingThread.join();。

【讨论】：