Zoom-Pipeline--基于Qt的高伸缩TCP服务实现

摘要：

本文介绍了一种基于Qt的高伸缩TCP服务实现。该实现针对C/S客户端-服务集群应用需求而搭建。连接监听、数据传输、数据处理均在独立的线程池中进行，根据特定任务不同，可安排负责监听、传输、处理的线程数目，从而在高传输负荷、高计算符合上达成取舍。数据处理采用流水线结构，以避免少量客户的密集计算请求影响其他客户端的处理。本文对应的代码符合LGPL协议，可直接从Github签出。

关键词：Qt; TCP/IP; 流水线; 并发; 架构; 开源

Zoom-Pipeline--基于Qt的高伸缩TCP服务实现	1

1	综述	3

1.1	系统结构	3

1.2	主要指标	5

2	网络传输模块	6

2.1	模块结构	6

2.2	系统原理	8

2.2.1	监听器与监听线程	8

2.2.2	接受连接过程	10

2.2.3	数据接收	12

2.2.4	数据发送	13

2.2.5	终止连接过程	15

3	流水线结构线程池设计	16

3.1	模块结构	17

3.1.1	zp_pipeline类	17

3.1.2	zp_plWorkingThread类	18

3.1.3	zp_plTaskbase类	19

3.2	工作原理	19

3.2.1	任务入队	19

3.2.2	任务执行	20

4	数据库的多线程操作	21

4.1	线程内注册与连接数据库的竞争问题	21

4.2	数据库连接意外断裂后，恢复连接的问题	22

4.3	数据库插件的依赖性问题	22

4.4	Qt 数据库特性	22

5	单层无中心集群	22

5.1	跨服务器传输通道	23

5.1.1	集群指令	23

5.1.2	连接流程	25

5.2	流式解析	26

5.2.1	数据接收	27

5.2.2	数据处理	28

5.3	集群模块外部接口	31

6	整合各个模块实现功能	32

6.1	类结构	32

6.2	客户端哈希	34

6.3	沟通信号与槽	34

6.4	集群应用层实现	36

6.5	均衡建议	38

7	结尾	39

本文用图：

图 1‑1 系统结构总体示意图	5

图 2‑1 network 模块的类合作关系图	7

图 3‑1 传统的跟随者模式示意图	16

图 3‑2线程池模式示意图	16

图 3‑3 带有流水线的线程池结构	17

图 3‑4 流水线线程池类结构合作关系	18

图 5‑1 集群设置界面	23

图 6‑1 范例服务应用层类合作模式图	33

本文用表：

表 1‑1 系统指标一览表	5

综述

在许多行业应用中，客户端-服务架构不可替代。在客户端-服务架构下，服务扮演着重要的角色。本文旨在介绍一种较为通用的服务实现形式，这个形式可以根据具体应用需求灵活配置，从而减少重复开发的时间。该服务实现至少支持下列两种需求。

第一类是转发型服务，主要负责沟通两个客户端之间的通信，本身不做复杂的处理。这种服务一般要承受密集的连接，在某些应用中，还要承受沉重的传输负荷。典型的包括即时通信、两个内网之间的视频通信等。

第二类是计算型服务，主要负责接收客户端的原始数据，处理后把结果返回。这种服务一般运行在高性能服务器上，调用后台的并行计算、异构计算资源，处理后把结果返回给客户端。这些服务的客户端数量一般不多，但在某些应用中，当输入输出数据量大时，传输负荷也很重。

为了满足上述两类需求，服务设计具备如下特点。

可灵活设置监听端口

监听、传输、处理均在独立的线程池中运行。

可根据需要，灵活设置传输、处理线程的数目。

对某个单一客户端的处理，不会显著拖慢所有客户端的响应。

需要支持分布式的集群服务，典型的情况是基于若干物理计算机构成的高速局域网实现进程集群。

可同时支持SSL与普通TCP连接。

这些特点决定了本系统的架构设计。

系统结构

服务由以下几个模块组成.

网络传输模块。负责管理用于监听、传输的套接字，并控制数据流在不同线程中流动。数据收发由一定规模的线程池负责，实现方法完全得益于Qt的线程事件循环。被绑定到某个Qthread上的Qobject对象，其信号-槽事件循环由该线程负责。这样，便可方便的指定某个套接字对象使用的线程。同样，受惠于Qt的良好封装，直接支持Tcp套接字及SSL套接字，且在运行时可动态调整。（注：编译这个模块需要Qt的SSL支持，即在 configure 时加入 -openssl 选项）

任务流水线模块。负责数据的处理。在计算密集型的应用中，数据处理负荷较重，需要和网络传输划分开。基于普通线程池的处理模式，也存在队列阻塞的问题——若干个客户端请求的耗时操作，阻塞了其他客户端的响应，哪怕其他客户端的请求很短时间就能处理完毕，也必须排队等待。采用流水线线程池避免了这个问题。每个客户端把需要做的操作进行粒度化，在一个环形的队列中，线程池对单个客户端，每次仅处理一个粒度单位的任务。单个粒度单位完成后，该客户端的剩余任务便被重新插入到队列尾部。这个机制保证了客户端的整体延迟较小。

服务集群管理模块。该模块使用了网络传输模块、任务流水线模块的功能，实现了跨进程的服务器((服务器链路。在高速局域网中，连接是快速、稳定的。因此，该模块被设计成一种星型无中心网络。任意新增服务器节点选择现有服务器集群中的任意一个节点，接入后，通过广播自动与其他服务器节点建立点对点连接。本模块只是提供一个服务器到服务器的通信隧道，不负责具体通信内容的解译。对传输内容的控制，由具体应用决定。

数据库管理模块。该模块基于Qt的插件式数据库封装 QtSql。数据库被作为资源管理，支持在多线程的条件下，使用数据库资源。

框架界面。尽管常见的服务运行时表现为一个后台进程，但为了更好的演示服务器的功能，避免繁琐的配置，还是需要一个图形界面来显示状态、设置参数。本范例中，界面负责轮训服务器的各个状态，并设置参数。设置好的参数被存储在一个ini文件中，并在服务开启时加载。

应用专有部分模块。上述1-4共四个主要模块均是通用的。他们互相之间没有形成联系，仅仅是作为一种资源存在于程序的运行时（Runtime）之中。应用专有部分模块根据具体任务需求，灵活的使用上述资源，以实现功能。在范例代码中，实现了一种点对点的转发机制。演示者虚拟出一些工业设备，以及一些操作员使用的客户端软件。设备与客户端软件在成功认证并登录后，需要交换数据。改变这个模块的代码，即可实现自己的功能。

本文结构便是按照上述6个模块的次序展开。本文使用的代码可从https://github.com/goldenhawking/zpserver 下载。

系统总体结构如下图所示：

图 1‑1 系统结构总体示意图

主要指标

系统的主要指标如下。

表 1‑1 系统指标一览表

项目	指标说明	指标值	备注
支持的连接类型	取决于Qt的支持类型	SSL和Plain	运行时同时支持两类.
协议
Ipv4 TCP , IPV6 Tcp
设计单进程连接数	在4\8传输线程配置下计算	512\1024	取决于实际机器的性能和任务形式.
设计集群规模	星型集群的规模受到节点数影响。	100节点，9900个跨服务器连接。	局域网中的连接数为节点数*（节点数-1）。
设计总用户规模	100节点、每节点1024个用户	10万

网络传输模块

模块对应代码命名空间	(namespace ZPNetwork)

模块对应代码存储文件夹	(\ZoomPipeline_FuncSvr\network)

模块结构

网络传输模块负责管理监听器，并根据各个传输线程目前的负荷，把新申请接入的客户套接字描述符引导到最空闲的传输线程中执行“接受连接（Accept）”操作。该模块由如下几个类组成。

1、zp_net_Engine 类，派生自Qobject。模块的外部接口类，同时也是功能管理者。提供了设置监听器、配置线程池的功能。

2、zp_netListenThread类：派生自QObject。用于绑定在各个监听线程的事件循环中，不断的接受客户端连接请求。该类会在信号中把套接字描述符（socket descriptor）泵出，由zp_net_Engine类进行负荷均衡，选取当前负荷最小的传输线程（zp_netTransThread）接受该接入申请。

3、zp_netTransThread类：派生自QObject。用于绑定在各个传输线程的事件循环中，具体承担数据传输。一个zp_netTransThread线程可以承担多个客户端的收发请求。

4、ZP_TcpServer类：派生自QtcpServer。重载了ZP_TcpServer::incomingConnection，不在监听线程中进行Accept操作，而是直接发出evt_NewClientArrived信号，把套接字描述符（socket descriptor）泵出，由zp_net_Engine类进行负荷均衡，选取当前负荷最小的传输线程（zp_netTransThread）接受该接入申请。

这四个类的合作关系图如下：

图 2‑1 network 模块的类合作关系图

系统原理

为了提供基于线程池的TCP服务，zp_net_engine类有几个重要成员。下面，按照一次客户端发起连接的过程，逆向的逐一来介绍这些类的合作原理.

监听器与监听线程

1、监听器ZP_TcpServer

系统运行时，负责监听工作的是 QtcpServer 派生类，名称叫ZP_TcpServer。该类重载了 QtcpServer的incomingConnection()方法 。当网络中一个客户端发起连接时，这个函数会被立刻调用。在本派生类中，并没有直接产生套接字。它仅仅触发了一个称为“evt_NewClientArrived”的信号 。这个信号把套接字描述符泵出给接受者，用于在其他的线程中创建套接字所用。其流程见2.2.2节所述。

2、监听器线程对象zp_netListenThread

ZP_TcpServer类的实例具体是由zp_netListenThread类中一个指针 m_tcpServer操作的。m_tcpServer是一个指向ZP_TcpServe类实例的指针（参见zp_netlistenthread.h ）。该实例在zp_netListenThread::startListen()中创建。StartListen是一个关键的函数，创建了ZP_TcpServer对象。核心代码如下：

m_tcpServer = new ZP_TcpServer(this);

connect (m_tcpServer,&ZP_TcpServer::evt_NewClientArrived,this,&zp_netListenThread::evt_NewClientArrived,Qt::QueuedConnection);

上面两行代码中，第一行创建一个监听服务，第二行，把监听服务的evt_NewClientArrived事件直接和zp_netListenThread 的 同名事件连接起来。

3、操作监听器的模块接口类zp_net_Engine

zp_netListenThread类本身是从Qobject派生。它本身不是一个线程对象，而是被“绑定”在一个线程对象中运行的。一个进程可以拥有若干监听端口，这些监听端口对应了不同的zp_netListenThread对象。这些监听线程对象由zp_net_Engine类管理，存储在这个类的成员变量中。下面两个成员变量

//This map stores listenThreadObjects

QMap<QString,zp_netListenThread *> m_map_netListenThreads;

//Internal Threads to hold each listenThreadObjects' message Queue

QMap<QString,QThread *> m_map_netInternalListenThreads;

第一个存储了各个端口的线程对象，第二个存储了各个端口的线程。

由于具体下达监听任务的线程是主线程（UI），但执行任务的线程是工作线程，所以，所有的指令均不是通过直接的函数调用来实现，取而代之的是使用Qt的信号与槽。比如，UI按钮被点击，则触发了startListen 信号，转而由zp_netListenThread的startListen槽来响应。这里需要注意的是，由于Qt的信号与槽系统是一种广播系统，意味着一个zp_net_Engine类管理多个zp_netListenThread对象时，zp_net_Engine发出的信号会被所有zp_netListenThread对象接收。因此，信号与槽中含有一个唯一标示，用于指示本次信号触发是为了操作具体哪个对象。这种技术在类似的场合被多次使用。

void zp_net_Engine::AddListeningAddress(QString id,const QHostAddress & address , quint16 nPort,bool bSSLConn /*= true*/)

{

if (m_map_netListenThreads.find(id)==m_map_netListenThreads.end())

{

//Start Thread

QThread * pThread = new QThread(this);

zp_netListenThread * pListenObj = new zp_netListenThread(id,address,nPort,bSSLConn);

pThread->start();

//m_mutex_listen.lock();

m_map_netInternalListenThreads[id] = pThread;

m_map_netListenThreads[id] = pListenObj;

//m_mutex_listen.unlock();

//Bind Object to New thread

connect(this,&zp_net_Engine::startListen,pListenObj,&zp_netListenThread::startListen,Qt::QueuedConnection);

connect(this,&zp_net_Engine::stopListen,pListenObj,&zp_netListenThread::stopListen,Qt::QueuedConnection);

connect(pListenObj,&zp_netListenThread::evt_Message,this,&zp_net_Engine::evt_Message,Qt::QueuedConnection);

connect(pListenObj,&zp_netListenThread::evt_ListenClosed,this,&zp_net_Engine::on_ListenClosed,Qt::QueuedConnection);

connect(pListenObj,&zp_netListenThread::evt_NewClientArrived,this,&zp_net_Engine::on_New_Arrived_Client,Qt::QueuedConnection);

pListenObj->moveToThread(pThread);

//Start Listen Immediately

emit startListen(id);

}

else

emit evt_Message(this,"Warning>"+QString(tr("This ID has been used.")));

}

接受连接过程

客户端发起接入请求后，首先触发了ZP_TcpServer的incomingConnection方法。在下面这个方法中，套接字的描述符作为事件的参数被泵出。

void ZP_TcpServer::incomingConnection(qintptr socketDescriptor)

{

emit evt_NewClientArrived(socketDescriptor);

}

上面的信号对应的槽为zp_net_Engine::on_New_Arrived_Client槽函数。在这个函数中，网络模块首先从当前可用的传输线程中确定最空闲的那个线程,而后把套接字描述符转交给传输线程。这个部分的核心代码：

void zp_net_Engine::on_New_Arrived_Client(qintptr socketDescriptor)

{

zp_netListenThread * pSource = qobject_cast<zp_netListenThread *>(sender());

if (!pSource)

{

emit evt_Message(this,"Warning>"+QString(tr("Non-zp_netListenThread type detected.")));

return;

}

emit evt_Message(this,"Info>" + QString(tr("Incomming client arriverd.")));

int nsz = m_vec_NetTransThreads.size();

int nMinPay = 0x7fffffff;

int nMinIdx = -1;

for (int i=0;i<nsz && nMinPay!=0;i++)

{

if (m_vec_NetTransThreads[i]->isActive()==false ||

m_vec_NetTransThreads[i]->SSLConnection()!=pSource->bSSLConn()

)

continue;

int nPat = m_vec_NetTransThreads[i]->CurrentClients();

if (nPat<nMinPay)

{

nMinPay = nPat;

nMinIdx = i;

}

//qDebug()<<i<<" "<<nPat<<" "<<nMinIdx;

}

//...

if (nMinIdx>=0 && nMinIdx<nsz)

emit evt_EstablishConnection(m_vec_NetTransThreads[nMinIdx],socketDescriptor);

else

{

emit evt_Message(this,"Warning>"+QString(tr("Need Trans Thread Object for clients.")));

}

}

上面的代码中, evt_EstablishConnection 事件携带了由均衡策略确定的承接线程、socketDescriptor 描述符。这个事件广播给所有的传输线程对象。在各个对象的incomingConnection槽中,具体生成用于传输的套接字对象.注意, 这个槽函数是运行在各个传输线程的事件循环中的,因此,创建的套接字直接属于特定线程.

/**

* @brief This slot dealing with multi-thread client socket accept.

* accepy works start from zp_netListenThread::m_tcpserver, end with this method.

* the socketDescriptor is delivered from zp_netListenThread(a Listening thread)

* to zp_net_Engine(Normally in main-gui thread), and then zp_netTransThread.

*

* @param threadid if threadid is not equal to this object, this message is just omitted.

* @param socketDescriptor socketDescriptor for incomming client.

*/

void zp_netTransThread::incomingConnection(QObject * threadid,qintptr socketDescriptor)

{

if (threadid!=this)

return;

QTcpSocket * sock_client = 0;

if (m_bSSLConnection)

sock_client = new QSslSocket(this);

else

sock_client = new QTcpSocket(this);

if (sock_client)

{

//Initial content

if (true ==sock_client->setSocketDescriptor(socketDescriptor))

{

connect(sock_client, &QTcpSocket::readyRead,this, &zp_netTransThread::new_data_recieved,Qt::QueuedConnection);

connect(sock_client, &QTcpSocket::disconnected,this,&zp_netTransThread::client_closed,Qt::QueuedConnection);

connect(sock_client, SIGNAL(error(QAbstractSocket::SocketError)),this, SLOT(displayError(QAbstractSocket::SocketError)),Qt::QueuedConnection);

connect(sock_client, &QTcpSocket::bytesWritten, this, &zp_netTransThread::some_data_sended,Qt::QueuedConnection);

m_mutex_protect.lock();

m_clientList[sock_client] = 0;

m_mutex_protect.unlock();

if (m_bSSLConnection)

{

QSslSocket * psslsock = qobject_cast<QSslSocket *>(sock_client);

assert(psslsock!=NULL);

QString strCerPath = QCoreApplication::applicationDirPath() + "/svr_cert.pem";

QString strPkPath = QCoreApplication::applicationDirPath() + "/svr_privkey.pem";

psslsock->setLocalCertificate(strCerPath);

psslsock->setPrivateKey(strPkPath);

connect(psslsock, &QSslSocket::encrypted,this, &zp_netTransThread::on_encrypted,Qt::QueuedConnection);

psslsock->startServerEncryption();

}

emit evt_NewClientConnected(sock_client);

emit evt_Message(sock_client,"Info>" + QString(tr("Client Accepted.")));

}

else

sock_client->deleteLater();

}

}

数据接收

在成功创建了套接字后, 数据的收发都在传输线程中运行了.当套接字收到数据后,简单的触发事件

evt_Data_recieved

void zp_netTransThread::new_data_recieved()

{

QTcpSocket * pSock = qobject_cast<QTcpSocket*>(sender());

if (pSock)

{

QByteArray array = pSock->readAll();

int sz = array.size();

g_mutex_sta.lock();

g_bytesRecieved +=sz;

g_secRecieved += sz;

g_mutex_sta.unlock();

emit evt_Data_recieved(pSock,array);

}

}

数据发送

尽管Qt的套接字本身具备缓存,塞入多大的数据都会成功, 但是本实现仍旧使用额外的队列, 每次缓存一个固定长度的片段并顺序发送. 这样的好处,是可以给代码使用者一个机会,来加入代码检查缓冲区的大小,并作一些持久化的工作. 比如,队列超过100MB后,就把后续的数据缓存在磁盘上, 而不是继续放在内存中,实现这个策略的变量是两个缓存.

//sending buffer, hold byteArraies.

QMap<QObject *,QList<QByteArray> > m_buffer_sending;

QMap<QObject *,QList<qint64> > m_buffer_sending_offset;

第一个缓存存储各个套接字的队列.另一个存储各个数据块的发送偏移. 这样做是有性能缺陷的, 更好的办法是从 QTcpSocket 派生自己的类,并把各个套接字的缓存直接存储在派生类实例中去. 在本实现中, 直接使用了 QTcpSocket和QSSLSocket类, 因而有一定的性能损失.

一个槽方法 SendDataToClient 负责接受发送数据的请求.

void zp_netTransThread::SendDataToClient(QObject * objClient,QByteArray dtarray)

{

m_mutex_protect.lock();

if (m_clientList.find(objClient)==m_clientList.end())

{

m_mutex_protect.unlock();

return;

}

m_mutex_protect.unlock();

QTcpSocket * pSock = qobject_cast<QTcpSocket*>(objClient);

if (pSock&&dtarray.size())

{

QList<QByteArray> & list_sock_data = m_buffer_sending[pSock];

QList<qint64> & list_offset = m_buffer_sending_offset[pSock];

if (list_sock_data.empty()==true)

{

qint64 bytesWritten = pSock->write(dtarray.constData(),qMin(dtarray.size(),m_nPayLoad));

if (bytesWritten < dtarray.size())

{

list_sock_data.push_back(dtarray);

list_offset.push_back(bytesWritten);

}

}

else

{

list_sock_data.push_back(dtarray);

list_offset.push_back(0);

}

}

}

在上面的函数中,将检查队列是否为空.为空的话,将触发 QTcpSocket::write方法发出m_nPayload大小的数据块.当这些数据块发送完毕,将触发QTcpSocket::bytesWritten事件,由下面的槽响应.

/**

* @brief this slot will be called when internal socket successfully

* sent some data. in this method, zp_netTransThread object will check

* the sending-queue, and send more data to buffer.

*

* @param wsended

*/

void zp_netTransThread::some_data_sended(qint64 wsended)

{

g_mutex_sta.lock();

g_bytesSent +=wsended;

g_secSent += wsended;

g_mutex_sta.unlock();

QTcpSocket * pSock = qobject_cast<QTcpSocket*>(sender());

if (pSock)

{

emit evt_Data_transferred(pSock,wsended);

QList<QByteArray> & list_sock_data = m_buffer_sending[pSock];

QList<qint64> & list_offset = m_buffer_sending_offset[pSock];

while (list_sock_data.empty()==false)

{

QByteArray & arraySending = *list_sock_data.begin();

qint64 & currentOffset = *list_offset.begin();

qint64 nTotalBytes = arraySending.size();

assert(nTotalBytes>=currentOffset);

qint64 nBytesWritten = pSock->write(arraySending.constData()+currentOffset,qMin((int)(nTotalBytes-currentOffset),m_nPayLoad));

currentOffset += nBytesWritten;

if (currentOffset>=nTotalBytes)

{

list_offset.pop_front();

list_sock_data.pop_front();

}

else

break;

}

}

}

终止连接过程

在传输终止后, 会进行一定的清理. 对于多线程的传输,最重要的是确保各个对象的生存期. 有兴趣的读者可以使用 sharedptr来管理动态分配的对象, 这样操作起来会很方便. 在本范例中, 所有代码均进行了 7*24 调试.

流水线结构线程池设计

为了无阻塞地实现并发通信及处理，传统的小规模服务器采用每用户一线程的多线程技术，称为“任务伴随者”模式。该模式示意图如下：

图 3‑1 传统的跟随者模式示意图

然而，当客户端很多时，开启上百组线程，远远超过计算机的物理线程规模，导致大量计算资源浪费在线程上下文切换和环境恢复等维护工作中，有效计算能力显著降低。

在多线程并行计算技术中，能够有效利用CPU物理核心，避免上下文频繁切换的经典模式是线程池模式。系统仅开启与CPU核心数相等的工作线程，形成线程池（ThreadPool）。各个任务在队列中排队，按照先进先出次序（FIFO）送入线程池中处理。该模式的示意图如下：

图 3‑2线程池模式示意图

该模式尽管避免了任务量较大时，实际计算能力降低的问题，但单位时间内仅能处理一定数量的任务，存在阻塞的可能。万一现在处理的几个任务均很耗时，则其他客户的简单任务也会被耽误。

在非阻塞的情况下利用线程池模式有效处理大量用户数据，要靠基于线程池的流水线技术，实现最优线程配置条件下低阻塞处理。

该模式的关键是对每个客户的处理任务进行细化，比如每K个指令为一个粒度单位，无论该客户的指令队列缓存了多少指令，一次仅处理K个，随后让出计算资源分配给其他任务使用。示意图如下：

图 3‑3 带有流水线的线程池结构

采用该方法，流水线结构保证了各个客户工作在大粒度上并行化，线程池技术保证了处理器资源的最大利用，可以显著提高系统的吞吐能力。另一个附加好处，是可以让VIP获得高级优先级。

模块结构

在范例代码中，本模块的所有文件位于 pipeline 文件夹下。

命名空间：ZPTaskEngine

主要有三个类组成。

zp_pipeline类

该类是流水线线程池的接口类。其管理了各个执行者线程，以及任务队列。执行者线程存储在本类的成员变量中。

//working threads

QVector<zp_plWorkingThread *> m_vec_workingThreads;

QVector<QThread *> m_vec_InternalworkingThreads;

通过方法addThreads可以控制线程池的规模。

图 3‑4 流水线线程池类结构合作关系

zp_plWorkingThread类

这个类是用于执行任务的线程对象。在其被创建时，绑定在一个QThread的线程事件循环中运行。创建的方法位于 zp_pipeline::addThreads中，

/**

* @brief Add nThreads to the thread pool

*

* @fn zp_pipeline::addThreads

* @param nThreads how many threads you want to add.

* @return int current threads count after add.

*/

int zp_pipeline::addThreads(int nThreads)

{

if (nThreads>=1 && nThreads <=128)

{

for (int i=0;i<nThreads;i++)

{

zp_plWorkingThread * thread = new zp_plWorkingThread(this);

m_vec_workingThreads.push_back(thread);

QThread * pTh = new QThread(this);

m_vec_InternalworkingThreads.push_back(pTh);

thread->moveToThread(pTh);

connect (this,&zp_pipeline::evt_start_work,thread,&zp_plWorkingThread::FetchNewTask,Qt::QueuedConnection);

connect (this,&zp_pipeline::evt_stop_work,thread,&zp_plWorkingThread::setStopMark,Qt::QueuedConnection);

connect (thread,&zp_plWorkingThread::taskFinished,this,&zp_pipeline::on_finished_task,Qt::QueuedConnection);

pTh->start();

m_mutex_protect.lock();

m_nExistingThreads++;

m_mutex_protect.unlock();

}

}

return m_vec_workingThreads.size();

}

zp_plTaskbase类

本类是一个纯虚基类，用于给应用者重载具体的执行任务。该类的核心方法是 run()，用于在线程池的某个线程中运行。

工作原理

任务入队

当外部需要执行任务时，调用 zp_pipeline::pushTask方法，向任务队列m_list_tasks中传入zp_plTaskbase类型的指针。一旦队列中被插入了新任务，会立刻判断是否有空闲的线程可以执行这个任务。如果有，立刻触发执行。核心代码:

void zp_pipeline::pushTask(zp_plTaskBase * task,bool bFire )

{

m_mutex_protect.lock();

m_list_tasks.push_back(task);

task->addRef();

m_mutex_protect.unlock();

int nsz = m_vec_workingThreads.size();

if (bFire==true)

for (int i=0;i<nsz;i++ )

{

if (m_vec_workingThreads[i]->m_bBusy==false)

{

on_finished_task (m_vec_workingThreads[i]);

break;

}

}

}

任务执行

zp_pipeline:: on_finished_task 槽既是任务的起始，也是任务的结束。当某个zp_plWorkingThread对象执行完了一次任务，便会触发本方法。在本方法中，zp_pipeline对象检查自己的队列，看看是否还有任务需要执行。如果有，则读入一个任务继续执行.

void zp_pipeline::on_finished_task (zp_plWorkingThread * task)

{

int res = 0;

m_mutex_protect.lock();

res = m_list_tasks.size();

m_mutex_protect.unlock();

if (res)

emit evt_start_work(task );

}

触发执行任务是使用事件 evt_start_work 触发的，这个信号发给 task线程，使得它在自己的槽函数中获取新的任务。

/**

* @brief Call zp_plTaskBase::popTask to fetch new tasks.

*

* @fn zp_plWorkingThread::FetchNewTask

* @param obj the zp_plWorkingThread object recieved by signal-slot system.

* this method will omit zp_plWorkingThread objs except for it self.

*/

void zp_plWorkingThread::FetchNewTask(zp_plWorkingThread * obj)

{

if (obj != this)

return;

if (m_bRuning)

{

bool bValid = false;

zp_plTaskBase * ptr = this->m_pipeline->popTask(&bValid);

if (bValid==true && ptr!=NULL)

{

m_bBusy = true;

if (ptr->LockRun()==true)

{

int res = ptr->run();

ptr->delRef();

if (res!=0 )

this->m_pipeline->pushTask(ptr,false);

ptr->UnlockRun();

}

else

{

ptr->delRef();

this->m_pipeline->pushTask(ptr,false);

}

m_bBusy = false;

}

emit taskFinished(this);

}

}

在FetchNewTask槽中，会调用 zp_pipeline::popTask方法，弹出一个任务。一旦任务弹出，则会调用虚函数 run() 运行。run()的返回结果为0， 表示本任务彻底完成了，不再塞入队列的尾部；如果返回值非0， 说明任务还没有完成，只是执行了一部分，任务自己就释放了资源，防止阻塞整个流程。在这个情况下，任务被重新 push回队列。

数据库的多线程操作

在多线程服务中，数据库是必不可少的组成部分。Qt通过 QtSql模块支持数据库的通用访问。通过插件，可支持目前主流的数据库系统。Qt的数据库操作注意事项在我前面的博文中已经有了涉及，这里简要重复一遍。

线程内注册与连接数据库的竞争问题

Qt文档上对多线程下数据库应用的注意事项写的很简明，一个线程创建的 QSqlDatabase 对象和 查出来的 QSqlQuery 对象只能给本线程用（注意，是对象，不是数据库连接本身，连接本身用名字可以多线程使用），其他情况是“不支持的”。在一个需要有几个线程并发访问不同数据库的应用中，试图在各个线程的起始分别以不同的名称调用 addDatabase / database 、open，程序偶然会崩溃，跟踪后发现，虽然Qt 声称很多方法是“线程安全”的，但是几个方法串起来，就出问题了。Qt 会动态的加载数据库的plugin, 加载 plug in 的部分，涉及到对本地库文件的管理，这一部分，出现了竞争。在初始连接部分设置 Mutex 保护，从 addDatabase / database到 open 的部分，要保证其原子性。

数据库连接意外断裂后，恢复连接的问题

在MFC 中，一旦中途TCP连接断裂，直接重新 Open 就可以了。在Qt 里，这一招不好使了。即便 调用了 close ，再次open 也是不行的。处理方法：

在检测到问题出现后，关闭连接，并 removeDatabase； 而后，不要立刻 addDatabase, 反而是要回到该连接所在的事件循环。没有详细跟源码，很可能在 removeDatabase 后的事件循环中，Qt 内部做了一些释放操作。 实际应用中，直接显式调用processEvent() 方法强制循环。 在多线程下，注意1中的问题，需要 Mutex保护。

数据库插件的依赖性问题

在 Windows 下，有时我们的机器上按了好几个 Qt 版本，依赖开发环境的继承环境适应不同的版本。这有两个问题。一是发布程序的时候，数据库驱动依赖的dll 也要与可执行文件在同一路径下发布。比如 mysql 的 dll, PostgreSQL 的依赖等。二是在集成开发环境中，这些依赖也要位于执行档文件夹下。否则，会造成虽然可以枚举到可用驱动，但是连接不上。调试一下就知道，原来是在路径中找不到依赖项，导致dll加载失败！

Qt 数据库特性

Qt的数据库操作自成一派，相对于复杂的 ADO \ODBC\DAO\OLEDB 等传统 C++ 访问数据库的方法，还是很先进的，充分体现了 OO 的理念。设计者把进程内的数据库连接作为一种资源，每个连接有一个唯一的名字，可以通过全局的 addDatabase, removeDatabase, cloneDatabase 来增删，想用的时候，直接用全局的 database 来获取。这样的好处，是大大节省了开发者的负担。以前为了传递一个数据库连接的变量，必须在很多方法入库处添加指向这个变量的指针或者引用，有时候不得不在对象的属性中加入静态的变量，来记录这个连接。现在，什么时候想用，给个名字就可以了，不需要传递。当然，文档说的是比较简化的，至少有两点要注意，

1）这些增删方法号称是线程安全的，但是，在实际应用中，还是要注意用 Mutex 保护全局创建流程，或者，重载这些函数，创建自己的安全版本。

2）一个线程创建的数据库对象（如 addDatabase 的返回值）只能在同一线程使用，但是，addDatabase 注册的连接（名字是开发者定）可以跨线程使用，唯一需要注意的是，在调用全局方法的时候，要有原子保护。

单层无中心集群

对40万用户规模以内的服务器，使用星形的无中心连接是较为简便的实现方式。分布在各个物理服务器上的服务进程共同工作，每个进程承担若干连接。为了实现这个功能，需要解决几个关键问题。

跨服务器传输通道

设计在高速局域网中的连接可直接采用TCP，并用第二章介绍的网络传输工具、第三章介绍的流水线线程池共同搭建。引用上述两个工具的代码在cluster子文件夹的 zp_clusterterm.h中定义：

ZPNetwork::zp_net_Engine * m_pClusterNet;

ZPTaskEngine::zp_pipeline * m_pClusterEng;

服务器在专用的集群网络中监听，需要参数如下：

1、监听的地址、端口

2、本节点唯一名称

3、对服务器集群内其他节点发布的连接端口、地址

4、对公网客户端发布的连接端口、地址。

比如，服务器高速局域网网段可能是 10.129.XX.XX，而有些服务器可能以虚拟机(192.168.11.XX)+NAT(10.129.XX.XX)的方式在内网的子网中映射，因此，需要告诉别的服务器节点，如何连接到自己。同时，对公网客户端来说，每个服务器的连接地址又不同了。很有可能也是通过NAT的方式，把数十个内网IP映射到一个外网出口的连续端口上去。这个策略的配置页面如下：

图 5‑1 集群设置界面

集群的连接策略是，新的服务器进程选取任意一个现有节点，连接后，通过集群内广播系统自动接收其它各个节点的地址，并继续发起连接，直到与现有节点两两相通为止。

为了支持这个策略，集群传输需要定义一些指令。

集群指令

集群指令在 cluster 文件夹的cross_svr_message.h 定义:

#ifndef CROSS_SVR_MESSAGES_H

#define CROSS_SVR_MESSAGES_H

#include <qglobal.h>

namespace ZP_Cluster{

#pragma pack (push,1)

typedef struct tag_cross_svr_message{

struct tag_header{

quint16 Mark; //Always be 0x1234

quint8 messagetype;

quint32 data_length;

} hearder;

union uni_payload{

quint8 data[1];

struct tag_CSM_heartBeating{

quint32 nClients;

} heartBeating;

struct tag_CSM_BasicInfo{

quint8 name [64];

quint8 Address_LAN[64];

quint16 port_LAN;

quint8 Address_Pub[64];

quint16 port_Pub;

} basicInfo;

struct tag_CSM_Broadcast{

quint8 name [64];

quint8 Address_LAN[64];

quint16 port_LAN;

quint8 Address_Pub[64];

quint16 port_Pub;

} broadcastMsg[1];

} payload;

} CROSS_SVR_MSG;

#pragma pack(pop)

}

#endif // CROSS_SVR_MESSAGES_H

指令由头部、载荷两部分组成。

头部header说明：

Mark是一个固定的起始，用于验证流解译的正确性。如果流解译不正确，第二块指令的起始将不是这个值。

messagetype 是一个用来标定指令类型的字节，决定了载荷联合体该采用哪个策略解译

data_length是长度，这里代表载荷的长度

载荷 payload说明：

有三种指令结构体， 心跳结构体用来维持各个服务器之间的心跳，基本信息（basicInfo)用于在连接建立后，向对方告知本节点的信息。广播结构体是用于在本机的服务器列表发生变更时，向所有现有节点广播新的列表。

对传输的用户数据，直接存储在data中。

连接流程

第一步，准备加入集群的服务器选取集群中任一个节点作为对象，发起P2P连接。

第二步，双方互换信息(basicInfo)

第三步，双方将对方的信息添加到本地的服务器节点表中。服务器节点表是一群zp_ClusterNode类的实例，该类由ZPTaskEngine::zp_plTaskBase派生。这个基类在第三章有介绍。服务器节点对象的实例负责具体的指令解译。该列表如下（在cluster子文件夹的 zp_clusterterm.h中定义)：

//important hashes. server name to socket, socket to server name

QMutex m_hash_mutex;

QMap<QString , zp_ClusterNode *> m_hash_Name2node;

QMap<QObject *,zp_ClusterNode *> m_hash_sock2node;

节点的指针存放在映射中，一个是名称到对象的映射，一个是套接字到对象的映射

第四步，由于节点表发生变化，因此，会触发对现有节点的广播(broadCasting)

第五步，各个节点收到广播后，会比较广播中的节点信息和自己目前的节点信息，并发起向新增节点的连接。

最终，当一对一连接完成，系统重新处于稳定状态。解译这段信息的代码片段在中cluster文件夹zp_clusternode.cpp的deal_current_message_block方法中实现:

switch(m_currentHeader.messagetype)

{

\\...

case 0x01://basicInfo, when connection established, this message should be used

if (bytesLeft==0)

{

QString strName ((const char *)pMsg->payload.basicInfo.name);

if (strName != m_pTerm->name())

{

this->m_strTermName = strName;

m_nPortLAN = pMsg->payload.basicInfo.port_LAN;

m_addrLAN = QHostAddress((const char *)pMsg->payload.basicInfo.Address_LAN);

m_nPortPub = pMsg->payload.basicInfo.port_Pub;

m_addrPub = QHostAddress((const char *)pMsg->payload.basicInfo.Address_Pub);

if (false==m_pTerm->regisitNewServer(this))

{

this->m_strTermName.clear();

emit evt_Message(this,tr("Info: New Svr already regisited. Ignored.")+strName);

emit evt_close_client(this->sock());

}

else

{

emit evt_NewSvrConnected(this->termName());

m_pTerm->BroadcastServers();

}

}

else

{

emit evt_Message(this,tr("Can not connect to it-self, Loopback connections is forbidden."));

emit evt_close_client(this->sock());

}

}

break;

case 0x02: //Server - broadcast messages

if (bytesLeft==0)

{

int nSvrs = pMsg->hearder.data_length / sizeof(CROSS_SVR_MSG::uni_payload::tag_CSM_Broadcast);

for (int i=0;i<nSvrs;i++)

{

QString strName ((const char *)pMsg->payload.broadcastMsg[i].name);

if (strName != m_pTerm->name() && m_pTerm->SvrNodeFromName(strName)==NULL)

{

QHostAddress addrToConnectTo((const char *)pMsg->payload.broadcastMsg[i].Address_LAN);

quint16 PortToConnectTo = pMsg->payload.broadcastMsg[i].port_LAN;

if (strName > m_pTerm->name())

emit evt_connect_to(addrToConnectTo,PortToConnectTo,false);

else

emit evt_Message(this,tr("Name %1 <= %2, omitted.").arg(strName).arg(m_pTerm->name()));

}

}

}

break;

...

流式解析

TCP 是面向连接的流式传输。对用户发送的一个大数据包，虽然保证收发的完整性，旦接收方每次接收的数据片段长度是有限的，也是不定的。一种简单的思路是按照指令结构体的长度，直接缓存完整的数据包，而后集中处理。这样有一个问题，在数据包很大时，内存开销过高。因此，本应用设计的思路是边接收、边处理。具体步骤：

1、检查收到的头部是否合法

2、存储当前指令的头部

3、一旦得到一段载荷数据，就回调一次处理过程，处理过程根据需求等待更多数据，或者处理完后清空缓存。这对一次传输100MB数据的应用是很关键的。流式处理需要完成的步骤关键代码如下：

数据接收

在 zp_ClusterTerm的接收槽里，直接把数据片段压入zp_ClusterNode对象的队列中，并压入流水线。

//some data arrival

void zp_ClusterTerm::on_evt_Data_recieved(QObject * clientHandle,QByteArray datablock )

{

//Push Clients to nodes if it is not exist

zp_ClusterNode * pClientNode = ...;

int nblocks = pClientNode->push_new_data(datablock);

if (nblocks<=1)

m_pClusterEng->pushTask(pClientNode);

//...

}

pushTask方法把Block压入zp_ClusterNode的处理队列m_list_Rawdata里，这部分的状态变量如下：

class zp_ClusterNode : public ZPTaskEngine::zp_plTaskBase

{

//.....

//Data Process

//The raw data queue and its mutex

QList<QByteArray> m_list_RawData;

QMutex m_mutex_rawData;

//The current Read Offset, from m_list_RawData's beginning

int m_currentReadOffset;

//Current Message Offset, according to m_currentHeader

int m_currentMessageSize;

//Current un-procssed message block.for large blocks,

//this array will be re-setted as soon as some part of data has been

//dealed, eg, send a 200MB block, the 200MB data will be splitted into pieces

QByteArray m_currentBlock;

CROSS_SVR_MSG::tag_header m_currentHeader;

//...

};

变量 m_currentReadOffset 指的是队列的首部元素已经处理的偏移。比如首部的Block有2341字节，处理了1099字节，本指令已经结束，则此值为1099

变量 m_currentMessageSize 指的是当前接收的信息的大小。比如100MB 的信息，接受了23MB，这个值就是23MB

变量 m_currentBlock 是当前的缓存。这个缓存会不断的递交处理，负责处理的程序可以根据情况适时清空它。对短指令，不清也是可以的。

变量 m_currentHeader 是当前的信息头部，这个值记录了当前结构体的首部信息。

数据处理

在线程池中，会调用 zp_ClusterNode::run 虚拟方法。这个方法的关键代码如下(实际代码因为有线程同步，要复杂一些）：

int zp_ClusterNode::run()

{

//nMessageBlockSize 是静态变量，表示最多处理几个块就释放CPU给其他节点

int nMessage = m_nMessageBlockSize;

int nCurrSz = -1;

while (--nMessage>=0 && nCurrSz!=0 )

{

QByteArray block;

block = *m_list_RawData.begin();

m_currentReadOffset = filter_message(block,m_currentReadOffset);

if (m_currentReadOffset >= block.size())

{

m_list_RawData.pop_front();

m_currentReadOffset = 0;

}

nCurrSz = m_list_RawData.size();

}

if (nCurrSz==0)

return 0;

return -1;

}

其中，filter_message 是对信息进行初步处理，输入当前队列的首部、处理偏移，返回新的处理偏移

这个方法的关键代码如下：

//!deal one message, affect m_currentRedOffset,m_currentMessageSize,m_currentHeader

//!return bytes Used.

int zp_ClusterNode::filter_message(QByteArray block, int offset)

{

const int blocklen = block.length();

while (blocklen>offset)

{

const char * dataptr = block.constData();

//先确保信息的头标志被接收

while (m_currentMessageSize<2 && blocklen>offset )

{

m_currentBlock.push_back(dataptr[offset++]);

m_currentMessageSize++;

}

if (m_currentMessageSize < 2) //First 2 byte not complete

continue;

if (m_currentMessageSize==2)

{

const char * headerptr = m_currentBlock.constData();

memcpy((void *)&m_currentHeader,headerptr,2);

}

const char * ptrCurrData = m_currentBlock.constData();

//判断头2个字节是不是1234

if (m_currentHeader.Mark == 0x1234)

//Valid Message

{

//试图接收完整的头部信息

if (m_currentMessageSize< sizeof(CROSS_SVR_MSG::tag_header) && blocklen>offset)

{

int nCpy = sizeof(CROSS_SVR_MSG::tag_header) - m_currentMessageSize;

if (nCpy > blocklen - offset)

nCpy = blocklen - offset;

QByteArray arrCpy(dataptr+offset,nCpy);

m_currentBlock.push_back(arrCpy);

offset += nCpy;

m_currentMessageSize += nCpy;

}

//如果头部还没收完则返回

if (m_currentMessageSize < sizeof(CROSS_SVR_MSG::tag_header)) //Header not completed.

continue;

//除了头部以外，还有数据可用，并且头部刚刚接收完

else if (m_currentMessageSize == sizeof(CROSS_SVR_MSG::tag_header))//Header just completed.

{

//保存头部

const char * headerptr = m_currentBlock.constData();

memcpy((void *)&m_currentHeader,headerptr,sizeof(CROSS_SVR_MSG::tag_header));

//继续处理后续的载荷

if (block.length()>offset)

{

//确定还有多少字节没有接收

qint32 bitLeft = m_currentHeader.data_length + sizeof(CROSS_SVR_MSG::tag_header)

-m_currentMessageSize ;

//继续接收载荷

if (bitLeft>0 && blocklen>offset)

{

int nCpy = bitLeft;

if (nCpy > blocklen - offset)

nCpy = blocklen - offset;

QByteArray arrCpy(dataptr+offset,nCpy);

m_currentBlock.push_back(arrCpy);

offset += nCpy;

m_currentMessageSize += nCpy;

bitLeft -= nCpy;

}

//处理一次数据

deal_current_message_block();

if (bitLeft>0)

continue;

//This Message is Over. Start a new one.

m_currentMessageSize = 0;

m_currentBlock = QByteArray();

continue;

}

}

//除了头部以外，还有数据可用

else

{

if (block.length()>offset)

{

//确定还有多少字节没有接收

qint32 bitLeft = m_currentHeader.data_length + sizeof(CROSS_SVR_MSG::tag_header)

-m_currentMessageSize ;

//继续接收载荷

if (bitLeft>0 && blocklen>offset)

{

int nCpy = bitLeft;

if (nCpy > blocklen - offset)

nCpy = blocklen - offset;

QByteArray arrCpy(dataptr+offset,nCpy);

m_currentBlock.push_back(arrCpy);

offset += nCpy;

m_currentMessageSize += nCpy;

bitLeft -= nCpy;

}

//deal block, may be processed as soon as possible;

deal_current_message_block();

if (bitLeft>0)

continue;

//This Message is Over. Start a new one.

m_currentMessageSize = 0;

m_currentBlock = QByteArray();

continue;

}

} // end if there is more bytes to append

} //end deal trans message

else

//...

} // end while block len > offset

return offset;

}

在处理当前块数据的方法 deal_current_message_block里，即可逐一判断消息类型，加以处理了。

集群模块外部接口

集群模块只负责在服务器之间建立连接，并提供一套传输用户数据的通路。在集群建立连接后，用户直接通过

void zp_ClusterTerm::SendDataToRemoteServer(QString svrName,QByteArray SourceArray)

{

int nMsgLen = sizeof(CROSS_SVR_MSG::tag_header) + SourceArray.size();

QByteArray array(nMsgLen,0);

CROSS_SVR_MSG * pMsg =(CROSS_SVR_MSG *) array.data();

pMsg->hearder.Mark = 0x1234;

pMsg->hearder.data_length = SourceArray.size();

pMsg->hearder.messagetype = 0x03;

memcpy (pMsg->payload.data,SourceArray.constData(),SourceArray.size());

m_hash_mutex.lock();

if (m_hash_Name2node.contains(svrName))

netEng()->SendDataToClient(m_hash_Name2node[svrName]->sock(),array);

m_hash_mutex.unlock();

}

向服务器 svrName发送 SourceArray， 并响应

void evt_RemoteData_recieved(QString /*svrHandle*/,QByteArray /*svrHandle*/ );

信号来接收数据。用户不用关心传输协议的封装和解析。

但是，以下问题是不涉及的。

1、传输的数据的具体意义解释

2、全局客户端的UUID哈希和同步

3、客户端数据是否被真正接收。

这些部分留给应用相关部分来具体实现。

整合各个模块实现功能

在前面的章节中，介绍了网络传输、任务线程池、数据库和集群四个主要功能模块。到现在为止，这些模块都还只是一种资源，没有产生实际的运行效果。对一个具备真实功能的应用来说，需要有一个整合的过程。整合方法很多，这里以典型的客户 -客户通信来举例说明。

类结构

1、“客户端” 这个概念被抽象为一个节点类st_clientNode，每个客户端连接对应了该类的一个实例。这个类不但存储了有关该连接的所有背景信息（比如聊天程序中的用户名等），还提供了正确解释数据流的代码实现。由于想分开传输层和应用层的代码，实际例子中，该类被分为基础传输类st_clientNode_baseTrans和应用层类st_clientNodeAppLayer两部分。

2、 对没有在本服务器登录的客户端来说，通过集群的服务器-服务器传输链路实现跨服务器通信。“其他服务器”这个概念被抽象为一个节点类st_cross_svr_node，集群内每个服务器连接对应了该类的一个实例。这个类不但存储了有关该连接的所有背景信息（比如服务器的名字、地址等），还提供了正确解释数据流的代码实现。该类实现了三种集群消息类型。

(1) 新客户端登入广播，用于通知集群内所有服务器，有新客户端登入

(2) 客户端退出广播。

(3) 客户端数据流打包传输

3、在最上层，有一个本服务器进程的管理者，称作st_client_table，用于封装所有的服务功能。这个类在每个服务器进程中仅有一个实例。它主要的工作有：

(1) 提供一个盛放、管理各个客户端节点类（st_clientNodeAppLayer实例）、各个集群服务器节点类(st_cross_svr_node实例)的容器；

(2) 提供一个管理本地客户端哈希表的执行者

(3) 提供一个管理全局客户端哈希表的执行者

(4) 把网络传输、任务线程池、数据库和集群四个主要功能模块的信号、槽全部关联起来

(5) 提供负载均衡建议, 在本服务器满员后，建议客户端到集群内最空闲的服务器进程登录。

这些类的合作关系如下：

图 6‑1 范例服务应用层类合作模式图

客户端哈希

客户端哈希的目的是为了迅速通过用户名、套接字找到对应的套接字和客户端对象。他们在st_client_table类中定义。为了在较大规模下，仍然获得较好的效率，使用了STL的 unordered_map类。使用基于树的map、QMap也是可以的，但是访问效率不如哈希快。

//st_client_table 本地客户端Hash 成员

QMutex m_hash_mutex;

std::unordered_map<quint32,st_clientNode_baseTrans *> m_hash_uuid2node;

std::unordered_map<QObject *,st_clientNode_baseTrans *> m_hash_sock2node;

//st_client_table 远程客户端Hash 成员

std::unordered_map<quint32,QString> m_hash_remoteClient2SvrName;

QMutex m_mutex_cross_svr_map;

*本地哈希成员存储了从用户UUID(相当于用户的全局标示)到用户节点的映射。使用UUID可迅速得到节点指针，从而执行发送数据、踢出节点、修改信息等功能

*本地哈希成员存储了从套接字到用户节点的映射。使用套接字可迅速得到节点指针。这样，在收到数据时，可直接定位到用户节点。之所以没有从QTcpSocket派生一个子类，存放从套接字到用户节点的映射，是因为套接字的生存周期控制复杂，客户端频繁的登入、登出，一旦套接字对象删除失效，程序中其他位置使用套接字指针直接获取客户端节点指针的操作就会溢出。

*远程哈希成员存储了从客户端 UUID到集群服务器名称的映射。用于向远程客户端发送数据使用。

沟通信号与槽

为了把本文涉及的所有模块沟通起来，st_client_table类做了很多工作。

首先，网络模块、集群会发出信号，他们由st_client_table类响应并处理或移交。

//连接新用户来到信号，将分配新的用户节点、登记哈希表(对一般的Sock)、广播集群信息

connect (m_pThreadEngine,&ZPNetwork::zp_net_Engine::evt_NewClientConnected,this,&st_client_table::on_evt_NewClientConnected,Qt::QueuedConnection);

//连接新用户已保护信号，将分配新的用户节点、登记哈希表（对SSL Sock)

connect (m_pThreadEngine,&ZPNetwork::zp_net_Engine::evt_ClientEncrypted,this,&st_client_table::on_evt_ClientEncrypted,Qt::QueuedConnection);

//连接新用户断开信号，将清除哈希表、广播集群信息

connect (m_pThreadEngine,&ZPNetwork::zp_net_Engine::evt_ClientDisconnected,this,&st_client_table::on_evt_ClientDisconnected,Qt::QueuedConnection);

//连接用户数据到来信号，将直接向各个客户端的处理队列中push ,以待线程池解译客户端消息

connect (m_pThreadEngine,&ZPNetwork::zp_net_Engine::evt_Data_recieved,this,&st_client_table::on_evt_Data_recieved,Qt::QueuedConnection);

//连接用户数据成功发送信号，目前只用于统计流量

connect (m_pThreadEngine,&ZPNetwork::zp_net_Engine::evt_Data_transferred,this,&st_client_table::on_evt_Data_transferred,Qt::QueuedConnection);

//连接新集群节点接入信号，将向该节点发送HELLO包，并交换持有的客户端情况

connect (m_pCluster,&ZP_Cluster::zp_ClusterTerm::evt_NewSvrConnected,this,&st_client_table::on_evt_NewSvrConnected,Qt::QueuedConnection);

//连接集群节点断开信号，将清除属于该节点的客户端全局哈希

connect (m_pCluster,&ZP_Cluster::zp_ClusterTerm::evt_NewSvrDisconnected,this,&st_client_table::on_evt_NewSvrDisconnected,Qt::QueuedConnection);

//集群节点数据接收，将直接向各个集群节点的处理队列中push ,以待线程池解译集群消息

connect (m_pCluster,&ZP_Cluster::zp_ClusterTerm::evt_RemoteData_recieved,this,&st_client_table::on_evt_RemoteData_recieved,Qt::QueuedConnection);

//集群节点数据发送, 只用于流量统计

connect (m_pCluster,&ZP_Cluster::zp_ClusterTerm::evt_RemoteData_transferred,this,&st_client_table::on_evt_RemoteData_transferred,Qt::QueuedConnection);

接着，在各个新客户端接入的时候，创建属于该客户端的节点类对象，并建立信号连接。

//this event indicates new client connected.

void st_client_table::on_evt_NewClientConnected(QObject * clientHandle)

{

bool nHashContains = false;

st_clientNode_baseTrans * pClientNode = 0;

m_hash_mutex.lock();

nHashContains = (m_hash_sock2node.find(clientHandle)!=m_hash_sock2node.end())?true:false;

if (false==nHashContains)

{

st_clientNode_baseTrans * pnode = new st_clientNodeAppLayer(this,clientHandle,0);

//using queued connection of send and revieve;

connect (pnode,&st_clientNode_baseTrans::evt_SendDataToClient,m_pThreadEngine,&ZPNetwork::zp_net_Engine::SendDataToClient,Qt::QueuedConnection);

connect (pnode,&st_clientNode_baseTrans::evt_close_client,m_pThreadEngine,&ZPNetwork::zp_net_Engine::KickClients,Qt::QueuedConnection);

connect (pnode,&st_clientNode_baseTrans::evt_Message,this,&st_client_table::evt_Message,Qt::QueuedConnection);

m_hash_sock2node[clientHandle] = pnode;

nHashContains = true;

pClientNode = pnode;

}

else

{

pClientNode = m_hash_sock2node[clientHandle];

}

m_hash_mutex.unlock();

assert(nHashContains!=0 && pClientNode !=0);

}

上面的代码中连接了三组信号，第一组把客户端发送数据的信号与网络模块连接起来，如客户端A发出该信号，含有客户端B的标示，将由网络模块响应，并最终发给客户端B。第二组连接了客户端踢出信号，比如客户端A想踢出客户端B，则把B的标示泵出，由网络模块响应并踢出B。第三组连接了消息显示信号。

集群应用层实现

前述，集群模块只提供了各个集群节点的物理连接，并没有实现具体的功能。因此，作为一个特定的应用，需要在集群模块上实现一个应用层功能。这个应用层功能有两部分组成。

一个是集群的应用层实现类st_cross_svr_node, 由集群节点类 ZP_Cluster::zp_ClusterNode派生，负责信令解译。 为了让集群知道这个类的存在，并使用该类而不是基类作为节点类，在st_client_table类构造函数中注册了本应用层实现类的工厂，这个工厂方法被用于产生st_cross_svr_node类的实例。

//zp_clusterterm.h

//...

/**

* The factory enables user-defined sub-classes inherits from zp_ClusterNode

* Using SetNodeFactory , set your own allocate method.

*注册工厂的方法/

void SetNodeFactory(std::function<

zp_ClusterNode * (

zp_ClusterTerm * /*pTerm*/,

QObject * /*psock*/,

QObject * /*parent*/)

>

);

std::function<zp_ClusterNode * (

zp_ClusterTerm * /*pTerm*/,

QObject * /*psock*/,

QObject * /*parent*/)> m_factory;

//zp_clusterterm.cpp

zp_ClusterTerm::zp_ClusterTerm(/*...*/ )

{

//...

m_factory = std::bind(&zp_ClusterTerm::default_factory,this,_1,_2,_3);

}

/**

* @brief The factory enables user-defined sub-classes inherits from zp_ClusterNode

* Using SetNodeFactory , set your own allocate method.

* @fn zp_ClusterTerm::default_factory the default factory function. just return zp_ClusterTerm *

* @param pTerm Term object

* @param psock Sock Object

* @param parent Parent

* @return zp_ClusterNode * 默认工厂产生zp_ClusterNode 类的实例

*/

zp_ClusterNode * zp_ClusterTerm::default_factory(

zp_ClusterTerm * pTerm,

QObject * psock,

QObject * parent)

{

return new zp_ClusterNode(pTerm,psock,parent);

}

//st_client_table.cpp

st_client_table::st_client_table(...)

{

m_pCluster->SetNodeFactory(

std::bind(&st_client_table::cross_svr_node_factory,

this,

_1,_2,_3)

);//绑定工厂回调方法

}

ZP_Cluster::zp_ClusterNode * st_client_table::cross_svr_node_factory(

ZP_Cluster::zp_ClusterTerm * pTerm,

QObject * psock,

QObject * parent)

{

st_cross_svr_node * pNode = new st_cross_svr_node(pTerm,psock,parent);

pNode->setClientTable(this);

return pNode;//实际产生的是st_cross_svr_node类的实例

}

另一个是集群的应用层消息，如下：

#ifndef ST_CROSS_SVR_MSG_H

#define ST_CROSS_SVR_MSG_H

namespace ExampleServer{

#pragma pack (push,1)

#if defined(__GNUC__)

#include <stdint.h>

typedef struct tag_example_crosssvr_msg{

struct tag_msgHearder{

__UINT16_TYPE__ Mark; //Always be "0x4567"

__UINT16_TYPE__ version; //Structure Version

__UINT8_TYPE__ mesageType;

__UINT32_TYPE__ messageLen;

} header;

union union_payload{

__UINT8_TYPE__ data[1];

__UINT32_TYPE__ uuids[1];

} payload;

} EXAMPLE_CROSSSVR_MSG;

#endif

#if defined(_MSC_VER)

typedef struct tag_example_crosssvr_msg{

struct tag_msgHearder{

unsigned __int16 Mark; //Always be 0x4567

unsigned __int16 version; //Structure Version

unsigned __int8 mesageType;

unsigned __int32 messageLen;

} header;

union union_payload{

unsigned __int8 data[1];

unsigned __int32 uuids[1];

} payload;

} EXAMPLE_CROSSSVR_MSG;

#endif

#pragma pack(pop)

}

#endif

基于上述的数据结构和处理方法，实现了三个messageType消息类型。

0x01 为 远程客户端登入消息，payload为各个新登入到该服务器客户端的UUID

0x02 为 远程客户端退出消息，payload为各个从该服务器退出的客户端的UUID

0x03 为 客户端载荷消息，封装了从远程客户端发给本地客户端之间的客户端-客户端消息。

均衡建议

无中心的服务器集群通过心跳广播在各个节点上存储了当前所有节点的负荷。当客户端登入时，会首先检查当前服务器负荷是否超过门限，一旦超过，则触发均衡建议。

bool st_client_table::NeedRedirect(quint8 bufAddresses[/*64*/],quint16 * pnPort)

{

if (m_pCluster->clientNums()<m_nBalanceMax)

return false;

QString strServerName = m_pCluster->minPayloadServer(bufAddresses,pnPort);

if (strServerName==m_pCluster->name())

return false;

return true;

}

bool st_clientNodeAppLayer::LoginClient()

{

//...

stMsg_ClientLoginRsp & reply = new stMsg_ClientLoginRsp (...);

//...

//Cluster-Balance.

if (m_pClientTable->NeedRedirect(reply.Address_Redirect,&reply.port_Redirect))

{

reply.DoneCode = 1;

//

}

//...

emit evt_SendDataToClient(this->sock(),reply);

}

客户端根据建议，可选择重新按新地址连接，或者继续保持。

结尾

C/S 架构服务器实现方式很多，应用案例成千上万。本范例为了演示基本的知识点，采用的设计思路并不是单纯从性能出发的。基于Qt实现，有助于利用Qt本身的引用计数、跨平台等特性，同时，Qt也是封装的非常棒的库，使得我们可以抛开繁琐的API，直接研究问题本身，为Linux, Windows下的同学提供统一的参考范例。感谢为Qt的进步付出心血的贡献者们，他们的不懈坚持让C++语言拥有了一个完整的跨平台UI框架。

见zp_tcpserver.cpp中ZP_TcpServer::incomingConnection()

见zp_tcpserver.h