以下是一个可用的基本实现。它使用内部信号槽对将数据推送到另一个端点。这样，连接的任一端都可以存在于任何线程中，并且两端可以在线程之间移动，而不会丢失数据或引发任何竞争。
私人的QRingBuffer用于代替重新发明轮子。添加QT += core-private to the .pro文件以使其可用。 Qt PIMPL 用于访问 Qt 5.7 及更高版本中的内部设备缓冲区。
如果您希望实例化一个开放管道（通常是这种情况），您可以将 I/O 模式传递给构造函数。典型用途：
int main(/*…*/)
{
/*…*/
AppPipe end1 { QIODevice::ReadWrite };
AppPipe end2 { &end1, QIODevice::ReadWrite };
AppPipe end3 { &end1, QIODevice::ReadOnly };
// the pipes are open ready to use
/*…*/
}
无论您向一个管道写入什么内容，最终都会在另一个连接的管道中作为可读数据，反之亦然。在上面的例子中，数据写入end1两者均可读取end2 and end3独立。数据写入end2可以从以下位置读取end1. end3实际上是一个只听管道。连接额外的管道很便宜 - 向多个管道发送大块数据不会产生 O(N) 成本，因为读取QRingBuffer连接的管道存储从原始管道发送的整个字节数组的浅副本。
所有的QIODevice语义保持 - 您可以连接到readyRead信号，使用带有QDataStream or a QTextStream等。与任何QIODevice，您只能使用其中的类thread()，但另一个端点可以存在于任何线程中，并且两者都可以根据需要在线程之间移动，而不会丢失数据。
如果管道的另一端未打开且可读，则即使写入成功，写入也不会执行。关闭管道会清除读写缓冲区，以便可以重新打开以供重用。
管道缓冲区默认写入数据，可以使用以下命令强制刷新写入缓冲区AppPipe::flush()，除非打开于QIODevice::Unbuffered mode.
The hasIncoming and hasOutgoing信号可用于监控 https://stackoverflow.com/questions/32486198通过管道传输的数据。
// https://github.com/KubaO/stackoverflown/tree/master/questions/local-pipe-32317081
// This project is compatible with Qt 4 and Qt 5
#include <QtTest>
#include <private/qiodevice_p.h>
#include <private/qringbuffer_p.h>
#include <algorithm>
#include <climits>
#ifndef Q_DECL_OVERRIDE
#define Q_DECL_OVERRIDE
#endif
class AppPipePrivate : public QIODevicePrivate {
public:
#if QT_VERSION < QT_VERSION_CHECK(5,7,0)
QRingBuffer buffer;
QRingBuffer writeBuffer;
int writeBufferChunkSize;
#endif
const QByteArray *writeData;
AppPipePrivate() : writeData(0) { writeBufferChunkSize = 4096; }
};
/// A simple point-to-point intra-process pipe. The other endpoint can live in any
/// thread.
class AppPipe : public QIODevice {
Q_OBJECT
Q_DECLARE_PRIVATE(AppPipe)
static inline int intLen(qint64 len) { return std::min(len, qint64(INT_MAX)); }
Q_SLOT void _a_write(const QByteArray &data) {
Q_D(AppPipe);
if (!(d->openMode & QIODevice::ReadOnly)) return; // We must be readable.
d->buffer.append(data); // This is a chunk shipped from the source.
emit hasIncoming(data);
emit readyRead();
}
void hasOutgoingLong(const char *data, qint64 len) {
while (len) {
int const size = intLen(len);
emit hasOutgoing(QByteArray(data, size));
data += size;
len -= size;
}
}
public:
AppPipe(QIODevice::OpenMode mode, QObject *parent = 0) :
QIODevice(*new AppPipePrivate, parent) {
open(mode);
}
AppPipe(AppPipe *other, QIODevice::OpenMode mode, QObject *parent = 0) :
QIODevice(*new AppPipePrivate, parent) {
open(mode);
addOther(other);
}
AppPipe(AppPipe *other, QObject *parent = 0) :
QIODevice(*new AppPipePrivate, parent) {
addOther(other);
}
~AppPipe() Q_DECL_OVERRIDE {}
void addOther(AppPipe *other) {
if (other) {
connect(this, SIGNAL(hasOutgoing(QByteArray)), other, SLOT(_a_write(QByteArray)), Qt::UniqueConnection);
connect(other, SIGNAL(hasOutgoing(QByteArray)), this, SLOT(_a_write(QByteArray)), Qt::UniqueConnection);
}
}
void removeOther(AppPipe *other) {
disconnect(this, SIGNAL(hasOutgoing(QByteArray)), other, SLOT(_a_write(QByteArray)));
disconnect(other, SIGNAL(hasOutgoing(QByteArray)), this, SLOT(_a_write(QByteArray)));
}
void flush() {
Q_D(AppPipe);
while (!d->writeBuffer.isEmpty()) {
QByteArray const data = d->writeBuffer.read();
emit hasOutgoing(data);
emit bytesWritten(data.size());
}
}
void close() Q_DECL_OVERRIDE {
Q_D(AppPipe);
flush();
QIODevice::close();
d->buffer.clear();
}
qint64 write(const QByteArray &data) { // This is an optional optimization. The base method works OK.
Q_D(AppPipe);
QScopedValueRollback<const QByteArray*> back(d->writeData);
if (!(d->openMode & Text))
d->writeData = &data;
return QIODevice::write(data);
}
qint64 writeData(const char *data, qint64 len) Q_DECL_OVERRIDE {
Q_D(AppPipe);
bool buffered = !(d->openMode & Unbuffered);
if (buffered && (d->writeBuffer.size() + len) > d->writeBufferChunkSize)
flush();
if (!buffered
|| len > d->writeBufferChunkSize
|| (len == d->writeBufferChunkSize && d->writeBuffer.isEmpty()))
{
if (d->writeData && d->writeData->data() == data && d->writeData->size() == len)
emit hasOutgoing(*d->writeData);
else
hasOutgoingLong(data, len);
}
else
memcpy(d->writeBuffer.reserve(len), data, len);
return len;
}
bool isSequential() const Q_DECL_OVERRIDE { return true; }
Q_SIGNAL void hasOutgoing(const QByteArray &);
Q_SIGNAL void hasIncoming(const QByteArray &);
#if QT_VERSION >= QT_VERSION_CHECK(5,7,0)
// all the data is in the read buffer already
qint64 readData(char *, qint64) Q_DECL_OVERRIDE { return 0; }
#else
qint64 readData(char *data, qint64 len) Q_DECL_OVERRIDE {
Q_D(AppPipe);
qint64 hadRead = 0;
while (len && !d->buffer.isEmpty()) {
int size = d->buffer.read(data, intLen(len));
hadRead += size;
data += size;
len -= size;
}
return hadRead;
}
bool canReadLine() const Q_DECL_OVERRIDE {
Q_D(const AppPipe);
return d->buffer.indexOf('\n') != -1 || QIODevice::canReadLine();
}
qint64 bytesAvailable() const Q_DECL_OVERRIDE {
Q_D(const AppPipe);
return QIODevice::bytesAvailable() + d->buffer.size();
}
qint64 bytesToWrite() const Q_DECL_OVERRIDE {
Q_D(const AppPipe);
return QIODevice::bytesToWrite() + d->writeBuffer.size();
}
#endif
};
// ...
#include "main.moc"
最小的测试工具：
class TestAppPipe : public QObject {
Q_OBJECT
QByteArray data1, data2;
struct PipePair {
AppPipe end1, end2;
PipePair(QIODevice::OpenMode mode = QIODevice::NotOpen) :
end1(QIODevice::ReadWrite | mode), end2(&end1, QIODevice::ReadWrite | mode) {}
};
Q_SLOT void initTestCase() {
data1 = randomData();
data2 = randomData();
}
Q_SLOT void sizes() {
QCOMPARE(sizeof(AppPipe), sizeof(QIODevice));
}
Q_SLOT void basic() {
PipePair p;
QVERIFY(p.end1.isOpen() && p.end1.isWritable() && p.end1.isReadable());
QVERIFY(p.end2.isOpen() && p.end2.isWritable() && p.end2.isReadable());
static const char hello[] = "Hello There!";
p.end1.write(hello);
p.end1.flush();
QCOMPARE(p.end2.readAll(), QByteArray(hello));
}
static QByteArray randomData(int const size = 1024*1024*32) {
QByteArray data;
data.resize(size);
char *const d = data.data();
for (char *p = d+data.size()-1; p >= d; --p)
*p = qrand();
Q_ASSERT(data.size() == size);
return data;
}
static void randomChunkWrite(AppPipe *dev, const QByteArray &payload) {
for (int written = 0, left = payload.size(); left; ) {
int const chunk = std::min(qrand() % 82931, left);
dev->write(payload.mid(written, chunk));
left -= chunk; written += chunk;
}
dev->flush();
}
void runBigData(PipePair &p) {
Q_ASSERT(!data1.isEmpty() && !data2.isEmpty());
randomChunkWrite(&p.end1, data1);
randomChunkWrite(&p.end2, data2);
QCOMPARE(p.end1.bytesAvailable(), qint64(data2.size()));
QCOMPARE(p.end2.bytesAvailable(), qint64(data1.size()));
QCOMPARE(p.end1.readAll(), data2);
QCOMPARE(p.end2.readAll(), data1);
}
Q_SLOT void bigDataBuffered() {
PipePair p;
runBigData(p);
}
Q_SLOT void bigDataUnbuffered() {
PipePair p(QIODevice::Unbuffered);
runBigData(p);
}
Q_SLOT void cleanupTestCase() {
data1.clear(); data2.clear();
}
};
QTEST_MAIN(TestAppPipe)
# local-pipe-32317081.pro
QT = core
greaterThan(QT_MAJOR_VERSION, 4): QT = core-private testlib
else: CONFIG += qtestlib
DEFINES += \
QT_DEPRECATED_WARNINGS \
QT_DISABLE_DEPRECATED_BEFORE=0x060000 \
QT_RESTRICTED_CAST_FROM_ASCII
CONFIG += console c++14
CONFIG -= app_bundle
TEMPLATE = app
SOURCES = main.cpp
