在 C++ 中将数据传递给线程的好方法是什么？

Question

我正在学习使用 c++ 进行多线程编码。我需要做的是不断从键盘读取单词，并将其传递给数据线程进行数据处理。我使用 global 变量 word[] 来传递数据。当 word[0] != 0 表示来自键盘的新输入。 数据线程会在读取数据后将word[0]设置为0。有用！但我不确定它是否安全，或者有更好的方法来做到这一点。这是我的代码：

#include <iostream> 
#include <thread>
#include <cstdio>
#include <cstring>

using namespace std;

static const int buff_len = 32;
static char* word = new char[buff_len];


static void data_thread () {         // thread to handle data
  while (1)               
  {
    if (word[0]) {                   // have a new word
      char* w = new char[buff_len];
      strcpy(w, word);
      cout << "Data processed!\n";
      word[0] = 0;                   // Inform the producer that we consumed the word
    }
  }
};

static void read_keyboard () {
  char * linebuf = new char[buff_len];
  thread * worker = new thread( data_thread );
  
  while (1)                                     //enter "end" to terminate the loop
  {
    if (!std::fgets( linebuf, buff_len, stdin)) // EOF?
      return;
    linebuf[strcspn(linebuf, "\n")] = '\0';     //remove new line '\n' from the string
    
    word = linebuf;                             // Pass the word to the worker thread
    while (word[0]);                            // Wait for the worker thread to consume it
  }
  worker->join();                               // Wait for the worker to terminate
}

int main ()
{
  read_keyboard(); 
  return 0;
}

Answer 1

这种类型的多线程实现的问题是忙于等待。输入阅读器和数据消费者都忙于等待并浪费 CPU 周期。要克服这个问题，您需要 Semaphore。

Semaphore s_full(0);
Semaphore s_empty(1);

void data_processor ()
{
    while (true) {
        // Wait for data availability.
        s_full.wait();
            // Data is available to you, consume it.
            process_data();
        // Unblock the data producer.
        s_empty.signal();
    }
}

void input_reader()
{
    while (true) {
        // Wait for empty buffer.
        s_empty.wait();
            // Read data.
            read_input_data();
        // Unblock data com=nsumer.
        s.full.signal();
    }
}

此外，此解决方案仅适用于单个数据消费者线程。但是对于多个数据消费者线程，您将需要线程安全的缓冲区队列和正确实现生产者 - 消费者问题。 有关解决此问题的更多信息，请参见下面的博客链接： 线程安全缓冲区队列： https://codeistry.wordpress.com/2018/03/08/buffer-queue-handling-in-multithreaded-environment/

生产者-消费者问题： https://codeistry.wordpress.com/2018/03/09/unordered-producer-consumer/

Answer 2

您的方法存在一些问题：

• 此方法不可扩展。如果您有超过 1 个处理线程怎么办？
• 您需要一个互斥锁来同步对word 存储的内存的读写访问。在这个例子的规模上，没什么大不了的。在“严肃”的应用程序中，您可能没有等到数据线程停止处理的奢侈。在这种情况下，您可能会想删除 while(word[0])，但这是不安全的。
• 您启动了一个“守护程序”线程（不完全但足够接近）来处理您的计算。大多数时候线程都在等待你的输入，没有它就无法继续。这是低效的，现代 C++ 为您提供了一种解决方法，而无需显式使用 std::async 范式处理原始线程。

#include <future>
#include <string>
#include <iostream>

static std::string worker(const std::string &input)
{
    // assume this is a lengthy operation
    return input.substr(1);
}

int main()
{
    while (true)
    {
        std::string input;
        std::getline (std::cin, input); 
        
        if (input.empty())
            break;
            
        std::future<std::string> fut= std::async(std::launch::async, &worker, input);
        // Other tasks
        // size_t n_stars = count_number_of_stars();
        //
        std::string result = fut.get(); // wait for the task to complete
        printf("Output : %s\n", result.c_str());
    }

    return 0;
}

在我看来，这样的方法是更好的方法。 std::async 将启动一个线程（如果指定了 std::launch::async 选项）并返回一个 waitable future。计算将在后台继续，您可以在主线程中进行其他工作。当你需要得到你的计算结果时，你可以get()future的结果（顺便说一下future也可以是void）。

您的 C++ 代码中也有很多 C-isms。除非有理由这样做，否则为什么不使用std::string？

Answer 3

在现代 CPP 多线程中，您应该使用 condition_variable、mutex 和 queue 来处理这个问题。互斥锁防止相互到达队列，条件变量使读取器线程休眠，直到写入器写入它所写的内容。下面是一个例子

static void data_thread (std::queue<char> & dataToProcess, std::mutex & mut, std::condition_variable & cv, std::atomic<bool>& finished) {         // thread to handle data
    std::string readData;
    while (!finished)
    {
        {
            std::unique_lock lock{mut};
            cv.wait(lock, [&] { return !dataToProcess.empty() || finished; });
            if (finished) {
                while (!dataToProcess.empty()){
                    readData += dataToProcess.front();
                    dataToProcess.pop();

                }
            }
            else{
                readData += dataToProcess.front();
                dataToProcess.pop();
            }
        }
        std::cout << "\nData processed\n";
    }
    std::cout << readData;
};

static void read_keyboard () {
    std::queue<char> data;
    std::condition_variable cv;
    std::mutex mut;
    std::atomic<bool> finished = false;
    std::thread worker = std::thread( data_thread, std::ref(data), std::ref(mut), std::ref(cv), std::ref(finished) );
    char temp;
    while (true)                                     //enter "end" to terminate the loop
    {
        if (!std::cin.get(temp)) // EOF?
        {
            std::cin.clear();
            finished = true;
            cv.notify_all();
            break;
        }

        {
            std::lock_guard lock {mut};
            data.push(temp);
        }
        cv.notify_all();
    }
    worker.join();                               // Wait for the worker to terminate
}

int main ()
{
    read_keyboard();
    return 0;
}

Answer 4

您正在寻找的是一个消息队列。这需要互斥体和条件变量。

这是github上的一个（不是我的，但我搜索时弹出）https://github.com/khuttun/PolyM

还有一个

https://www.justsoftwaresolutions.co.uk/threading/implementing-a-thread-safe-queue-using-condition-variables.html

我会因为发布链接而被告知，但我不会在这里输入整个代码，而且 github 也不会很快去任何地方