c++ - 如何添加到 QtConcurrent::map 操作的 QStringList 中？

Question

我正在尝试创建一个应用程序，它将永久监控一个文件夹，并将任何新文件添加到队列中以进行多线程处理。

这是我的:

int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);
    QDir dir("/home/boat/Programming/");
    QStringList folders = QStringList() << "/home/boat/Programming/";
    QStringList filters = QStringList() << "*.boat";
    QStringList boatFileNames = dir.entryList(filters);

    /* Start Threading | Files from list are deleted on completion */
    QtConcurrent::map(boatFileNames, &someFunction);

    /* Monitor folder for new files to add to queue */
    QFileSystemWatcher fsw(folders);
    QObject::connect(&fsw,&QFileSystemWatcher::directoryChanged,[&](){
            ????
    });
    a.exec();
}

我唯一需要确保线程安全的是我不能让同一个函数的两个线程同时操作一个文件。

在我看来，我有两个选择，但我都不知道该怎么做:

1. 等到映射完成，再次检查文件夹中的文件，重新填充 QStringList，然后重新启动映射。
2. 添加到提供 map 的 QStringList 中，假设 map 永远不会关闭，并且您可以继续向队列中添加项目，它将对它们运行函数。

这些都是可能的吗，或者我应该用另一种方法来解决这个问题吗？

Answer 1

How can you add to a QStringList being operated on byQtConcurrent::map? I'm trying to create an application that will perpetually monitor a folder, and add any new files to a queue for multi-threaded processing.

我什至不会使用 QtConcurrent::map，因为任务可以用简单的堆栈或类似队列的容器来实现，并由互斥量保护。

将所有工作(消费者)线程连接到自己的信号，例如 void MyClass::newFileAvailable()。

制作 void QFileSystemWatcher::directoryChanged(const QString &path) 信号来填充类队列(或者你可以使它类栈)结构:

class FilePathQueue
{
public:
   void push(const QString& path)
   {
      QMutexLocker lock(&mutex);
      list.push_back(path);
   }
   QString pop()
   {
      QMutexLocker lock(&mutex);
      if (list.isEmpty())
         return QString();
      QString ret = list.front();
      list.pop_front();
      return ret;
   }
private:
   QMutex mutex;
   QStringList list;
};

因此，所有工作线程都应订阅 newFileAvailable() 信号，并将通过 FilePathQueue::pop() 使用新文件路径并检查是否有任何项目左边可用。

/* File path thread-safe container */
FilePathQueue fileList;

/* Monitor folder for new files to add to queue */
QFileSystemWatcher fsw(folders);
QObject::connect(&fsw, &QFileSystemWatcher::directoryChanged,
   [&](const QString& path){
        fileList.push(path);     // push in new path
        emit newFileAvailable(); // let workers know
        // TODO: here we can have condition.notify_all() or notify_one()
        // instead of signal emit for the higher performance
});

// launch new workers
for(int n=0; n<8; ++n)
{
   Worker* worker = new Worker(this, fileList);  // create worker and
   worker->start(); // starts own thread etc.

   connect(this, &MyClass::newFileAvailable,     // connect it to new
           worker, &MyWorker::processNewFile);   // file signal
}

worker 类(Class):

// the worker thread supposed to be running in own thread
// see QObject::moveToThread()
class MyWorker : public QObject
{
   public:
      MyWorker(QObject* parent, FilePathQueue& list) :
         QObject(parent), fileList(list) {}
   ///
   public slots:
      void processNewFile()
      {
          QString path = fileList.pop();
          if (path.isEmpty()) return;

          process(path);
      }

      void process(const QString& path);
   private:
      FilePathQueue& fileList;
};

这使用 Qt 很容易完成，并且非常易于管理，例如，每个工作人员每秒数千个文件到达信号，但对于更多消费者或更多实时性能系统的更高速率 QWaitCondition::wait() 应该在工作线程端被利用，而发布者将根据逻辑执行 QWaitCondition::notify_one() 或 notify_all()。要实现该逻辑，需要提供更详细的工作线程示例。