//為了支持移植 沿用Ptypes的頭文件 #define WIN32 1 #include "ptime.h" #include "pinet.h" #include "ptypes.h" #include "pasync.h"#ifdef WIN32 #pragma comment(lib, "ptypes.lib") #pragma comment(lib, "ws2_32.lib") #endif#ifdef WIN32 # include <winsock2.h> #else # include <sys/time.h> # include <sys/types.h> # include <sys/socket.h> # include <netinet/in.h> # include <arpa/inet.h> # include <netdb.h> # include <unistd.h> # include <time.h> #endifUSING_PTYPESconst int testport = 8101; const int maxtoken = 4096; const int timeout = -1; const int maxsetsize = 1024*10; const int maxthreads = 30;const int MSG_MYJOB = MSG_USER + 1;fd_set g_allset; //保存所有要檢測的socket； int g_arrClientSocket[maxsetsize]; struct SocketInfo {int nSocket;int nIndex; };class myjobthread: public thread { protected:int id;jobqueue* jq;virtual void execute(); public:myjobthread(int iid, jobqueue* ijq): thread(false), id(iid), jq(ijq) {}~myjobthread() { waitfor(); } };class myjob: public message { public:SocketInfo* m_pSocketInfo;myjob(SocketInfo* pSocketInfo): message(MSG_MYJOB), m_pSocketInfo(pSocketInfo) {}~myjob() {delete m_pSocketInfo; } };void myjobthread::execute() {bool quit = false;while (!quit){// get the next message from the queuemessage* msg = jq->getmessage();try{switch (msg->id){case MSG_MYJOB:{pout.putf("%t,線程：%d\n", now(), id);int nClientSocket = ((myjob*)msg)->m_pSocketInfo->nSocket;int nIndex = ((myjob*)msg)->m_pSocketInfo->nIndex;BYTE buffer[maxtoken];memset(buffer, 0x00, sizeof(buffer));int nLen1=0,nLen2 = 0,nRemainLen=0;nLen1 = ::recv(nClientSocket, (char*)buffer, maxtoken, 0);//nLen1 = ::recv(nClientSocket, (char*)buffer, sizeof(UINT32)*3, 0);pout.putf("%t,%d接收到nLen1:%d字節(jié)\n", now(),id, nLen1);if (nLen1 <= 0){//客戶端關閉g_arrClientSocket[nIndex] = -1;FD_CLR(nClientSocket,&g_allset);::closesocket(nClientSocket);pout.putf("%t, %d關閉socket:%d\n", now(), id, nClientSocket);break;}int nSend = ::send(nClientSocket, (char*)buffer, nLen1, 0);pout.putf("%t,%d返回:%d字節(jié)\n",now(), id, nSend);}break;case MSG_QUIT:// MSG_QUIT is not used in our examplequit = true;break;}}catch(exception*){// the message object must be freed!delete msg;throw;}delete msg;} }int main() {//啟動線程池jobqueue jq;tobjlist<myjobthread> threads(true);// create the thread poolfor(int n = 0; n < maxthreads; n++){myjobthread* j = new myjobthread(n + 1, &jq);j->start();threads.add(j);}int nReady = 0;int nMaxfd = 0;int nMaxi = 0;int nSvrSocket;int nConSocket; //int nClientSocket;sockaddr_in clientaddr; //客戶端地址;int nCaLen; //客戶端地址長度int i; //循環(huán)用#ifdef WIN32WSADATA wsaData;int iError = WSAStartup(MAKEWORD(2,0), &wsaData);if (iError != 0){return false;}if ( LOBYTE( wsaData.wVersion ) < 2 ) {WSACleanup();return false; }//nSvrSocket = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); #endifnSvrSocket = ::socket(AF_INET, SOCK_STREAM, 0);if (nSvrSocket < 0){pout.putf("Couldn't create socket!\n");return false;}#ifndef WIN32// set SO_REAUSEADDR to true, unix only. on windows this option causes// the previous owner of the socket to give up, which is not desirable// in most cases, neither compatible with unix.int one = 1;if (::setsockopt(svrSocket, SOL_SOCKET, SO_REUSEADDR,(sockval_t)&one, sizeof(one)) != 0)pout.putf("Can't reuse local address!\n"); #endif// set up sockaddr_in and try to bind it to the socketsockaddr_in sa;memset(&sa, 0, sizeof(sa));sa.sin_family = AF_INET;sa.sin_port = htons(testport);sa.sin_addr.S_un.S_addr = INADDR_ANY;if (::bind(nSvrSocket, (sockaddr*)&sa, sizeof(sa)) != 0){#ifdef WIN32WSACleanup();#endifpout.putf("Couldn't bind address!\n");return -1;}if (::listen(nSvrSocket, SOMAXCONN) != 0){#ifdef WIN32WSACleanup();#endif pout.putf("Couldn't listen on socket!\n");return -1;}pout.putf("Ready to answer queries on port %d\n", testport);//初始化變量nMaxfd = nSvrSocket;nMaxi = 0;for (i =0; i < maxsetsize; i++){g_arrClientSocket[i] = -1;}fd_set set;FD_ZERO(&g_allset); //初始化給情空;FD_SET(nSvrSocket,&g_allset);//設定超時時間timeval t;t.tv_sec = timeout / 1000;t.tv_usec = (timeout % 1000) * 1000;while(true){set = g_allset;nReady = ::select(nMaxfd + 1, &set, nil, nil,(timeout < 0) ? nil : &t);if (nReady <= 0) continue;if (FD_ISSET(nSvrSocket, &set)){nCaLen = sizeof(clientaddr);nConSocket = ::accept(nSvrSocket,(sockaddr*)&clientaddr,&nCaLen);if (nConSocket < 0){pout.putf("Couldn't listen on socket!\n"); continue; }//打印連接的客戶端。pout.putf("%t連接進來的IP: %s! Port:%d\n",now(),inet_ntoa(clientaddr.sin_addr),ntohs((unsigned short)clientaddr.sin_port));//pout.putf("連接進來的IP: %s! Port:%d\n",inet_ntoa(AF_INET, &clientaddr.sin_addr, 4, NULL),ntohs(clientaddr.sin_port)); //設定超時時間和緩存區(qū)；int nNetTimeOut= 300000; //5分////int time, nlen;//::getsockopt(nConSocket, SOL_SOCKET, SO_SNDTIMEO, (char *)&time, &nlen);// pout.putf("timeout%d!\n",time); //::getsockopt(nConSocket, SOL_SOCKET, SO_RCVBUF, (char *)&time, &nlen);//pout.putf("SO_RCVBUF%d!\n",time);//::getsockopt(nConSocket, SOL_SOCKET, SO_SNDBUF, (char *)&time, &nlen);//pout.putf("SO_SNDBUF%d!\n",time);//::setsockopt(nConSocket, SOL_SOCKET, SO_SNDTIMEO, (char *)&nNetTimeOut, sizeof(nNetTimeOut));::setsockopt(nConSocket, SOL_SOCKET, SO_RCVTIMEO, (char *)&nNetTimeOut, sizeof(nNetTimeOut));int nBufLen = 1024;::setsockopt(nConSocket, SOL_SOCKET, SO_RCVBUF, (char *)&nBufLen, sizeof(nBufLen));::setsockopt(nConSocket, SOL_SOCKET, SO_SNDBUF, (char *)&nBufLen, sizeof(nBufLen));for (i = 0; i < maxsetsize; i++){ if (g_arrClientSocket[i] < 0){g_arrClientSocket[i] = nConSocket;break;} }if (i == maxsetsize -1){//太多的連接pout.putf("too many clients!\n"); }FD_SET(nConSocket, &g_allset); //把新的連接加到集合里。if (nConSocket > nMaxfd){nMaxfd = nConSocket;}if (i > nMaxi)nMaxi = i; /* max index in client[] array */if (--nReady <= 0)continue; /* no more readable descriptors */}pout.putf("\nbegin:%t:%d %d\n", now(), msecs(now()),nMaxi);for (i = 0; i <= nMaxi; i++){//nClientSocket = g_arrClientSocket[i];if (g_arrClientSocket[i]< 0) continue;if (FD_ISSET(g_arrClientSocket[i], &set)){//接收和發(fā)送pout.putf("%t拋 socket 到線程池%d:%d!\n",now(),i,g_arrClientSocket[i]); struct SocketInfo* pSocketInfo = new SocketInfo;pSocketInfo->nIndex = i;pSocketInfo->nSocket = g_arrClientSocket[i];jq.post(new myjob(pSocketInfo));if (--nReady <= 0)break; /* no more readable descriptors */}}//pout.putf("end:%t:%d\n", now(), msecs(now()));}closesocket(nSvrSocket);#ifdef WIN32WSACleanup();#endifreturn 0;};

總結

以上是生活随笔為你收集整理的select + 线程池 回应服务器(windows)的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。