使用 QEvent 而非 信号和槽 进行高流量吞吐

taskbusplatform/core/tasknode.cpp

@@ -15,6 +15,8 @@
 #include "process_prctl.h"
 #include "../watchdog/tbwatchdog.h"
 #include "../watchdog/profile_log.h"
+
+QEvent::Type taskNode::m_nPackEvent = (QEvent::Type)QEvent::registerEventType();
 taskNode::taskNode(QObject *parent)
 	: QObject(parent)
 	,m_process(new QProcess(this))
@@ -38,8 +40,8 @@ taskNode::taskNode(QObject *parent)
 			&taskNode::slot_readyReadStandardError/*,Qt::QueuedConnection*/);
 	connect(m_process,&QProcess::bytesWritten,this,
 			&taskNode::slot_sended);
-	connect(this,&taskNode::sig_new_package,this,&taskNode::slot_write,Qt::QueuedConnection);
 	m_nBp_TimerID = startTimer(200);
+
 }
 taskNode::~taskNode()
 {
@@ -232,8 +234,8 @@ void taskNode::slot_readyReadStandardOutput()
 								emit sig_new_command(map_z);
 						}
 					}
-					else
-						emit_package(pack_header,pack_size);
+					else if (m_currPrj)
+						m_currPrj->routing_new_package(this,pack_header, pack_size);
 					if (m_bDebug)
 						log_package(true,pack_header,pack_size);
 					startByte += pack_size;
@@ -286,19 +288,6 @@ void taskNode::emit_message(QByteArray arred)
 	}
 }
 
-/*!
- * \brief taskNode::emit_package
- * * Signals and Slots is easy to use, but it will take too much cpu resource with out bufferring .
- * 没有缓存机制，频繁的发送信号，将降低系统的性能。
- * We will check the timestamp between 2 signals, catch it , and send it out as a batch list.
- * \param package
- */
-void taskNode::emit_package(char * package, qsizetype sz)
-{
-	if (m_currPrj)
-		m_currPrj->routing_new_package(this,package, sz);
-}
-
 void taskNode::flush_from_stderr()
 {
 	if (m_arr_Strerr.size())
@@ -380,10 +369,21 @@ bool taskNode::enqueue_write(const char * pack, qsizetype sz)
 	m_mtxPackage.unlock();
 	++m_spackage_recieved;
 	if (!z)
-		emit sig_new_package();
+	{
+		QCoreApplication::postEvent(this,new QEvent(m_nPackEvent),Qt::HighEventPriority);
+		//emit sig_new_package();
+	}
 	return  true;
 }
-void taskNode::slot_write()
+
+void taskNode::customEvent(QEvent * evt)
+{
+	if (evt->type()==m_nPackEvent)
+	{
+		flush_write();
+	}
+}
+void taskNode::flush_write()
 {
 	m_mtxPackage.lock();
 	if (m_bDebug)

taskbusplatform/core/tasknode.h

@@ -19,7 +19,9 @@
 #include <QByteArrayList>
 #include <QFile>
 #include <QMutex>
+#include <QEvent>
 class taskCell;
+
 class taskProject;
 /*!
  * \brief The taskNode class 管理一个构件进程的类
@@ -38,7 +40,6 @@ signals:
 	void sig_pro_started();
 	void sig_pro_stopped(int exitCode, QProcess::ExitStatus exitStatus);
 	void sig_new_command(QMap<QString, QVariant> cmd);
-	void sig_new_package();
 	void sig_new_errmsg(QByteArrayList);
 	void sig_iostat(qint64 pid,quint64 pr,quint64 ps,quint64 br, quint64 bs);
 public slots:
@@ -50,7 +51,6 @@ private slots:
 	void slot_readyReadStandardOutput();
 	void slot_readyReadStandardError();
 	void slot_sended(qint64 );
-	void slot_write();
 	void slot_started( );
 	void slot_stopped();
 private:
@@ -66,6 +66,7 @@ private:
 	void flush_from_stderr();
 private:
 	void log_package(bool fromStdOut,char * pkg, qsizetype sz);
+	void flush_write();
 	QFile m_dbgfile_stdin;
 	QFile m_dbgfile_stdout;
 	QFile m_dbgfile_stderr;
@@ -77,7 +78,6 @@ private:
 	QString dbgdir();
 	//Call these functions to emit message and packs with bufferring approach.
 	void emit_message(QByteArray a);
-	void emit_package(char * pack, qsizetype sz);
 public:
 	void setBlockFlag(bool b);
 	void setCurrentProject(taskProject * p);
@@ -85,8 +85,10 @@ public:
 protected:
 	int  m_nBatchTime = 20;
 	int  m_nBp_TimerID = -1;
+	static QEvent::Type	 m_nPackEvent ;
 	taskCell * m_pCell = nullptr;
 	void timerEvent(QTimerEvent *event);
+	void customEvent(QEvent * evt) override;
 	//statistic
 protected:
 	quint64 m_spackage_recieved = 0;