QTcpSocket：读写

Question

我知道可能已经有类似的问题被问过了，但我发现那些问题的回答只涵盖了非常具体的问题，而且我还没有想通。

在我的程序中，我创建了一个 QObject（称为 QPeer），它使用 QTcpSocket 与另一个这样的对象进行通信。QPeer 有一个插槽，接受一个 QByteArray 数据 (sendData(QByteArray))。该数组的全部内容被视为一个“消息”，并写入到套接字。我想做以下操作：每次写入一条消息时，我希望接收端的 QPeer 恰好触发其信号 dataReceived(QByteArray)，该 QByteArray 包含整个消息。（注意：所有信号/插槽，包括将 QPeer 与其套接字连接的私有信号/插槽以及公共信号/插槽，例如 sendData(QByteArray)，都会在必要时使用 Qt::QueuedConnection 进行序列化。）

我使用信号 QTcpSocket::readyRead() 进行从套接字异步读取。现在我知道我不能只在 sendData 中调用一次 QTcpSocket::write()，然后假设对于我执行的每次写入，位于另一侧的 QTcpSocket 产生了恰好一个 readyRead 信号。那我该怎么办呢？

这是我的想法，请告诉我这是否可行：

写操作：

void QPeer::sendData(QByteArray data)
{
    // TODO: write data.size() as raw int of exactly 4 bytes to socket
    const char *bytes = data.constData();
    int bytesWritten = 0;
    while (bytesWritten < data.size())
        bytesWritten += _socket->write(bytes + bytesWritten);
}

阅读：

现在，我想要read函数（连接到 QTcpSocket ::readyRead()）使用头部（指定消息长度的4字节整数），然后读取该数量的字节；接下来，以恰好这些字节emit dataReceived。我试图做到这一点时遇到了严重的困难。例如：如果已发出readyRead并且我可以读取消息的头部但无法读取指定数量的字节怎么办？或者如果仅部分收到标题怎么办？

1. 如何正确地将头（4字节int）写入套接字？

2. 如何正确实现read函数以使其按照我的意愿工作？

欢迎任何提示。谢谢！

- DiscobarMolokai

Answer 1

我曾经参与了一个项目，它能够做你所期望的事情。这里是我为我们的问题开发的解决方案，经过简化以便更易于理解：

修改后，支持服务器处理多个客户端连接。

Client.h:

#include <QtCore>
#include <QtNetwork>

class Client : public QObject
{
    Q_OBJECT
public:
    explicit Client(QObject *parent = 0);

public slots:
    bool connectToHost(QString host);
    bool writeData(QByteArray data);

private:
    QTcpSocket *socket;
};

Client.cpp：

#include "client.h"

static inline QByteArray IntToArray(qint32 source);

Client::Client(QObject *parent) : QObject(parent)
{
    socket = new QTcpSocket(this);
}

bool Client::connectToHost(QString host)
{
    socket->connectToHost(host, 1024);
    return socket->waitForConnected();
}

bool Client::writeData(QByteArray data)
{
    if(socket->state() == QAbstractSocket::ConnectedState)
    {
        socket->write(IntToArray(data.size())); //write size of data
        socket->write(data); //write the data itself
        return socket->waitForBytesWritten();
    }
    else
        return false;
}

QByteArray IntToArray(qint32 source) //Use qint32 to ensure that the number have 4 bytes
{
    //Avoid use of cast, this is the Qt way to serialize objects
    QByteArray temp;
    QDataStream data(&temp, QIODevice::ReadWrite);
    data << source;
    return temp;
}

Server.h：

#include <QtCore>
#include <QtNetwork>

class Server : public QObject
{
    Q_OBJECT
public:
    explicit Server(QObject *parent = 0);

signals:
    void dataReceived(QByteArray);

private slots:
    void newConnection();
    void disconnected();
    void readyRead();

private:
    QTcpServer *server;
    QHash<QTcpSocket*, QByteArray*> buffers; //We need a buffer to store data until block has completely received
    QHash<QTcpSocket*, qint32*> sizes; //We need to store the size to verify if a block has received completely
};

Server.cpp：

#include "server.h"

static inline qint32 ArrayToInt(QByteArray source);

Server::Server(QObject *parent) : QObject(parent)
{
    server = new QTcpServer(this);
    connect(server, SIGNAL(newConnection()), SLOT(newConnection()));
    qDebug() << "Listening:" << server->listen(QHostAddress::Any, 1024);
}

void Server::newConnection()
{
    while (server->hasPendingConnections())
    {
        QTcpSocket *socket = server->nextPendingConnection();
        connect(socket, SIGNAL(readyRead()), SLOT(readyRead()));
        connect(socket, SIGNAL(disconnected()), SLOT(disconnected()));
        QByteArray *buffer = new QByteArray();
        qint32 *s = new qint32(0);
        buffers.insert(socket, buffer);
        sizes.insert(socket, s);
    }
}

void Server::disconnected()
{
    QTcpSocket *socket = static_cast<QTcpSocket*>(sender());
    QByteArray *buffer = buffers.value(socket);
    qint32 *s = sizes.value(socket);
    socket->deleteLater();
    delete buffer;
    delete s;
}

void Server::readyRead()
{
    QTcpSocket *socket = static_cast<QTcpSocket*>(sender());
    QByteArray *buffer = buffers.value(socket);
    qint32 *s = sizes.value(socket);
    qint32 size = *s;
    while (socket->bytesAvailable() > 0)
    {
        buffer->append(socket->readAll());
        while ((size == 0 && buffer->size() >= 4) || (size > 0 && buffer->size() >= size)) //While can process data, process it
        {
            if (size == 0 && buffer->size() >= 4) //if size of data has received completely, then store it on our global variable
            {
                size = ArrayToInt(buffer->mid(0, 4));
                *s = size;
                buffer->remove(0, 4);
            }
            if (size > 0 && buffer->size() >= size) // If data has received completely, then emit our SIGNAL with the data
            {
                QByteArray data = buffer->mid(0, size);
                buffer->remove(0, size);
                size = 0;
                *s = size;
                emit dataReceived(data);
            }
        }
    }
}

qint32 ArrayToInt(QByteArray source)
{
    qint32 temp;
    QDataStream data(&source, QIODevice::ReadWrite);
    data >> temp;
    return temp;
}

注意：不要使用此方法传输大文件，因为此方法会在发送之前将整个消息内容放入内存中，这会导致高内存使用率。而且由于32位有符号INT的最大值为2,147,483,647，如果输入数据的大小超过该值（以字节为单位），它将无法正常工作。请注意。

- Antonio Dias

---

非常感谢！之前的答案是有效的，但这让我重新考虑了我的类设计。它变得有点...不好看了。 - DiscobarMolokai

信号将被发射，因为套接字是一个指针，它不会在作用域结束时被销毁，除非您调用 delete 或使用智能指针。但是，您不能再从该连接接收数据，因为 readyRead() 方法仅适用于存储在变量中的指针（最后连接的客户端）。要处理多个客户端，您需要创建包含自己的套接字指针和相应缓冲区的类的多个实例。 - Antonio Dias

我已经编辑了答案，以支持多个客户端，无需创建类的多个实例。 - Antonio Dias

顺便问一下，你的实现是否保证对各个套接字发出的信号进行序列化处理？例如，如果两个套接字同时从它们各自的对等方接收数据会发生什么？ - DiscobarMolokai

我的实现能够保证同时处理多个套接字，因为对于所有 readyRead() 信号，程序都会等待线程空闲来处理该信号，有时候需要等待接收数据，但数据最终会到来。 - Antonio Dias

Answer 2

正如你所说，你需要等待头部完全发送后才能读取它，然后读取足够的字节数并发出数据可用的信号。

以下是一个示例（未经测试）：

//header file

class Peer {
//[...]
protected:
   bool m_headerRead; //initialize to false
   unsigned int m_size_of_data_to_read;
//[...]
};

//source file
void QPeer::sendData(QByteArray data)
{
  int size = data.size();
  _socket->write((const char*) &size, sizeof(int);
  //use directly QIODevice::write(QByteArray)
  _socket->write(data);
}

void QPeer::readData()
{
    int bytes = _socket->bytesAvailable();
    bool contains_enough_data = true;

    while (contains_enough_data) {
       if (! m_headerRead && _socket->bytesAvailable() >= sizeof(int)) {
         //read header only and update m_size_of_data_to_read
         m_headerRead = true;
        } else if (m_headerRead && _socket->bytesAvailable >= m_size_of_data_to_read) {
          //read data here
          m_headerRead = false;
          emit dataAvailable();
       } else {
           contains_enough_data = false; //wait that further data arrived
       }
    }
}

- epsilon

---

那么你是说 _socket->write(data) 保证了整个 QByteArray 一次性写入？因为 Qt 文档中说：_qint64 QIODevice::write(const QByteArray & byteArray)这是一个重载函数。将 byteArray 的内容写入设备。返回实际写入的字节数，如果发生错误则返回 -1。所以它说它返回实际写入的字节数。如果整个数组总是一次性写入，那么为什么要这样做呢？ - DiscobarMolokai

好的，我在处理 QByteArray 并检查返回值为 -1 时也经常使用这种方法。你之前的做法也不错，你可以放心地用这个替换它。我会搜索一下是否总是完全写入 QByteArray。 - epsilon

我没有检查-1错误；但是我已经将套接字的错误信号连接到QPeer中处理错误的槽。这样足够吗？感谢您的帮助！您的实现很好，我还没有进行彻底测试，但目前看来很好，谢谢。 - DiscobarMolokai

我认为你不会因此类错误而收到错误信号。该信号更多地与独立错误有关（例如DNS故障、物理断开等），请参见AbstractSocket :: Error。这可能更安全地处理-1返回代码;) - epsilon