基于Qt实现高性能串口通讯应用程序，要求多线程，数据分包，高性能，高稳定性。-牛翰网

2. 核心模块实现

2.1 线程安全的串口管理类

class SerialPortWorker : public QObject {
    Q_OBJECT
public:
    explicit SerialPortWorker(QObject *parent = nullptr)
        : m_serial(new QSerialPort), m_buffer(1024*1024) // 1MB环形缓冲区
    {
        connect(m_serial, &QSerialPort::readyRead, 
            this, &SerialPortWorker::onDataReceived, Qt::DirectConnection);
    }

signals:
    void packetReady(QByteArray packet);
    void errorOccurred(QString err);

public slots:
    void startPort(const SerialSettings &settings) {
        QMutexLocker lock(&m_mutex);
        m_serial->setPortName(settings.portName);
        m_serial->setBaudRate(settings.baudRate);
        m_serial->setFlowControl(QSerialPort::HardwareControl);
        if(!m_serial->open(QIODevice::ReadWrite)) {
            emit errorOccurred(m_serial->errorString());
        }
        m_lastActiveTime = QDateTime::currentMSecsSinceEpoch();
    }

private slots:
    void onDataReceived() {
        const qint64 now = QDateTime::currentMSecsSinceEpoch();
        const QByteArray data = m_serial->readAll();
      
        // 写入环形缓冲区
        if(!m_buffer.write(data)) {
            emit errorOccurred("Buffer overflow");
            return;
        }

        // 分包处理（示例：固定头尾协议）
        while(m_buffer.size() >= 4) {
            if(m_buffer.peek(0) == 0xAA && m_buffer.peek(1) == 0x55) {
                const int length = m_buffer.peek(2);
                if(m_buffer.size() >= length + 4) {
                    QByteArray packet;
                    m_buffer.read(packet, length + 4);
                    if(packet.back() == 0x0D) { // 校验尾字节
                        emit packetReady(packet);
                    }
                }
            } else {
                m_buffer.skip(1); // 滑动窗口寻找包头
            }
        }

        m_lastActiveTime = now;
    }

private:
    QScopedPointer<QSerialPort> m_serial;
    RingBuffer m_buffer;
    QMutex m_mutex;
    qint64 m_lastActiveTime = 0;
};

2.2 高性能环形缓冲区实现

class RingBuffer {
public:
    explicit RingBuffer(size_t capacity) 
        : m_capacity(capacity), 
          m_buffer(new char[capacity]) {}
  
    bool write(const QByteArray &data) {
        if(data.size() > freeSpace()) return false;
      
        std::lock_guard<std::mutex> lock(m_mutex);
        const size_t tailSpace = m_capacity - m_tail;
        if(data.size() <= tailSpace) {
            memcpy(m_buffer.data() + m_tail, data.data(), data.size());
        } else {
            memcpy(m_buffer.data() + m_tail, data.data(), tailSpace);
            memcpy(m_buffer.data(), data.data() + tailSpace, data.size() - tailSpace);
        }
        m_tail = (m_tail + data.size()) % m_capacity;
        return true;
    }

    void read(QByteArray &out, size_t len) {
        std::lock_guard<std::mutex> lock(m_mutex);
        out.resize(len);
        if(m_head + len <= m_capacity) {
            memcpy(out.data(), m_buffer.data() + m_head, len);
        } else {
            const size_t firstPart = m_capacity - m_head;
            memcpy(out.data(), m_buffer.data() + m_head, firstPart);
            memcpy(out.data() + firstPart, m_buffer.data(), len - firstPart);
        }
        m_head = (m_head + len) % m_capacity;
    }

    size_t size() const { 
        return (m_tail >= m_head) ? 
            (m_tail - m_head) : 
            (m_capacity - m_head + m_tail);
    }

private:
    std::mutex m_mutex;
    std::unique_ptr<char[]> m_buffer;
    size_t m_capacity = 0;
    size_t m_head = 0;
    size_t m_tail = 0;
};

3. 线程管理与通信

3.1 线程初始化

// 主线程配置
SerialSettings settings {
    "/dev/ttyUSB0", 
    QSerialPort::Baud115200,
    QSerialPort::Data8,
    QSerialPort::NoParity,
    QSerialPort::OneStop
};

QThread serialThread;
SerialPortWorker worker;
worker.moveToThread(&serialThread);

// 错误处理和UI更新连接
connect(&worker, &SerialPortWorker::packetReady,
        this, &MainWindow::processPacket, 
        Qt::QueuedConnection);
connect(&worker, &SerialPortWorker::errorOccurred,
        this, &MainWindow::showError);

QTimer::singleShot(0, &worker, [=](){
    worker.startPort(settings);
});
serialThread.start();

3.2 数据写入线程

class WriteHandler : public QObject {
    Q_OBJECT
public slots:
    void enqueueWrite(const QByteArray &data) {
        QMutexLocker lock(&m_queueMutex);
        m_writeQueue.enqueue(data);
        if(!m_isWriting) {
            QTimer::singleShot(1, this, &WriteHandler::processQueue);
        }
    }

private slots:
    void processQueue() {
        QByteArray packet;
        {
            QMutexLocker lock(&m_queueMutex);
            if(m_writeQueue.isEmpty()) {
                m_isWriting = false;
                return;
            }
            packet = m_writeQueue.dequeue();
        }

        qint64 bytesWritten = 0;
        while(bytesWritten < packet.size()) {
            const qint64 ret = m_port->write(packet.constData() + bytesWritten, 
                                           packet.size() - bytesWritten);
            if(ret == -1) {
                emit writeFailed(m_port->errorString());
                return;
            }
            bytesWritten += ret;
            if(!m_port->waitForBytesWritten(100)) {
                emit timeout("Write operation timeout");
                return;
            }
        }

        QTimer::singleShot(1, this, &WriteHandler::processQueue);
    }

private:
    QSerialPort* m_port;
    QQueue<QByteArray> m_writeQueue;
    QMutex m_queueMutex;
    bool m_isWriting = false;
};

4. 性能优化要点

零拷贝设计
- 使用memcpy直接操作环形缓冲区内存
- 避免QByteArray的多次构造和析构

硬件加速

// 在Linux平台启用DMA加速
m_serial->setReadBufferSize(1024*1024); // 1MB硬件缓冲区
m_serial->setDevicePolicy(QSerialPort::FlushOnAccess);

实时性保障

// 设置线程优先级
QThread::currentThread()->setPriority(QThread::TimeCriticalPriority);
// 禁用系统缓冲
m_serial->setDataTerminalReady(true);
m_serial->setRequestToSend(true);

5. 稳定性保障措施

心跳检测机制

QTimer *watchdog = new QTimer(this);
connect(watchdog, &QTimer::timeout, [=](){
    if(QDateTime::currentMSecsSinceEpoch() - m_lastActiveTime > 1000) {
        restartPort(); // 自动重连
    }
});
watchdog->start(500);

错误恢复流程

void SerialPortWorker::handleError(QSerialPort::SerialPortError error) {
    if(error == QSerialPort::NoError) return;
  
    m_serial->clearError();
    if(error == QSerialPort::ResourceError) {
        QThread::msleep(1000);
        m_serial->close();
        if(!m_serial->open(QIODevice::ReadWrite)) {
            emit fatalError("Cannot reopen port");
        }
    }
}

数据完整性校验

bool verifyChecksum(const QByteArray &packet) {
    quint8 sum = 0;
    for(int i=2; i<packet.size()-1; ++i) {
        sum += static_cast<quint8>(packet[i]);
    }
    return (sum == static_cast<quint8>(packet.back()));
}