c++ - 在 C++ 中使用 Boost 的线程池无法正常工作

Question

我已经实现了来自 How to create a thread pool using boost in C++? 的解决方案但是我遇到了 io_service::stop() 函数停止进一步处理我的线程的问题。

在我的例子中，我的池中有 3 个线程，并试图通过它运行大约 11000 条记录。每条记录都独立于其他记录，因此我只是想通过为每条记录创建并行运行来加快处理速度。

void processRecord (unsigned int i, unsigned int numRecords)
{
    cout << i << "/" << numRecords << endl;

    // do Processing...
}

#define MAX_THREADS 3
unsigned int numRecords=11000

boost::asio::io_service ioService;
boost::thread_group threadPool;

boost::asio::io_service::work work (ioService);

for (unsigned int i=0 ; i<MAX_THREADS ; ++i)
{
    threadPool.create_thread (boost::bind (&boost::asio::io_service::run, &ioService));
}

for (unsigned int i=0 ; i<numRecords ; ++i)
{
    ioService.post (boost::bind (processRecord, i, numRecords);
}

// ioService.stop ();          // Was causing ioService to stop
work.reset();                  // Wait for all work to be finished.
threadPool.join_all ();

processAllRecords ();

我看到的问题是，在调用 ioService.post() 将进程推送到池中的循环完成后，它会触发 ioService.stop() 调用并停止所有进一步处理。这通常发生在实际处理了大约 400 条记录之后。

因此，大约 11000 条记录中只有大约 400 条正在处理。

我是在 C++ 中使用线程的新手，所以我不确定我遗漏了什么或如何更正此问题。任何帮助将不胜感激。

编辑:我修改了上面的代码以反射(reflect)我为使其工作所做的更改。本质上，ioService.stop() 调用导致所有进一步处理停止。我用 work.wait() 替换了它，这样它就会等到所有工作完成。

Edit2:我在之前的编辑中使用了错误的功能。它应该是 work.reset()。

Answer 1

使用你的代码，我将 boost::asio 用于线程组的方式， 将工作括在括号中并使用 scoped_ptr。只是一个想法。

void processRecord (unsigned int i, unsigned int numRecords)
{
    cout << i << "/" << numRecords << endl;

    // do Processing...
}

#define MAX_THREADS 3
unsigned int numRecords=11000

boost::asio::io_service ioService;
boost::thread_group threadPool;

// by using a scoped pointer for the io_service::work
// and enclosing the threading in brackets
// this should run until all the jobs have finished
// and you don't need to call work.reset()  

// added brackets around threading
{
    // made work a boost::scoped_ptr
    boost::scoped_ptr< boost::asio::io_service::work > 
          work ( new boost::asio::io_service(ioService) );

    for (unsigned int i=0 ; i<MAX_THREADS ; ++i)
    {
        threadPool.create_thread (
           boost::bind (&boost::asio::io_service::run, &ioService));
    }

    for (unsigned int i=0 ; i<numRecords ; ++i)
    {
        ioService.post (boost::bind (processRecord, i, numRecords);
    }
}
// now just have to join
threadPool.join_all ();

processAllRecords ();