结构化数据接入系统-代码文件

//#include"Modbus.h"#include<unistd.h>#include<sys/stat.h>#include<sstream>#include<string>#include"Map.h"#include<time.h>#include"MyQueue.h"#include"Threads.h"#include<sys/time.h>externstringusingnamespacestd;usingnamespaceesca;CSocket*ConnectSocket;AddrQueue*addrQueue;Trans_Map*trans_Map;queue<string>void*SendWorker(void*ParaTmp){Para*para=AddrQueue*aQueue=para->addrQueue;intDi[15];intLastDi[15];intAi[15];intfor(intt=0;t<15;Di[t]=LastDi[t]=Ai[t]=LastAi[t]=}charrequest[12];while(1){structtimevaltv_begin,tv_end;unsignedlongTotal_time;intcount=for(inti=1;i<16;i++)intCountDi=LastDi[i-1]- Di[i-1]+1;intF_D=Di[i-1];while(CountDi>0)request[0]=0x00;request[1]=request[2]=0x00;request[3]=request[4]=0x00;request[5]=request[6]=i;request[7]=0x02;request[8]=F_D>>8;request[9]=F_D&0xFF;if(CountDi>=50)request[10]=50>>8;request[11]=50&0xFF;}else}

request[10]=CountDi>>8;request[11]=CountDi&0xFF;cout<<"connectionbreakdownwhenwaitforresponse!"<<endl;}unsignedshorttmp=F_D;CountDi-=F_D+=}}for(inti=1;i<16;i++)intCountAi=LastAi[i-1]-Ai[i-1]+1;intF_A=Ai[i-1];while(CountAi>0)request[0]=0x00;request[1]=request[2]=0x00;request[3]=request[4]=0x00;request[5]=request[6]=i;request[7]=0x04;request[8]=F_A>>8;request[9]=F_A&0xFF;if(CountAi>=50)request[10]=50>>8;request[11]=50&0xFF;}else}

request[10]=CountAi>>8;request[11]=CountAi&0xFF;if(connectSocket->Send(request,12)<0)cout<<"sendfail"<<endl;unsignedshorttmp=F_A;//cout<<"SendNO."<<++count<<endl;//usefortesthowmanydatagramwassendduringaroundCountAi-=F_A+=}}gettimeofday(&tv_end,Total_time=(tv_end.tv_sec-tv_begin.tv_sec)*1000+(tv_end.tv_usec-tv_begin.tv_usec)/printf发送线程总共成功运行了：%ldUseconds\nTotal_time);}}Para*para=AddrQueue*aQueue=para->addrQueue;Trans_Map*transfer=para->trans_Map;unsignedshortAddr;charresponse[260];intcount=0;structtimevaltv_begin,tv_end;unsignedlongTotal_time;chartmp[50]; while(1)int//cout<<"receivenothing,continuetowaitfordata!"<<}if(ret<0)cout<<"connectionbreakdownwhenwaitforresponse!"<<endl;}if(aQueue->Empty())cout<<"ERROR:::::howcanthishappen"<<endl;}Addr=aQueue-////cout<<"enter"<<intstation=(unsignedchar)response[6];if(response[7]==0x02){intnum=((unsignedchar)response[8]*256)+(unsignedfor(intj=10;j<(10+num);j+=2)shorttemp=(((unsignedshort)response[j])<<8)+(unsignedshort)response[j+for(intk=0;k<16;k++){stringstreamss;stringO_Addr;intZ_Part,ss<<Addr<<"."<<k;O_Addr=ss.str();if(!transfer->Di(make_pair(station,O_Addr),Z_Part,X_Part))AFC_DP[station-1][Z_Part-1]((short)(~(1<<X_Part))&AFC_DP[station-1][Z_Part-1])|(short)(((temp>>k)&1)<<sprintf(tmp,"%d&%d.%d&DI&%d",station,Z_Part,X_Part,(temp>>X_Part)&1);strings(tmp);}}}elseif(response[7]==0x04)intnum=((unsignedchar)response[8]*256)+(unsignedfor(intj=10;j<(10+num);j+=2)shorttemp=(((unsignedshort)((unsignedchar)response[j]))<<8)+(unsigned

stringO_Addr;intN_Addr;ss<<Addr;O_Addr=ss.str();if(!transfer->Ai(make_pair(station,O_Addr),N_Addr)){}AFC_DP[station-1][N_Addr-1]=temp;sprintf(tmp,"%d&%d&AI&%d",station,N_Addr,temp);strings(tmp);}}////cout<<"out"<<if(++count==345)gettimeofday(&tv_end,Total_time=(tv_end.tv_sec-tv_begin.tv_sec)*1000+(tv_end.tv_usec-tv_begin.tv_usec)/printf("接收处理线程总共成功运行了：%ldUseconds\nTotal_time);gettimeofday(&tv_beginNULL);//cout<<"AFC_DP[0][0]"<<AFC_DP[0][0]<< AFC_DP[0][1499]="<<AFC_DP[0][1499]<<count=}}}void*ClientWorker(void*ParaTmp)cout启动客户端线程endl;CSocketconnectSocket;structsockaddr_inserverAddr;serverAddr.sin_family=AF_INET;serverAddr.sin_addr.s_addr=inet_addr(ISCS_IPif(connectSocket.Connect(serverAddr)!=0){cout<<errno<<endl;perror("connecttoservererror"); //connectingtoISCSfailswhenstartupasmajor,so}cout<<"connectsuccess"<<AddrQueue*aQueue=newAddrQueue;Para*para1=nepara1->addrQueue=aQueue;//ParaCThread*SendThread=newCThread(cnt++);CThread*HandThread=new//CThread*ServerThread=newSendThread->Start(1024,SendWorker,(void*)para1);while}}void*HandleClientWorker(void*ParaTmp)intlen,char*response=(char*)malloc(sizeof(char)*1024);while(1){response[0]=response[1]=//s+=//s+=ind=while(!sqempty()){strings=sq.front();if((ind+20)>(1024)||sq.empty()){if(sq.empty()){cout<<s<<//len=strlen((*it).c_str());len=s.size();cout<<len<<endl;cout<<ind<<endl;memcpy(response+ind,s.c_str(),len);ind+=len;}if(connectSocket->Send(response,ind+1)<0){cout<<"sendfail"<<endl;}elsecout<<string(response)<<endl;cout<<"sendsuccess"<<endl;}ind=}elsecout<<s<<len=s.size();cout<<len<<endl;cout<<ind<<endl;memcpy(response+ind,s.c_str(),len);ind+=len;}} cout<<"HandleClientWorker"<<endl;continue}strings=sq.front();response[0]=response[1]=ind=len=cout<<len<<//cout<<ind<<endl;cout<<s<<endl;response[ind]=len;memcpy(response+ind,s.c_str(),len);ind+=len;if(connectSocket->Send(response,ind+1)<0){cout<<"sendfail"<<endl;}elsecout<<string(response)<<endl;cout<<"sendsuccess"<<endl;} }}void*ServerWorker(void*ParaTmp)cout启动服务器线程endl;CSocketlistenSocket;structsockaddr_insin;memset(&sin,0,sizeof(sin));sinsin_family=sinsin_port=listenSocket.Bind(TCC_CLIENT_PORT,sin);while(1)boolindicator=cout<<"等待客户端"<<endl;connectSocket=newCSocket;cout<<"客户端成功"<<endl;CThread*HandleClientThread=newdeleteconnectSocket;}}cnt=2;CThread*ClientThread=newCThread(cnt++);CThread*ServerThread=newCThread(cnt++);while(!sq.empty()){}ServerThread->Start(1024,ServerWorker,(void*)0);while}}usingnamespacestd;BStatebackupState;stringerrmsg;//错误信息charipaddr[50];//服务器IP地unsignedshorthbport;////初始化备机状态结voidinitBackupState(BState{}//设置备机状态为voidsetBackupStateOn(BState{}//设置备机状态为voidsetBackupStateOff(BState{}//获取备机intgetBackupState(BState{intstate;pthread_mutex_unlock(&bs-}//初始化主备标志结voidinitHostOrBackup(HOBFlag{}//设置当前机器为主voidsetHostFlag(HOBFlag{}//设置当前机器为备voidsetBackupFlag(HOBFlag{}intIsHost(HOBFlag*hb){boolpthread_mutex_lock(&hb-}}//设置服务器IP和端voidsetServerIPandPort(constchar*ip,constunsignedshortport){}int}//返回错误constchar*}//定时函voidusecsleep(intsec,intusec){fd_setrd;structtimeval}//心跳线void*HeartBeatThread(void*{{{{//std::cout<<getErrMsg()<<std::endl;//可以换成日志函}}{{//std::cout<<getErrMsg()<<std::endl;//可以换成日志函}}}}//启动心跳voidstartHeartBeat(constchar*ip,constunsignedshortpthread_ttid;interr;{}}返回0表示函数正常结束intcout<<"进入structsockaddr_inhostaddr;socklen_tbackupaddrlen=sizeof(backupaddr);{return1;}intsetsockopt(host_sockfd,SOL_SOCKET,SO_REUSEADDR,(consthostaddrsin_family=AF_INET;hostaddrsin_port=htons(hbport);hostaddrif(bind(host_sockfd,(structsockaddr*)&hostaddr,sizeof(structsockaddr))==-{errmsg=string("bind失败！");return2;}if(listen(host_sockfd,2)==-{errmsg=string("listen失败！");return3;}sigset_tnewsigset,oldsigset;{cout<<"clientlinksuccess"<<endl;{if(send(accept_sockfd,"o!",strlen("o!"),0)==-{{}errmsg=string("send失败！,TCP连接断开。");}}}}//检测TCP连接是否断掉,本函数实际没有使{structsockaddr_infrom;socklen_tfromlen=0;{returntrue;}}//客户端执行代码回-1表示创建socket失败，返回-2表示connect失败，返回0表示函数正常执intcout<<"进入structsockaddr_inhostaddr;charrecvbuf[50];if((backup_sockfd=socket(AF_INET,SOCK_STREAM,0))==-{return1;}hostaddrsin_family=AF_INET;hostaddrsin_port=htons(hbport);if(connect(backup_sockfd,(structsockaddr*)&hostaddr,sizeof(structsockaddr))==-{errmsg=string("客户端connect失败！");}structtimevaltimeout={IntervalTime+1,0};fd_setsetfd;intnfds;while(1){elseif(nfds==0){}{

{{}{}}}}}}#include<iostream>#include<cstdio>#include<cstdlib>#include<unistd.h>#include<string.h>#include<string>#include<errno.h>#include<netinet/in.h>#include<sys/select.h>#include<sys/types.h>#include<pthread.h>#include<arpa/inet.h>#include<signal.h>constintOK1;constintDOWN=0;constintHOST=1;constintBACKUP=0;//全局变{intbackupState;//1表示正常，0表示当pthread_mutex_ttypedefstruct{int_HostOrBackup;//1表示主，0pthread_mutex_tvoidinitBackupState(BState*bs);intgetBackupState(BState*bs);intvoidinitHostOrBackup(HOBFlag*hb);voidsetHostFlag(HOBFlag*hb);voidsetBackupFlag(HOBFlag*hb);intIsHost(HOBFlag*hb);voidusecsleep(intsec,intusec);voidsetServerIPandPort(constchar*ip,constunsignedshortport);voidstartHeartBeat(constchar*ip,constunsignedshortport);constchar*getErrMsg();int}#definePORT8899usingnamespacestd;usingnamespaceintmain(intargc,char*{startHeartBeat("222201.139.143",PORT);{}}//:#include"stdafx.h"{//在创建任何控件之前启用WindowsXP可视化效 //创建主窗口并运行它return0;usingnamespaceSystem;namespaceModel{publicrefclass{String^recvTime;String^lineId;String^stationId;String^dpAddr;String^dpType;String^dpValue;

intnumLineId;intpropertyString^{String^}}}propertyString^{String^}}}propertyString^{String^}}}propertyString^{String^}}}propertyString^{String^}}}propertyString^{String^}}}propertyString^{String^return}}}BooleanEqual(DPInfo^{returnfalse;returnfalse;returnfalse;returnfalse;}}#pragmaonce//#defineusingnamespaceSystem;usingnamespace usingnamespaceSystem::Collections;usingnamespaceSystem::Data;usingnamespaceSystem::Threading;usingnamespaceSystem::IO;usingnamespaceSystem::Xml;usingnamespaceModel;usingnamespaceSystem::Collections::Generic;usingnamespaceSystem::Net;//////Form1///publicrefclassForm1:public{{//TODO:在此处添加构造函数代}//////清理所有正在使用的资源///{if{}}private:System::Windows::Forms::DataGridView^private:System::Windows::Forms::ToolStrip^private:System::Windows::Forms::ToolStripLabel^toolStripLabel1Status;private:System::Windows::Forms::DataGridViewTextBoxColumn^ColTime;private:System::Windows::Forms::DataGridViewTextBoxColumn^ColSubsystem;private:System::Windows::Forms::DataGridViewTextBoxColumn^ColLine;private:System::Windows::Forms::DataGridViewTextBoxColumn^ColStation;private:System::Windows::Forms::DataGridViewTextBoxColumn^ColAddress;private:System::Windows::Forms::DataGridViewTextBoxColumn^ColDataType;private:System::Windows::Forms::DataGridViewTextBoxColumn^ColValue;//////必需的设计器变量///ponentModel::Container//////设计器支持所需的方法-不///使用代码编辑器修改此方法的内容/// {this->ColTime=(gcnewthis->ColLine=(gcnewSystem::Windows::Forms::DataGridViewTextBoxColumn());this->ColStation=(gcnewSystem::Windows::Forms::DataGridViewTextBoxColumn());this->ColAddress=(gcnewSystem::Windows::Forms::DataGridViewTextBoxColumn());this->toolStrip1=(gcnewSystem::Windows::Forms::ToolStrip());this->toolStripLabel1Status=(gcnewSystem::Windows::Forms::ToolStripLabel()); ponentModel::ISupportInitialize^>(this->DPInfoList))->BeginInit();this-//this->DPInfoList->AllowUserToAddRows=false;this->DPInfoList->AllowUserToResizeRows=false;this->DPInfoList- this->DPInfoList->Columns->AddRange(gcnewcli::array< this->ColLine,this->ColStation,this->ColAddress,this->ColDataType,this->ColValue});this->DPInfoList->GridColor=System::Drawing::SystemColors::Control;this->DPInfoList->Name=L"DPInfoList";this->DPInfoList->ReadOnly=this->DPInfoList->RowHeadersWidth=this->DPInfoList->RowTemplate->Height=this->DPInfoList->TabIndex=0;this->DPInfoList->VirtualMode=this->DPInfoList->CellValueNeeded+=gcnewSystem::Windows::Forms::DataGridViewCellValueEventHandler(this,this->DPInfoList->RowsAdded+=gcnewSystem::Windows::Forms::DataGridViewRowsAddedEventHandler(this,//this->ColTime->DataPropertyNameL"RecvTime";this->ColTime->HeaderText=L"时间";this->ColTime->Name=L"ColTime";this->ColTime->ReadOnly=true;this->ColTime->Width=150;//this->ColSubsystem->HeaderText=L"专业";this->ColSubsystem->ReadOnly=true;this->ColSubsystem->SortMode=//this->ColLine->DataPropertyNameL"LineId";this->ColLine->HeaderText=L"线路";this->ColLine->Name=L"ColLine";this->ColLine->ReadOnly=true;this->ColLine->SortMode=//this->ColStation->DataPropertyNameL"StationId";this->ColStation->HeaderText=L"车站";this->ColStation->Name=L"ColStation";this->ColStation->ReadOnly=true;this->ColStation->SortMode=//this->ColAddress->DataPropertyNameL"DpAddr";this->ColAddress->HeaderText=L"地址";this->ColAddress->Name=L"ColAddress";this->ColAddress->ReadOnly=true;this->ColAddress->SortMode=//this->ColDataType->HeaderText=L"数据类型";this->ColDataType->ReadOnly=true;this->ColDataType->SortMode=//this->ColValue->HeaderText=L"值";this->ColValue->ReadOnly=true;this->ColValue->SortMode=//this->toolStrip1->Items- {this-this->toolStrip1->Name=L"toolStrip1";this->toolStrip1->TabIndex=1;this->toolStrip1->Text=//this->toolStripLabel1Status->Name=L"toolStripLabel1Status";//this->AutoScaleDimensions=System::Drawing::SizeF(6,this->ClientSize=System::Drawing::Size(849,598);this->izeBox=false;this->Name=L"Form1";this->Text=L"Form1"; ponentModel::ISupportInitialize^>(this->DPInfoList))->EndInit();this-}#pragmaregionMyCodesIPAddress^serverIP1;IPAddress^Int32RefreshDelegate^refreshDelegate;delegatevoidRefreshStatusDelegate(String^);//实时数据//System::Collections::Generic::Queue<ACInfo^>^Booleanrunning;Thread^mainThread;staticBooleanlatter=false;{////数 ew//先服务器参数，配置正确才创建线//刷 ewewThread(gcnewThreadStart(this,&Form1::MainThremainThread- ew ew}}voidMyRefresh()//(List<DPInfo^>^{}

}}

this->rwList->AcquireReaderLock(-this->rwList-"+ex-

catch(ApplicationException^ex){

+")]-

+")]-

}}voidMyRefreshStatus(String^{}Boolean{^int^}

ewXmlxDoc-catch(Exception^{//配置文件加载失//Logger::info("Error:"+ex-"+ex-}

+")]-

+")]-

bgSvrElem=(XmlElement^)xDoc->Element-//服务器foreach(XmlElement^svrEleminbgSvrElem->ChildNodes){XmlElement^xmlElem;if(xmlElem-}

}}catch(Exception^{//Logger::trace("Error:ServerIPorPortnotConfiguredCorrectly.");MessageBox::Show("错误：服IP或端口地址未配置或配置错误");returnfalse;}}}//Logger::trace("Error:ServerIPorPortnotConfiguredCorrectly.");MessageBox::Show("错误：服IP或端口地址未配置或配置不正确");returnfalse;}elsereturn}voidUpdaist(List<DPInfo^>^list,DPInfo^{/*list中寻找与dpInfo相同（专业、线路、车站、地址相同）的数据点，找到后移除，然后将新数据list中 //先获取list的读锁，在需要修改list的时候再升级为写 this->rwList->AcquireReaderLock(- for(inti=0;i<list- if(dpInfo- //将读锁升级到写锁，然后再数 Locklc=rwList->UpgradeToWriterLock(- list- list- rwList- catch(ApplicationException^

+")]-

+")]-

//没有相同数据点，直接头 Locklc=rwList->UpgradeToWriterLock(- list- rwList- catch(ApplicationException^

+")]-

+")]-

this->rwList- //catch(ApplicationException^"+ex-

+")]-

+")]-

for(inti=0;i<list-if(dpInfo-//将读锁升级到写锁，然后再数}}{//没有相同数据点，直接头list-}}"+ex-

catch(Exception^

+")]-

+")]-

}}voidMainThre{String^lineKey,^subsystemKey;IPAddress^serverIP;Int32serverPort;TcpClient^tcpClient;BooleanreConnect=false;List<DPInfo^>^ ewarray<String^>(1); inttoRead,readLen;//要读/已读int

tcpClient=gcnew ewarray<Byte>(255);while(running){//while(!this->IsHandleCreated){};//等待窗体创statusArray[0]="正在连接服务器 ew}{"+ex-

+")]-

+")]-

}}#pragmaregion注

statusArray[0]=L"连接成功this-statusArray[0]="正在接收数据this-while(running){//中myArray[0]=storeList; ew//先读3个字节长度头//while(running){//里 if(readStream->CanRead&&readStream- int if((len=readStream- //如接收到数据，就将睡眠时间设为最

+")]-

+")]-

+")]"+ex- //MessageBox::Show(ex- else{//自适应睡眠时 //下次睡眠时间

//