129 lines
3.2 KiB
C
129 lines
3.2 KiB
C
|
#ifndef QDATABUFFER_H
|
|||
|
|
#define QDATABUFFER_H
|
|||
|
|
|
|||
|
|
#include <QObject>
|
|||
|
|
#include <QQueue>
|
|||
|
|
#include <QMutex>
|
|||
|
|
#include <QWaitCondition>
|
|||
|
|
#include <QTimer>
|
|||
|
|
#include <QElapsedTimer>
|
|||
|
|
#include <QMap>
|
|||
|
|
#include <QSet>
|
|||
|
|
#include "BZStruct.h"
|
|||
|
|
|
|||
|
|
// 数据包类型
|
|||
|
|
enum class DataPacketType {
|
|||
|
|
RealTimeQuote,
|
|||
|
|
OrderBook,
|
|||
|
|
KLine,
|
|||
|
|
Tick,
|
|||
|
|
BrokerQueue,
|
|||
|
|
Unknown
|
|||
|
|
};
|
|||
|
|
|
|||
|
|
// 数据包结构
|
|||
|
|
struct DataPacket {
|
|||
|
|
DataPacketType type;
|
|||
|
|
QString stockCode;
|
|||
|
|
QDateTime timestamp;
|
|||
|
|
QVariant data;
|
|||
|
|
int priority; // 优先级 0-9，9为最高
|
|||
|
|
|
|||
|
|
DataPacket(DataPacketType t, const QString& code, const QVariant& d, int prio = 5)
|
|||
|
|
: type(t), stockCode(code), timestamp(QDateTime::currentDateTime()), data(d), priority(prio) {}
|
|||
|
|
};
|
|||
|
|
|
|||
|
|
// 流量统计
|
|||
|
|
struct FlowStatistics {
|
|||
|
|
qint64 totalPackets = 0;
|
|||
|
|
qint64 droppedPackets = 0;
|
|||
|
|
qint64 processedPackets = 0;
|
|||
|
|
double packetsPerSecond = 0.0;
|
|||
|
|
double dropRate = 0.0;
|
|||
|
|
QDateTime lastUpdate;
|
|||
|
|
|
|||
|
|
void updateStats(qint64 total, qint64 dropped, qint64 processed) {
|
|||
|
|
totalPackets = total;
|
|||
|
|
droppedPackets = dropped;
|
|||
|
|
processedPackets = processed;
|
|||
|
|
dropRate = total > 0 ? static_cast<double>(dropped) / total : 0.0;
|
|||
|
|
lastUpdate = QDateTime::currentDateTime();
|
|||
|
|
}
|
|||
|
|
};
|
|||
|
|
|
|||
|
|
class QDataBuffer : public QObject
|
|||
|
|
{
|
|||
|
|
Q_OBJECT
|
|||
|
|
public:
|
|||
|
|
static QDataBuffer* instance();
|
|||
|
|
|
|||
|
|
// 缓冲区配置
|
|||
|
|
void setBufferSize(int maxSize);
|
|||
|
|
void setFlowControlEnabled(bool enabled);
|
|||
|
|
void setMaxPacketsPerSecond(int maxRate);
|
|||
|
|
|
|||
|
|
// 数据入队
|
|||
|
|
bool enqueue(const DataPacket& packet);
|
|||
|
|
bool enqueueHighPriority(const DataPacket& packet);
|
|||
|
|
|
|||
|
|
// 数据出队
|
|||
|
|
DataPacket dequeue(int timeoutMs = 1000);
|
|||
|
|
QList<DataPacket> dequeueBatch(int maxCount, int timeoutMs = 1000);
|
|||
|
|
|
|||
|
|
// 状态查询
|
|||
|
|
bool isEmpty() const;
|
|||
|
|
int size() const;
|
|||
|
|
int getQueueSize(DataPacketType type) const;
|
|||
|
|
FlowStatistics getStatistics() const;
|
|||
|
|
|
|||
|
|
// 流量控制
|
|||
|
|
void pause();
|
|||
|
|
void resume();
|
|||
|
|
bool isPaused() const;
|
|||
|
|
|
|||
|
|
// 清理
|
|||
|
|
void clear();
|
|||
|
|
void clearByType(DataPacketType type);
|
|||
|
|
void clearByStock(const QString& stockCode);
|
|||
|
|
|
|||
|
|
signals:
|
|||
|
|
void bufferOverflow(const QString& stockCode, DataPacketType type);
|
|||
|
|
void flowControlActivated(bool activated);
|
|||
|
|
void statisticsUpdated(const FlowStatistics& stats);
|
|||
|
|
void packetDropped(const DataPacket& packet, const QString& reason);
|
|||
|
|
|
|||
|
|
private:
|
|||
|
|
explicit QDataBuffer(QObject *parent = nullptr);
|
|||
|
|
~QDataBuffer();
|
|||
|
|
|
|||
|
|
void updateStatistics();
|
|||
|
|
bool shouldDropPacket(const DataPacket& packet) const;
|
|||
|
|
void cleanupExpiredPackets();
|
|||
|
|
|
|||
|
|
// 缓冲区
|
|||
|
|
QQueue<DataPacket> m_buffer;
|
|||
|
|
QQueue<DataPacket> m_highPriorityBuffer;
|
|||
|
|
|
|||
|
|
// 同步控制
|
|||
|
|
mutable QMutex m_mutex;
|
|||
|
|
QWaitCondition m_notEmpty;
|
|||
|
|
QWaitCondition m_notFull;
|
|||
|
|
|
|||
|
|
// 配置
|
|||
|
|
int m_maxBufferSize = 10000;
|
|||
|
|
int m_maxPacketsPerSecond = 1000;
|
|||
|
|
bool m_flowControlEnabled = true;
|
|||
|
|
bool m_paused = false;
|
|||
|
|
|
|||
|
|
// 统计
|
|||
|
|
FlowStatistics m_stats;
|
|||
|
|
QElapsedTimer m_rateTimer;
|
|||
|
|
qint64 m_currentSecondPackets = 0;
|
|||
|
|
|
|||
|
|
// 清理定时器
|
|||
|
|
QTimer* m_cleanupTimer;
|
|||
|
|
QTimer* m_statsUpdateTimer;
|
|||
|
|
};
|
|||
|
|
|
|||
|
|
#endif // QDATABUFFER_H
|