我有一个包含一系列节点的 XML 文件。每个节点代表一个我需要解析并添加到排序列表中的元素(顺序必须与文件中找到的节点的顺序相同)。

目前我使用的是顺序解决方案:

struct Graphic
{
    bool parse()
    {
        // parsing...
        return parse_outcome;
    }
};

vector<unique_ptr<Graphic>> graphics;

void producer()
{
    for (size_t i = 0; i < N_GRAPHICS; i++)
    {
        auto g = new Graphic();

        if (g->parse())
            graphics.emplace_back(g);
        else
            delete g;
    }
}

因此，仅当图形(实际上是派生自 Graphic 的类的实例、Line、Rectangle 等时，这就是 new 的原因) 可以正确解析，它将被添加到我的数据结构中。

因为我只关心这些图形添加到我的列表中的顺序，所以我想异步调用解析方法，这样 producer 的任务就是从文件中读取每个节点并将该图形添加到数据结构中，而消费者的任务是在新图形准备好解析时解析每个图形。

现在我有几个消费者线程(在主线程中创建)，我的代码如下所示:

queue<pair<Graphic*, size_t>> q;
mutex m;
atomic<size_t> n_elements;

void producer()
{
    for (size_t i = 0; i < N_GRAPHICS; i++)
    {
        auto g = new Graphic();
        graphics.emplace_back(g);
        q.emplace(make_pair(g, i));
    }

    n_elements = graphics.size();
}

void consumer()
{
    pair<Graphic*, size_t> item;

    while (true)
    {
        {
            std::unique_lock<std::mutex> lk(m);

            if (n_elements == 0)
                return;

            n_elements--;
            item = q.front();
            q.pop();
        }

        if (!item.first->parse())
        {
            // here I should remove the item from the vector
            assert(graphics[item.second].get() == item.first);
            delete item.first;
            graphics[item.second] = nullptr;
        }
    }
}

我首先在 main 中运行生产者，这样当第一个消费者启动时队列已经完全满了。

int main()
{
    producer();

    vector<thread> threads;

    for (auto i = 0; i < N_THREADS; i++)
        threads.emplace_back(consumer);

    for (auto& t : threads)
        t.join();

    return 0;
}

并发版本似乎至少比原始版本快两倍。 完整代码已上传here .

现在我想知道:

• 我的代码中是否存在任何(同步)错误？
• 有没有办法更快(或更好)地达到相同的结果？

此外，我注意到如果我将线程数设置为 8，我的计算机上会得到最好的结果(根据运行时间)。更多(或更少)的线程会得到最差的结果。为什么？

最佳答案

Blockquote There isn't synchronization errors, but I think that the memory managing could be better, since your code leaked if parse() throws an exception.

没有同步错误，但我认为您的内存管理可能会更好，因为如果 parse() 抛出异常，您将有泄漏。

Blockquote Is there a way to achieve the same result faster (or better)?

可能吧。您可以使用线程池的简单实现和为您执行 parse() 的 lambda。

下面的代码说明了这种方法。我使用线程池实现 here

#include <iostream>
#include <stdexcept>
#include <vector>
#include <memory>
#include <chrono>
#include <utility>
#include <cassert>
#include <ThreadPool.h>

using namespace std;
using namespace std::chrono;



#define N_GRAPHICS        (1000*1000*1)
#define N_THREADS       8


struct Graphic;
using GPtr = std::unique_ptr<Graphic>;

static vector<GPtr> graphics;

struct Graphic
{
    Graphic()
        : status(false)
    {
    }


    bool parse()
    {
        // waste time
        try
        {
            throw runtime_error("");
        } 
        catch (runtime_error)
        {
        }

        status = true;
        //return false;
        return true;
    }


    bool status;
};


int main()
{
    auto start = system_clock::now();

    auto producer_unit = []()-> GPtr {
        std::unique_ptr<Graphic> g(new Graphic);
        if(!g->parse()){
            g.reset(); // if g don't parse, return nullptr
        }
        return g;        
    };

    using ResultPool = std::vector<std::future<GPtr>>;
    ResultPool results;
    // ThreadPool pool(thread::hardware_concurrency());
    ThreadPool pool(N_THREADS);
    for(int i = 0; i <N_GRAPHICS; ++i){
     // Running async task
     results.emplace_back(pool.enqueue(producer_unit));
    }

   for(auto &t : results){
        auto value = t.get();
        if(value){
          graphics.emplace_back(std::move(value));
        }
    }

    auto duration = duration_cast<milliseconds>(system_clock::now() - start);
    cout << "Elapsed: " << duration.count() << endl;

    for (size_t i = 0; i < graphics.size(); i++)
    {
        if (!graphics[i]->status)
        {
            cerr << "Assertion failed! (" << i << ")" << endl;
            break;
        }
    }

    cin.get();
    return 0;
}

它在我的机器上快一点(1 秒)，更具可读性，并且消除了共享数据的必要性(同步是邪恶的，避免它或以可靠和有效的方式隐藏它)。

关于c++ - 如何将并发解决方案应用于类似生产者-消费者的情况，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/31357705/