分布式计算框架(二)服务器模块

二、服务器模块

2.1 服务器模块结构

服务器模块为整个系统的核心组成部分，负责数据的处理、任务的分发负载均衡等。由于功能数量较多，因此定义一个管理类统一管理，对外只暴露简单的接口方便调用，该模块由以下几个类组成。

Management:管理类，负责管理所有模块。

Handle_manage:处理管理类，负责管理处理线程,。

Handle:处理类，负责处理任务。

Log:日志类，负责保存日志。

Mysql:数据库类，负责与数据库的交互。

Setting:配置类，负责保存读取程序配置。

这几个类关系如图下

2.2 运行原理

2.2.1 Management类

    服务器运行时生成管理类Management，此类封装了其他功能类开启和关闭的方法。比如监听tcp端口时先调用tcpServer_init(int threadCount,int Port)方法初始化，再调用tcpServer_start()开始监听。以下是服务器初始化代码：

int main(int argc, char *argv[])
{
bool Log_Open;
QString SQL_IP,SQL_DBname,SQL_UserName,SQL_Password,LocalPort;
int TcpHandleThreadNum,TcpPort,Handle_Count,distribution_multiple;
QCoreApplication a(argc, argv);
qInstallMessageHandler(MessageOutput);
QSharedPointer<Management>management=QSharedPointer<Management>(new Management());
management.data()->Configur_init("balance_server");
management.data()->StartConfigur();
Common::ConfigureLock.lockForRead();
Log_Open=Common::Log_Open;
SQL_IP=Common::SQL_IP;
SQL_DBname=Common::SQL_DBname;
SQL_UserName=Common::SQL_UserName;
SQL_Password=Common::SQL_Password;
LocalPort=Common::LocalPort;
TcpHandleThreadNum=Common::TcpHandleThreadNum;
TcpPort=Common::TcpPort;
Handle_Count=Common::Handle_Count;
distribution_multiple=Common::distribution_multiple;
Common::ConfigureLock.unlock();
if(Log_Open)
{
management.data()->Log_init("balance_server");
management.data()->StartLog();
}
else
{
qInstallMessageHandler(NULL);
}
management.data()->mysql_set(SQL_IP,SQL_DBname,SQL_UserName,SQL_Password);
management.data()->Handle_Setcount(Handle_Count);
management.data()->Handle_Start();
management.data()->localServer_init(LocalPort);
management.data()->localServer_start();
management.data()->tcpServer_init(TcpHandleThreadNum,TcpPort);
management.data()->tcpServer_start();
management.data()->process_init();
management.data()->process_start();
qDebug()<<"任务分配倍率:"<<distribution_multiple;
return a.exec();
}

2.2.2、监听TCP

    监听TCP由tcp_server来完成，tcp_server继承了QTcpServer类兵重载incomingConnection函数，当有TCP连接时会触发incomingConnection函数，并在此函数中生成socket并分配给空闲线程，如果没有空闲线程则拒绝接入。incomingConnection函数代码如下：

void tcp_server::incomingConnection(qintptr socketDescriptor)
//当有新用户接入时触发此程序
{
qDebug()<<"新用户连接";
QThread *thread=Choice_Free_Thread();
//选择负载最小的线程
if(thread!=NULL)
{
tcp_socket *socket=new tcp_socket(socketDescriptor);
if(!socket->tcpSocket_init())
{
socket->deleteLater();
return;
}
socket->send_message("服务器","获取程序类型","");
QSharedPointer<M_Socket>
msocket=QSharedPointer<M_Socket>(new M_Socket());
msocket.data()->Stype=Tsocket;
msocket.data()->t_socket=socket;
Common::tcp_server_SocketMapLock.lockForWrite();
Common::tcp_server_SocketMap[socket]=msocket;
//tcp_server_SocketMap添加socket
Common::tcp_server_SocketMapLock.unlock();
Socket_Thread[socket]=thread;
//记录socket所在线程
Thread_Socket_Cout[thread]+=1;
//增添线程socket数量
connect(this,SIGNAL(disconnect_Allsocket()),socket,SLOT(disconnect_Allsocket()));
connect(socket,SIGNAL(disconnected()),this,SLOT(delete_tcpSocket()),Qt::QueuedConnection);
//断开时操作
connect(this,SIGNAL(delete_socket(tcp_socket*)),socket,SLOT(socket_delete(tcp_socket*)),Qt::QueuedConnection);
connect(socket,SIGNAL(send_ReplyData(QString,QString,QString)),handlemanage.data(),SLOT(receive(QString,QString,QString)),Qt::QueuedConnection);
//发送任务给处理线程管理类
socket->moveToThread(thread);
connect(handlemanage.data(),SIGNAL(t_SocketSendData(QString,QString,QString,tcp_socket*,emit_type,QString)),socket,SLOT(send_messagefromslot(QString,QString,QString,tcp_socket*,emit_type,QString)),Qt::QueuedConnection);
connect(&TcpClock,SIGNAL(timeout()),socket,SLOT(IsConnect()),Qt::QueuedConnection);
}
else
{
qDebug()<<"目前所有线程达到警戒值，拒绝接入!";
}
}

2.2.3 数据接收与发送

    tcp_socket继承QTcpSocket并且使用信号槽，当有数据接收时调用receive()函数以下是receive函数代码:

void tcp_socket::receive()
{
receive_data+=read(bytesAvailable());
if(receive_datasize!=0||receive_data.size()>=receive_datasize)
{
analyse_data();
}
}

当接收数据大小符合条件时调用analyse_data()函数来进行数据包处理。

TCP数据包格式约定为int32(数据包大小)+int64(特征值)+bool(压缩标志)+真正数据，当接收数据数据大于等于数据包大小则进行数据包解析,以下时analyse_data()函数代码:

void tcp_socket::analyse_data()
{
while(receive_data.size()>(sizeof(qint32)+sizeof(qint64)))
{
bool isCompress;
qint64 data_size;
qint32 feature;
QByteArray data;
QDataStream dts(&receive_data,QIODevice::ReadWrite);
dts>>feature>>data_size;
if(feature==this->feature)
{
if(data_size<=receive_data.size())
{
dts.device()->seek(sizeof(qint32)+sizeof(qint64));
dts>>isCompress;
dts>>data;
receive_data.remove(0,data_size);
if (!isCompress)
{
process_data(data);
}
else
{
process_data(qUncompress(data));
}
}
else
{
break;
}
}
else
{
qDebug()<<"特征值错误，断开socket";
disconnectFromHost();
break;
}
}
}

    当数据包接收完毕则进行数据解析，数据使用json格式进行封装，process_data()函数进行json解析，以下是process_data()函数代码：

void tcp_socket::process_data(QByteArray data)
{
this->LastConnectTime=QDateTime::currentDateTime();
QString id,type,information;
QStringList result=json_analyse(data);
if(result.size()>=3)
{
id=result[0];
type=result[1];
information=result[2];
if (type=="程序类型")
{
this->name=id;
this->type=information;
}
emit this->send_ReplyData(id,type,information);
}
}

数据解析完毕后则使用信号槽发出一个信号，将任务发送至Handle_Management类，这个类将会选择空闲线程进行任务处理。

数据的发送较为简单，在tcp_socket生成时会使用信号槽进行连接，当tcp_socket接收信号时调用send_data()函数，以下是send_data()函数代码

void tcp_socket::send_data(QByteArray data)
{
bool isCompress=false;
if(data.size()>qCompress(data).size())
{
isCompress=true;
data=qCompress(data);
}
QByteArray senddata;
QDataStream dts(&senddata,QIODevice::WriteOnly);
dts<<(qint32)0<<(qint64)0<<(bool)isCompress<<data;
dts.device()->seek(0);
dts<<feature;
dts.device()->seek(sizeof(qint32));
dts<<(qint64)senddata.size();
write(senddata);
flush();
}

2.2.4 任务的处理

    handle_manage类负责Handle处理线程的管理，并接收来自tcp_socket的任务并选择空闲Handle线程进行处理，同时也转发来自Handle线程的处理结果。以下是handle_manage初始化函数代码：

void handle_manage::init()
{
for (int a=0;a<this->handle_count;a++){
QThread
*thread=new QThread();
Handle
*handle=new Handle(QString::number(a+1),false);
thread_map[handle]=thread;
//设置线程map
thread_status[handle]=true;
thread_task_num[handle]=0;
//设置线程任务数
connect(this,SIGNAL(Handlelocaldata(QString,QString,QString,local_socket*,Handle*,emit_type)),handle,SLOT(handle_localSocket(QString,QString,QString,local_socket*,Handle*,emit_type)),Qt::QueuedConnection);
connect(this,SIGNAL(Handletcpdata(QString,QString,QString,tcp_socket*,Handle*,emit_type)),handle,SLOT(handle_tcpSocket(QString,QString,QString,tcp_socket*,Handle*,emit_type)),Qt::QueuedConnection);
connect(handle,SIGNAL(l_SocketSendData(QString,QString,QString,local_socket*,emit_type,QString)),this,SLOT(Cl_SocketSendData(QString,QString,QString,local_socket*,emit_type,QString)),Qt::QueuedConnection);
connect(handle,SIGNAL(t_SocketSendData(QString,QString,QString,tcp_socket*,emit_type,QString)),this,SLOT(Ct_SocketSendData(QString,QString,QString,tcp_socket*,emit_type,QString)),Qt::QueuedConnection);
connect(handle,SIGNAL(finish_Task()),this,SLOT(Handle_free()),Qt::QueuedConnection);
handle->moveToThread(thread);
thread->start();
}
sql_thread=new QThread();
sql_handle=new Handle("handle_sql",true);
connect(this,SIGNAL(Handlelocaldata(QString,QString,QString,local_socket*,Handle*,emit_type)),sql_handle,SLOT(handle_localSocket(QString,QString,QString,local_socket*,Handle*,emit_type)),Qt::QueuedConnection);
connect(this,SIGNAL(Handletcpdata(QString,QString,QString,tcp_socket*,Handle*,emit_type)),sql_handle,SLOT(handle_tcpSocket(QString,QString,QString,tcp_socket*,Handle*,emit_type)),Qt::QueuedConnection);
connect(sql_handle,SIGNAL(l_SocketSendData(QString,QString,QString,local_socket*,emit_type,QString)),this,SLOT(Cl_SocketSendData(QString,QString,QString,local_socket*,emit_type,QString)),Qt::QueuedConnection);
connect(sql_handle,SIGNAL(t_SocketSendData(QString,QString,QString,tcp_socket*,emit_type,QString)),this,SLOT(Ct_SocketSendData(QString,QString,QString,tcp_socket*,emit_type,QString)),Qt::QueuedConnection);
connect(sql_handle,SIGNAL(finish_Task()),this,SLOT(Handle_free()),Qt::QueuedConnection);
sql_handle->moveToThread(sql_thread);
sql_thread->start();
Handle
*handle= Choice_Free_Handle();
MessageSocket socket;
socket.Stype=Lsocket;
Handle_Message("handle_manage","自动读取未完成任务","",socket,One);
handle= Choice_Free_Handle();
emit
Handlelocaldata("handle_manage","自动读取DLL信息","",NULL,handle,One);
}

    Handle类为处理类，每个Handle拥有自己的Mysql连接(这里没有设计好，应该是一个数据库连接池)，接收来自Handle_management转发的任务并处理。由于代码较长且比较冗余就不放出来了，希望各位可以提一点建议。

2.2.5 本地计算模块启动与通信

     专门设计为本地计算模块就是防止调用DLL时崩溃波及到本程序，因此在服务器启动时会启动本地计算模块，并且监听对应端口，由于代码大致与TCP相同就略过。

2.2.6 本地配置

    由于监听端口、处理线程数量需要根据需要设定，因此设计读取本地配置初始化。当配置文件夹或者配置文件中配置项不存在时使用默认配置，由于代码较为简单则不放出来了。

2.2.7 任务的分配

    Assignment_task函数进行任务的分配，以下是函数代码

void Handle::Assignment_task()
//分配任务
{
QSharedPointer<M_Socket> socket=Common::GetFree_Caculatenode();
//分配时已经改变socket修改socket一个计算线程为计算，因此不用考虑多线程是否冲突
if (socket.isNull())
{
//
qDebug()<<"没有空闲计算节点！";
return;
}
QSharedPointer<Decomposition_Task> D_task=Common::GetFree_Decomposition_Task();
if(D_task.isNull())
{
//
qDebug()<<"没有任务,重置计算节点信息";
socket.data()->lock.lockForWrite();
if(socket.data()->task_id.size()<socket.data()->Caculate_status.size())
{
int num= socket.data()->Caculate_status.indexOf(Caculating);
socket.data()->Caculate_status[num]=Cfree;
}
socket.data()->lock.unlock();
return;
}
//
qDebug()<<"获取空闲计算任务完毕"<<QThread::currentThreadId();
D_task.data()->DTask_Lock.lockForRead();
//获取任务id
QString taskid=D_task.data()->getTaskid();
QString taskdata=D_task.data()->getTaskData();
QString tasktype=D_task.data()->getTaskType();
QString belongCtaskid=D_task.data()->getBelongTaskID();
D_task.data()->DTask_Lock.unlock();
socket.data()->lock.lockForWrite();
//获取computer_id
socket.data()->task_id.append(taskid);
QString computer_id=socket.data()->Socket_name;
socket.data()->lock.unlock();
D_task.data()->DTask_Lock.lockForWrite();
//设置BelongComputer
D_task.data()->setBelongComputer(computer_id);
QDateTime date=QDateTime::currentDateTime();
QString now_time=date.toString("yyyy-MM-dd-hh:mm:ss");
D_task.data()->setStartTime(now_time);
D_task.data()->DTask_Lock.unlock();
QStringList task_data;
task_data.append(taskid);
task_data.append(tasktype);
task_data.append(belongCtaskid);
task_data.append(taskdata);
QString senddata=task_data.join("**&&##");
emit this->t_SocketSendData("服务器","计算任务",senddata,socket->t_socket,One,"");
Assignment_task();
}

    大致思路就是从全局数据中获取空闲任务与空闲计算节点，并且使用锁进行同步，当符合条件则向空闲节点发送计算任务。

2.2.8 任务汇总

    每当接收分解任务结果是判断对应任务中未计算队列与计算中队列是否为空，为空代表任务计算完毕。以下是计算任务结构：

class Caculate_Task
{
public:
Caculate_Task(QString id,QString type,QString task_data);
QMap<QString,QSharedPointer<Decomposition_Task>> Decomposition_unfinish_task;//分解任务未完成列表
QMap<QString,QSharedPointer<Decomposition_Task>> Decomposition_Caculating_task;//正在计算的分解任务列表
QMap<QString,QSharedPointer<Decomposition_Task>> Decomposition_finish_task;//分解任务完成列表
bool
isfinish=false;
//是否计算完成
long
task_size=0;
//分解任务总数
QReadWriteLock
unfinish_task_lock;
QReadWriteLock
Caculating_task_lock;
QReadWriteLock
finish_task_lock;
QReadWriteLock
Data_task_lock;
void
setTaskid(QString id);
void
setTasktype(QString type);
void
setTaskdata(QString data);
void
set_Taskschedule(QString schedule);
void
set_TaskcreateId(QString createId);
void
set_TaskstartTime(QString StartTime);
void
set_TaskendTime(QString EndTime);
void
set_TaskResult(QString TaskResult);
void
set_TaskIsFinish(QString IsFinish);
void
set_TaskTimeConsuming(QString TimeConsuming);
void
set_Taskdecomposition_num(int num);
void
set_TaskLastupdate_time(QDateTime lasttime);
QString Taskid();
QString Tasktype();
QString Taskdata();
QString Taskschedule();
QString Taskcreateid();
QString Taskstarttime();
QString Taskendtime();
QString Taskresult();
QString Taskisfinish();
QString Tasktimeconsuming();
int
Taskdecomposition_num();
QDateTime TaskLastupdate_time();
private:
QString
caculate_task_id;
//任务id
QString
tasktype;
//任务类型
QString
task_schedule;
//任务进度
int
task_decomposition_num=0; //分解任务数量
QDateTime
Lastupdate_time;
//上次更新时间
QString
task_data;
//任务信息
QString
create_id;
//创建者ID
QString
start_time;
//创建时间
QString
end_time;
//结束时间
QString
task_result;
//任务结果
QString
task_isfinish;
//任务是否结束
QString
task_timeconsuming;
//任务消耗时间
};

    以下是分解任务结构：

class Decomposition_Task
{
public:
Decomposition_Task();
void init(QString task_id,QString taskdata);
//初始化
void setProblemStatus(Ctask_status status);
//设置任务状态
void setBelongComputer(QString computerId);
//设置电脑ID
void setStartTime(QString Time);
//设置开始时间
void setCreateTime(QString Time);
//设置创建时间
void setFinishTime(QString Time);
//设置结束时间
void setCaculateTime(QString time);
//设置计算耗时
void setTaskResult(QString result);
//设置任务结果
void setBelongTaskID(QString ID);
//设置主任务ID
void setTaskData(QString taskdata);
//设置任务信息
void setTaskType(QString tasktype);
//设置任务类型
QString getTaskid();
//任务ID
QString getTaskData();
//获取任务信息
Ctask_status getProblemStatus();
//获取任务状态
QString getBelongComputer();
//获取电脑ID
QString getStartTime();
//获取开始时间
QString getCreateTime();
//获取创建时间
QString getFinishTime();
//获取结束时间
QString getCaculateTime();
//获取计算耗时
QString getTaskResult();
//获取任务结果
QString getBelongTaskID();
//获取主任务ID
QString getTaskType();
//获取任务类型
QReadWriteLock
DTask_Lock;
//读写锁
//private:
QString task_data="";
//任务信息
QString task_id="";
//任务ID
Ctask_status problem_status =task_free;
//任务状态
QString problem_type="";
//任务类型
QString belong_computer_id="";
//任务电脑ID
QString starttime="";
//任务开始时间
QString create_time="";
//任务创建时间
QString finish_time="";
//任务结束时间
QString
caculate_time=0;
//任务耗费时间
QString
task_result="";
//任务结果
QString
belong_caculate_task_id="";
//任务属于ID
};

    当计算完毕后服务器调用FindFinsh_Task函数进行汇总，以下是代码：

void Handle::FindFinsh_Task()
{
QStringList Finish_Taskid;
Common::CaculateTask_lock.lockForRead();
QMap<QString,QSharedPointer<Caculate_Task>>::const_iterator i;
qDebug()<<"开始查找是否有完成任务!";
for (i = Common::CaculateTask.constBegin(); i != Common::CaculateTask.constEnd(); ++i)
{
bool Unfinishtask=false;
bool Caculatetask=false;
i.value().data()->unfinish_task_lock.lockForRead();
//加锁判断未完成任务队列是否为空
if(i.value().data()->Decomposition_unfinish_task.isEmpty())
{
Unfinishtask=true;
}
i.value().data()->unfinish_task_lock.unlock();
i.value().data()->Caculating_task_lock.lockForRead();
//加锁判断计算队列是否为空
if(i.value().data()->Decomposition_Caculating_task.isEmpty())
{
Caculatetask=true;
}
i.value().data()->Caculating_task_lock.unlock();
if(Unfinishtask&&Caculatetask)
//如果
{
i.value().data()->Data_task_lock.lockForRead();
Finish_Taskid.append(i.value().data()->Taskid());
i.value().data()->Data_task_lock.unlock();
}
}
Common::CaculateTask_lock.unlock();
for(int a=0;a<Finish_Taskid.size();a++)
{
SumResult(Finish_Taskid.at(a));
}
}

    简单而言就是效验一次是否符合情况，最后调用SumResult函数进行汇总。SumResult函数最主要功能就是发送汇总信息至本地计算模块，核心代码为：

emit l_SocketSendData("服务器","sum",send_data,l_socket->l_socket,One,"");

2.2.9 服务器模块小节

    处理线程逻辑代码较为复杂，涉及到多线程进程竞争资源、死锁等问题，当程序写完时经过半个月测试才使其稳定。在测试时发现数据库这块耗时较多，因此专门指定一个线程专门负责数据的更新。

最后

以上就是潇洒曲奇为你收集整理的分布式计算框架(二)服务器模块的全部内容，希望文章能够帮你解决分布式计算框架(二)服务器模块所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。