处理使用QThreadPool

Question

多个连接考虑，你需要保持与只是ocassionally发送命令的设备256个的TCP连接的情况。我想同时做到这一点（它需要阻止，直到它得到响应），我试图使用QThreadPool的这个目的，但我有一些怀疑，如果这是可能的。

我试图用QRunnable，但我不知道插座将如何线程之间的行为（即他们在创建插座应只用于线程？）

我也担心这个解决方案的效率如果有人能够提出一些替代方案，而不一定使用QT，我会很高兴。

下面我张贴的代码的某些代码段。

class Task : public QRunnable { 

    Task(){ 
     //creating TaskSubclass instance and socket in it 
    } 

private: 
    TaskSubclass    *sub; 

    void run() override { 
     //some debug info and variable setting... 
     sub->doSomething(args); 
     return; 
    } 
}; 

class TaskSubclass { 
    Socket   *sock;   // socket instance 
    //... 
    void doSomething(args) 
    { 
     //writing to socket here 
    } 
} 

class MainProgram : public QObject{ 
    Q_OBJECT 
private: 
    QThreadPool *pool; 
    Task *tasks; 

public: 
    MainProgram(){ 
     pool = new QThreadPool(this); 
     //create tasks here 
    } 

    void run(){ 
     //decide which task to start 
     pool->start(tasks[i]); 
    } 
}; 

Answer 1

我对这个问题的方案最是通过复用select()您的插座。这样你就不需要创建额外的线程，而且这是一个“非常POSIX”的方式。

见例如请参见本教程：

http://www.binarytides.com/multiple-socket-connections-fdset-select-linux/

或相关问题：

Using select(..) on client

Answer 2

由于OMD_AT有媒体链接指出，最好的解决办法是使用选择（）并让内核做的工作适合你:-)

在这里你有作为的一个例子ync方法和Syncron多线程方法。

在这个例子中，我们创建10个连接到一个谷歌的互联网服务和做一个简单的GET请求到服务器，我们测量所需的每一种方法的所有连接多长时间才能收到来自谷歌服务器的响应。

请注意您应该使用更快的网络服务器进行真正的测试，例如本地主机，因为网络延迟对结果有很大影响。

#include <QCoreApplication> 
#include <QTcpSocket> 
#include <QtConcurrent/QtConcurrentRun> 
#include <QElapsedTimer> 
#include <QAtomicInt> 

class Task : public QRunnable 
{ 
    public: 
     Task() : QRunnable() {} 
     static QAtomicInt counter; 
     static QElapsedTimer timer; 
     virtual void run() override 
     { 
      QTcpSocket* socket = new QTcpSocket(); 
      socket->connectToHost("www.google.com", 80); 
      socket->write("GET/HTTP/1.1\r\nHost: www.google.com\r\n\r\n"); 
      socket->waitForReadyRead(); 
      if(!--counter) { 
       qDebug("Multiple Threads elapsed: %lld nanoseconds", timer.nsecsElapsed()); 
      } 
     } 
}; 

QAtomicInt Task::counter; 
QElapsedTimer Task::timer; 

int main(int argc, char *argv[]) 
{ 
    QCoreApplication app(argc, argv); 

    // init 
    int connections = 10; 
    Task::counter = connections; 
    QElapsedTimer timer; 

    /// Async via One Thread (Select) 

    // handle the data 
    auto dataHandler = [&timer,&connections](QByteArray data) { 
     Q_UNUSED(data); 
     if(!--connections) qDebug(" Single Threads elapsed: %lld nanoseconds", timer.nsecsElapsed()); 
    }; 

    // create 10 connection to google.com and send an http get request 
    timer.start(); 
    for(int i = 0; i < connections; i++) { 
     QTcpSocket* socket = new QTcpSocket(); 
     socket->connectToHost("www.google.com", 80); 
     socket->write("GET/HTTP/1.1\r\nHost: www.google.com\r\n\r\n"); 
     QObject::connect(socket, &QTcpSocket::readyRead, [dataHandler,socket]() { 
      dataHandler(socket->readAll()); 
     }); 
    } 


    /// Async via Multiple Threads 

    Task::timer.start(); 
    for(int i = 0; i < connections; i++) { 
     QThreadPool::globalInstance()->start(new Task()); 
    } 

    return app.exec(); 
} 

打印：

Multiple Threads elapsed: 62324598 nanoseconds 
    Single Threads elapsed: 63613967 nanoseconds 

Answer 3

虽然，答案已经被接受了，我想和大家分享我的）

我从你的问题明白了什么：有256当前活动连接，您不时发送一个请求（命名为“command”）给其中一个请求并等待响应。同时，要进行这个过程中多线程和，虽然你说：“这需要阻塞，直到它获得响应”，我想你暗示阻止它处理请求 - 响应进程中的线程，但不是主线程。

如果我确实了解这个问题吧，这里是我建议使用Qt来做到这一点：

#include <functional> 

#include <QObject>   // need to add "QT += core" in .pro 
#include <QTcpSocket>  // QT += network 
#include <QtConcurrent>  // QT += concurrent 
#include <QFuture>   
#include <QFutureWatcher> 

class CommandSender : public QObject 
{ 
public: 
    // Sends a command via connection and blocks 
    // until the response arrives or timeout occurs 
    // then passes the response to a handler 
    // when the handler is done - unblocks 
    void SendCommand(
     QTcpSocket* connection, 
     const Command& command, 
     void(*responseHandler)(Response&&)) 
    { 
     const int timeout = 1000; // milliseconds, set it to -1 if you want no timeouts 

     // Sending a command (blocking) 
     connection.write(command.ToByteArray()); // Look QByteArray for more details 
     if (connection.waitForBytesWritten(timeout) { 
      qDebug() << connection.errorString() << endl; 
      emit error(connection); 
      return; 
     } 

     // Waiting for a response (blocking) 
     QDataStream in{ connection, QIODevice::ReadOnly }; 
     QString message; 
     do { 
      if (!connection.waitForReadyRead(timeout)) { 
       qDebug() << connection.errorString() << endl; 
       emit error(connection); 
       return; 
      } 
      in.startTransaction(); 
      in >> message; 
     } while (!in.commitTransaction()); 

     responseHandler(Response{ message }); // Translate message to a response and handle it 
    } 

    // Non-blocking version of SendCommand 
    void SendCommandAsync(
     QTcpSocket* connection, 
     const Command& command, 
     void(*responseHandler) (Response&&)) 
    { 
     QFutureWatcher<void>* watcher = new QFutureWatcher<void>{ this }; 
     connect(watcher, &QFutureWatcher<void>::finished, [connection, watcher]() 
     { 
      emit done(connection); 
      watcher->deleteLater(); 
     }); 

     // Does not block, 
     // emits "done" when finished 
     QFuture<void> future 
      = QtConcurrent::run(this, &CommandSender::SendCommand, connection, command, responseHandler); 
     watcher->setFuture(future); 
    } 

signals: 
    void done(QTcpSocket* connection); 
    void error(QTcpSocket* connection); 
} 

现在你可以使用从线程池取出一个单独的线程发送命令到插座：下引擎盖QtConcurrent::run()使用由Qt为您提供的QThreadPool的全局实例。该线程阻塞，直到它得到一个响应，然后用responseHandler来处理它。同时，管理所有命令和套接字的主线程将保持畅通。只需抓住done()信号即可知道响应已成功接收和处理。

有一点需要注意：异步版本只有在线程池中有空闲线程时才发送请求，否则就等待它。当然，这是任何线程池的行为（这正是这种模式的要点），但不要忘记这一点。

另外我写的没有Qt的代码很方便，所以可能会包含一些错误。

编辑：事实证明，这不是线程安全的，因为套接字在Qt中不可重入。

你可以做的是将一个互斥量与一个套接字关联，并在每次执行其功能时将其锁定。这可以很容易地创建QTcpSocket类的包装。请纠正我，如果我错了。