c++ - Qt 多线程信号量和条件变量

Question

我想在我的多线程应用程序上完成一个简单的任务，但我不知道该怎么做。

我有 3 个线程:

• 1 个线程作为主线程
• 2 个线程作为从线程

1) 2 个从线程 只需要获取一些数据，休眠几毫秒并再次处理。

2) 主线程等待从线程完成工作，处理从线程获取的数据等

如何确定两个从属线程都完成了它们的获取(而不是从属线程 n°1 在从属线程 n°2 可以完成之前完成了两倍的工作)n 主线程处理的顺序所有数据？

• 使用信号量，我无法控制两个线程是否都完成了它们的工作，我只能注意到数据已被获取两次。

• 使用条件变量，我不知道如何单独通知主线程两个从线程都完成了他们的工作

我希望你已经足够清楚地帮助我了。

编辑 1:我的目标之一是不阻止从属线程以拥有真正的多线程应用程序。我希望，在最好的情况下，主线程处理数据，而从属线程获取下一个数据。

我在 Linux 上使用 Qt 4.8

谢谢你的帮助

Answer 1

根据问题和标签，这不是要求的答案。正如我们在评论中讨论的那样，我想展示一下使用 atomic 和 sleep() 会是什么样子:

#include <atomic>
#include <chrono>
#include <iostream>
#include <sstream>
#include <string>
#include <thread>

using namespace std;

typedef atomic<const char*> TransferBuf;

enum { NLoops = 10 };
enum { dtSusp = 20000 };

atomic<bool> stop(false);

// fake acquisition of data
void acquire(int id, TransferBuf &pTransferBuf)
{
  string buffers[2]; size_t iBuf = 0;
  for (size_t step = 0;; ++step) {
    // acquire data
    ostringstream out;
    out << "thread " << id << ", " << step << ": ";
    for (size_t i = 0; i < 10; ++i) {
      if (stop) return;
      out << (i ? ", " : "") << "data " << step << i;
    }
    buffers[iBuf] = out.str();
    // polling: wait for transfer buffer free
    const char *pBuf = buffers[iBuf].c_str(), *pSent = 0;
    while (!pTransferBuf.compare_exchange_strong(pSent, pBuf)) {
      if (stop) return;
      // if storage fails sleep() (to give other threads a chance)
      this_thread::sleep_for(chrono::microseconds(dtSusp));
      // reset pSent (which is updated)
      pSent = 0;
    }
    // buffer sent -> flip buffers for next acquisition
    iBuf ^= 1;
    // sleep some time (on requirement)
    this_thread::sleep_for(chrono::microseconds(dtSusp));
  }
}

int main(int, char**)
{
  TransferBuf pBuf1(0), pBuf2(0);
  // start thread 1 for data aquisition
  thread thread1(&acquire, 1, ref(pBuf1));
  // start thread 2 for data aquisition
  thread thread2(&acquire, 2, ref(pBuf2));
  // start processing
  for (size_t i = 0; i < NLoops; ++i) {
    // polling: wait for data
    while (!pBuf1 || !pBuf2) {
      // if storage fails sleep() (to give other threads a chance)
      this_thread::sleep_for(chrono::microseconds(dtSusp));
    }
    // process data
    const char *pData1 = pBuf1, *pData2 = pBuf2;
    cout << pData1 << endl;
    cout << pData2 << endl;
    // release processed data
    pBuf1 = pBuf2 = 0;
  }
  // finish
  stop = true;
  thread1.join(); thread2.join();
  // done
  return 0;
}

抱歉，它不包含任何 Qt 符号，但自 C++11 以来，标准库中的所有内容都可用。

在 cygwin 上用 gcc 编译和测试:

$ g++ -std=c++11 -o test-master-slave test-master-slave.cc 

$ ./test-master-slave.exe 
thread 1, 0: data 00, data 01, data 02, data 03, data 04, data 05, data 06, data 07, data 08, data 09
thread 2, 0: data 00, data 01, data 02, data 03, data 04, data 05, data 06, data 07, data 08, data 09
thread 1, 1: data 10, data 11, data 12, data 13, data 14, data 15, data 16, data 17, data 18, data 19
thread 2, 1: data 10, data 11, data 12, data 13, data 14, data 15, data 16, data 17, data 18, data 19
thread 1, 2: data 20, data 21, data 22, data 23, data 24, data 25, data 26, data 27, data 28, data 29
thread 2, 2: data 20, data 21, data 22, data 23, data 24, data 25, data 26, data 27, data 28, data 29
thread 1, 3: data 30, data 31, data 32, data 33, data 34, data 35, data 36, data 37, data 38, data 39
thread 2, 3: data 30, data 31, data 32, data 33, data 34, data 35, data 36, data 37, data 38, data 39
thread 1, 4: data 40, data 41, data 42, data 43, data 44, data 45, data 46, data 47, data 48, data 49
thread 2, 4: data 40, data 41, data 42, data 43, data 44, data 45, data 46, data 47, data 48, data 49
thread 1, 5: data 50, data 51, data 52, data 53, data 54, data 55, data 56, data 57, data 58, data 59
thread 2, 5: data 50, data 51, data 52, data 53, data 54, data 55, data 56, data 57, data 58, data 59
thread 1, 6: data 60, data 61, data 62, data 63, data 64, data 65, data 66, data 67, data 68, data 69
thread 2, 6: data 60, data 61, data 62, data 63, data 64, data 65, data 66, data 67, data 68, data 69
thread 1, 7: data 70, data 71, data 72, data 73, data 74, data 75, data 76, data 77, data 78, data 79
thread 2, 7: data 70, data 71, data 72, data 73, data 74, data 75, data 76, data 77, data 78, data 79
thread 1, 8: data 80, data 81, data 82, data 83, data 84, data 85, data 86, data 87, data 88, data 89
thread 2, 8: data 80, data 81, data 82, data 83, data 84, data 85, data 86, data 87, data 88, data 89
thread 1, 9: data 90, data 91, data 92, data 93, data 94, data 95, data 96, data 97, data 98, data 99
thread 2, 9: data 90, data 91, data 92, data 93, data 94, data 95, data 96, data 97, data 98, data 99

如果主线程中的处理发生在 QTimer 信号处理程序中，它应该看起来像这样:

void timeout()
{
    // polling: wait for data
    if (!pBuf1 || !pBuf2) return; // try again next timeout
    // process data
    const char *pData1 = pBuf1, *pData2 = pBuf2;
    cout << pData1 << endl;
    cout << pData2 << endl;
    // release processed data
    pBuf1 = pBuf2 = 0;
}

顺便说一句。我发现这篇文章很好地概述了 atomic 的实际用法:Atomic pointers in c++ and passing objects between threads