如何启动多个线程和每个线程处理不同的文件？

Question

我有一个单线程应用程序，它通过调用 send_new_file 将文件发送到其他服务器

void send_new_file_command::start_sending_file()
{
    m_thread = thread(&send_new_file_command::execute_file, this);
}

void send_new_file_command::execute_file()
{
    for (auto it = files_need_to_send.begin(); it != files_need_to_send.end() && !is_complete(); ++it)
    {
        {
            std::unique_lock<spinning_lock> guard(lock_obj);
            m_current_file = *it;
        }
        // send a file.
        // I want to call this in parallel
        send_new_file(*it);
    }
}

有什么方法可以让我拥有多个线程，每个线程分别发送一个文件。例如，假设我们有 4 个线程，线程 1、2、3、4 将并行发送不同的文件。我要同时拨打send_new_file？

我正在使用std::thread。我正在查看有关如何在 C++ 中执行此操作的线程示例，但很困惑如何在此处划分每个线程的文件数并确保每个线程都在文件子集上工作。

  std::vector<std::thread> threads;
  for (int i = 0; i < 4; ++i)
    threads.push_back(std::thread(send_new_file(*it)));

我的背景是 Java，所以对如何使用 std::thread 在 C++ 中执行此操作有点困惑。

Answer 1

这是一种使用工作队列的相当简单的方法。您可以将代码 sn-ps 连接成一个独立的程序。我们将使用以下标准库头文件。

#include <fstream>
#include <iostream>
#include <mutex>
#include <string>
#include <thread>
#include <vector>

首先，我们定义一个函数，该函数接受一个文件名并将其发送到它应该去的任何地方。我将通过简单地将其写入/dev/null 来模拟它。

void
send_file(const std::string& filename)
{
  std::ifstream istr {};
  std::ofstream ostr {};
  std::string line {};
  istr.exceptions(std::ifstream::badbit);
  ostr.exceptions(std::ofstream::badbit);
  istr.open(filename);
  ostr.open("/dev/null");
  while (std::getline(istr, line))
    ostr << line << '\n';
}

接下来，我们定义一个函数，它接受一个指向仍需要发送的文件的std::vector 的指针和另一个指向应该保护该向量的std::mutex 的指针。我使用指针而不是引用，因为这使我以后可以更简单地创建std::threads。如果你不喜欢它，你不需要这样做。

int
send_files(std::vector<std::string> *const files_p, std::mutex *const mutex_p)
{
  auto count = 0;
  while (true)
    {
      std::string next {};
      {
        const std::unique_lock<std::mutex> lck {*mutex_p};
        if (files_p->empty())  // nothing left to do
          return count;
        next = std::move(files_p->back());
        files_p->pop_back();
      }
      send_file(next);
      count += 1;
    }
}

重要的是，在执行发送文件的实际工作时，我们不会持有锁。否则，我们将完全扼杀并发性。我还小心翼翼地在持有锁时不要分配任何内存。通常，您会看到std::lists 用作工作队列，std::condition_variables 用于在队列发生更改时发出信号。不久前，我在another answer 中发布了显示此内容的代码。 然而，在这个简单的例子中，队列只会被移除，所以 std::vector 非常适合。

最后，我们使用我们在一个简单程序中的内容，该程序为每个硬件并发单元创建一个线程，并要求这些线程发送命令行参数中指定的所有文件。请注意，正如所写，这将以相反的顺序处理列表。但是，如果这对您来说是个问题，那么进行更改是微不足道的。

int
main(int argc, char * * argv)
{
  const auto nthreads = std::thread::hardware_concurrency();
  std::mutex mutex {};
  std::vector<std::thread> threads {};
  std::vector<std::string> files {};
  files.reserve(argc - 1);
  for (auto i = 1; i < argc; ++i)
    files.push_back(argv[i]);
  threads.reserve(nthreads);
  for (auto t = 0U; t < nthreads; ++t)
    threads.emplace_back(send_files, &files, &mutex);
  for (auto t = 0U; t < nthreads; ++t)
    threads[t].join();
}

Answer 2

第一种方法

有第一个简单的解决方案：

• 您的类包含要处理的文件向量
• 只有一个线程通过函数execute_file()管理这个向量
• 此函数根据需要创建多个线程，每个线程处理一个文件
• 最后，所有线程都加入（强制）

代码如下所示：

struct send_new_file_command {
    vector<string> files_need_to_send;
public:
    send_new_file_command(vector<string> f) : files_need_to_send(f) {}
    void execute_file();
};
void send_new_file_command::execute_file()
{
    vector<thread> exec;
    for(auto it = files_need_to_send.begin(); it != files_need_to_send.end(); ++it)
    {
        exec.push_back(thread(send_new_file, *it));
    }
    for(auto &e : exec)
        e.join();
}

可以使用以下代码测试代码：

void send_new_file(string x) { // simulator 
    for(int i = 0; i<10; i++) {
        cout << x << endl;
        this_thread::sleep_for(chrono::milliseconds(500));
    }
}
int main() {
    vector<string>vs{"a", "b", "c", "d"};
    send_new_file_command sfc(vs);
    sfc.execute_file();
    return 0;
}

这个解决方案非常简单。它有两个主要缺点：

• 它可能会启动比您的硬件可以管理的线程多得多的线程。所以只有少数几个真正同时运行。
• 线程专用于文件。 F如果是短文件，线程又空闲了，就不会被重用了。

其他解决方案

还有很多其他解决方案。例如：

• 这种方法的一个变体是启动固定数量的线程，每个线程在准备好后立即查看要为下一项处理的文件向量。然后，您需要引入强锁定。

• 您可以考虑使用futures，而不是使用原始线程，启动std::async(std::launch::async, send_new_file, *it);

Answer 3

在性能方面是最好的方法：

1. 使用std::atomic<int>声明一个计数器变量
2. 在向量、数组等中创建线程
3. 为每个线程调用join

线程的主函数然后访问并递增共享计数器并将结果保存在循环中的局部变量中：

std::atomic<int> counter = 0;
for(int j = 0;j<4;j++)
{
    threads.push_back(std::thread([&](){
        for(int i; (i = counter++) < size;)//the counter variable must be atomic!
        {
            do_work(i);
        }
    }));
}

for(int j = 0;j<4;j++)
    threads[i].join();