//为了支持移植 沿用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字节\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字节\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;           //循环用#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)); //设定超时时间和缓存区；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)){//接收和发送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;};