在Windows socket I/O模型 之  select（1）中，我们只是在console中简单的模拟了select的处理方法。
还有很多特性不能修改，比如只能写，不能读。 没使用线程，也没有同步不同的读写线程。
先谈谈个人目前对select的理解。 select就是监控一组套接字的变化情况。 比如一个fd_set的变量（暂且定义为fdRead）里面有5个套接字，当你传给select后，假设只有2个被触发。 那么这个fdRead也就改变了。这是为什么select需要从全局fd_set拷贝一份给select的原因。
然后我们根据套接字的变化情况做相应的处理就OK了，但是大并发量还没有测试。
今晚我改进了。下面大家见代码。。。。。。 // Select_Server.cpp : Defines the entry point for the console application. // 服务端 #include "stdafx.h" #define STR_SERVER_IP "127.0.0.1" #define INT_SERVER_PORT 8001 #define INT_DATABUFFER_SIZE 256 SOCKET g_soClient; typedef std::list LstSocket; LstSocket g_lstSoClient; SOCKET g_soServer; fd_set g_fdSocketSet; // for thread synchronize CCriSec g_criSec; DWORD WINAPI ThreadRead(LPVOID lpvParam) { int iResult = 0; sockaddr_in addrAccept; int iAcceptLen = sizeof(addrAccept); SOCKET soClient; FD_ZERO(&g_fdSocketSet); FD_SET(g_soServer, &g_fdSocketSet); fd_set fdRead, fdWrite; while( TRUE ) { // initialize FD_ZERO(&fdRead); FD_ZERO(&fdWrite); fdRead = g_fdSocketSet; fdWrite = g_fdSocketSet; int n1 = fdRead.fd_count; int n2 = fdWrite.fd_count; int n3 = g_fdSocketSet.fd_count; int iResult = select(0, &fdRead, &fdWrite, NULL, NULL); if( iResult == SOCKET_ERROR) { break; } if(FD_ISSET(g_soServer, &fdRead)) { soClient = accept(g_soServer, (sockaddr*)&addrAccept, &iAcceptLen); CCriSecLock lock(g_criSec); if(soClient == INVALID_SOCKET) { continue; } else { printf(" [%s:%d] has connected to server! ", inet_ntoa(addrAccept.sin_addr), ntohs(addrAccept.sin_port)); FD_SET(soClient, &g_fdSocketSet); } } else { // check read for(int i=0; i < (int)fdRead.fd_count; i++) { if ( fdRead.fd_array[i] == g_soServer ) { continue; } if( FD_ISSET(fdRead.fd_array[i], &g_fdSocketSet) ) { sockaddr_in name; int namelen = sizeof(sockaddr_in); getpeername(fdRead.fd_array[i], (sockaddr *)&name, &namelen); char buf[256] = {0}; int len = 256; int ret = recv(fdRead.fd_array[i], buf, len, 0); CCriSecLock lock(g_criSec); if( ret == SOCKET_ERROR ) { int nErr = GetLastError(); if( nErr == 10054 ) { // Connection reset by peer. FD_CLR(fdRead.fd_array[i], &g_fdSocketSet); printf(" [%s:%d] disconnect from server. ", inet_ntoa(name.sin_addr), ntohs(name.sin_port) ); } else { printf(" fdread failed with %d ", nErr); } } else { printf(" Recv from [%s:%d] : %s ", inet_ntoa(name.sin_addr), ntohs(name.sin_port), buf); } } } // check write static bool b11 = false; for(int i=0; i < (int)fdWrite.fd_count; i++) { if( FD_ISSET(fdWrite.fd_array[i], &g_fdSocketSet) ) { char buf[256] = "abcd"; int len = 256; if( !b11 ) { b11 = true; //send(fdWrite.fd_array[i], buf, len ,0); } } } } } return 0; } DWORD WINAPI ThreadWrite(LPVOID lpvParam) { std::string str; { CCriSecLock lock(g_criSec); std::cout << "Please input message to client: "; } while( getline(std::cin, str) ) { if( str.compare("exit") == 0 ) { { CCriSecLock lock(g_criSec); printf("close write thread "); } break; } for(int i = 1; i < (int)g_fdSocketSet.fd_count; i++) { send(g_fdSocketSet.fd_array[i], str.data(), (int)str.size(), 0); } } return 0; } int _tmain(int argc, _TCHAR* argv[]) { WORD dwVersion = MAKEWORD(2, 2); WSAData wsaData; WSAStartup(WINSOCK_VERSION,&wsaData); g_soServer = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); if (INVALID_SOCKET == g_soServer) { printf("Failed to create socket! "); WSACleanup(); return -1; } sockaddr_in addrServer; memset(&addrServer,0,sizeof(sockaddr_in)); addrServer.sin_family = AF_INET; addrServer.sin_port = htons(INT_SERVER_PORT); addrServer.sin_addr.S_un.S_addr = htonl(INADDR_ANY); int iResult; bool bReuseAddr = true; iResult = setsockopt(g_soServer, SOL_SOCKET, SO_REUSEADDR, (char *)&bReuseAddr, sizeof(bReuseAddr)); if(SOCKET_ERROR == iResult) { printf("Failed to set resueaddr socket! "); WSACleanup(); return -1; } //设置非阻塞方式连接 unsigned long cmd = 1; iResult = ioctlsocket(g_soServer, FIONBIO, &cmd); iResult = bind(g_soServer, (sockaddr *)&addrServer, sizeof(addrServer)); if (SOCKET_ERROR == iResult) { printf("Failed to bind address! "); WSACleanup(); return -1; } if (0 != listen(g_soServer, 5)) { printf("Failed to listen client! "); WSACleanup(); return -1; } printf("Start server... "); HANDLE hWorkRead = CreateThread(NULL, 0, ThreadRead, NULL, 0, NULL); HANDLE hWorkWrite = CreateThread(NULL, 0, ThreadWrite, NULL, 0, NULL); ::WaitForSingleObject(hWorkRead, INFINITE); ::WaitForSingleObject(hWorkWrite, INFINITE); WSACleanup(); return 0; }
下面是客户端代码： // Select_Client.cpp : Defines the entry point for the console application. // #include "stdafx.h" #define INT_SERVER_PORT 8001 #define STR_SERVER_IP "127.0.0.1" #define INT_DATABUFFER_SIZE 256 #define STR_EXIT "exit" #define STR_RECV "recv" // SOCKET g_soClient; fd_set g_fdSocketSet; // for thread synchronize CCriSec g_criSec; DWORD WINAPI ThreadWorker(LPVOID lpvParam) { FD_ZERO(&g_fdSocketSet); FD_SET(g_soClient, &g_fdSocketSet); fd_set fdRead, fdWrite; while( TRUE ) { // initialize FD_ZERO(&fdRead); FD_ZERO(&fdWrite); fdRead = g_fdSocketSet; fdWrite = g_fdSocketSet; int iResult = select(0, &fdRead, &fdWrite, NULL, NULL); if( iResult == SOCKET_ERROR) { break; } else if( iResult == 0 ) { printf("Time limit expired "); } else { // check read if (FD_ISSET(fdRead.fd_array[0], &g_fdSocketSet)) { sockaddr_in name; int namelen = sizeof(sockaddr_in); getpeername(fdRead.fd_array[0], (sockaddr *)&name, &namelen); char buf[256] = {0}; int len = 256; int ret = recv(fdRead.fd_array[0], buf, len, 0); CCriSecLock lock(g_criSec); if( ret == SOCKET_ERROR ) { int nErr = GetLastError(); if( nErr == 10054 ) { // Connection reset by peer. FD_CLR(fdRead.fd_array[0], &g_fdSocketSet); printf( " [%s:%d] is closed. ", inet_ntoa(name.sin_addr), ntohs(name.sin_port) ); } else { printf("fdread failed with %d ", nErr); } } else { CCriSecLock lock(g_criSec); printf(" Recv from [%s:%d] : %s ", inet_ntoa(name.sin_addr), ntohs(name.sin_port), buf); } } // check write if (FD_ISSET(fdWrite.fd_array[0], &g_fdSocketSet)) { int a=2; int b=a; } } } return 0; } void main(void) { WSAData wsaData; WSAStartup(WINSOCK_VERSION,&wsaData); g_soClient = socket(AF_INET,SOCK_STREAM, IPPROTO_TCP); if (INVALID_SOCKET == g_soClient) { printf("Failed to create client! "); WSACleanup(); } sockaddr_in addrServer; addrServer.sin_addr.S_un.S_addr = inet_addr(STR_SERVER_IP); addrServer.sin_family = AF_INET; addrServer.sin_port = htons(INT_SERVER_PORT); int iResult; //设置非阻塞方式连接 //unsigned long ul = 1; //iResult = ioctlsocket(g_soClient, FIONBIO, (unsigned long*)&ul); iResult = connect(g_soClient, (sockaddr *)&addrServer, sizeof(sockaddr_in)); if (SOCKET_ERROR == iResult) { printf("Failed to connect server!(Error: %d) ", ::WSAGetLastError()); WSACleanup(); return; } HANDLE hWorker = CreateThread(NULL, 0, ThreadWorker, NULL, 0, NULL); std::string str; std::cout << "Please input message to server: "; while( getline(std::cin, str) ) { send(g_soClient, str.data(), str.size(), 0); std::cout << "Please input message to client: "; } closesocket(g_soClient); WSACleanup(); }
头文件 // stdafx.h : include file for standard system include files, // or project specific include files that are used frequently, but // are changed infrequently // #pragma once #ifndef _WIN32_WINNT // Allow use of features specific to Windows XP or later. #define _WIN32_WINNT 0x0501 // Change this to the appropriate value to target other versions of Windows. #endif #include #include #include #include #include #pragma comment(lib,"ws2_32.lib") // TODO: reference additional headers your program requires here #include "CriticalSection.h"
// CriticalSection.h /* * Copyright: JessMA Open Source (ldcsaa@gmail.com) * * Version : 2.3.2 * Author : Bruce Liang * Website : http://www.jessma.org * Project : https://github.com/ldcsaa * Blog : http://www.cnblogs.com/ldcsaa * Wiki : http://www.oschina.net/p/hp-socket * QQ Group : 75375912 * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ /**************************************************************************** * * * CriticalSection.h * * * * Create by : * * Kingfisher 2003-10-15 * * * * Description: * * 封装Win32临界量对象和互斥量内核对象 * ****************************************************************************/ #pragma once #include class CCriSec { public: CCriSec() {::InitializeCriticalSection(&m_crisec);} ~CCriSec() {::DeleteCriticalSection(&m_crisec);} void Lock() {::EnterCriticalSection(&m_crisec);} void Unlock() {::LeaveCriticalSection(&m_crisec);} private: CCriSec(const CCriSec& cs); CCriSec operator = (const CCriSec& cs); private: CRITICAL_SECTION m_crisec; }; class CCriSec2 { public: CCriSec2(BOOL bInitialize = TRUE) { if(bInitialize) { m_pcrisec = new CRITICAL_SECTION; ::InitializeCriticalSection(m_pcrisec); } else m_pcrisec = NULL; } ~CCriSec2() {Reset();} void Attach(CRITICAL_SECTION* pcrisec) { Reset(); m_pcrisec = pcrisec; } CRITICAL_SECTION* Detach() { CRITICAL_SECTION* pcrisec = m_pcrisec; m_pcrisec = NULL; return pcrisec; } void Lock() {::EnterCriticalSection(m_pcrisec);} void Unlock() {::LeaveCriticalSection(m_pcrisec);} private: CCriSec2(const CCriSec2& cs); CCriSec2 operator = (const CCriSec2& cs); void Reset() { if(m_pcrisec) { ::DeleteCriticalSection(m_pcrisec); delete m_pcrisec; m_pcrisec = NULL; } } private: CRITICAL_SECTION* m_pcrisec; }; template class CLocalLock { public: CLocalLock(CLockObj& obj) : m_lock(obj) {m_lock.Lock();} ~CLocalLock() {m_lock.Unlock();} private: CLockObj& m_lock; }; typedef CLocalLock CCriSecLock; typedef CLocalLock CCriSecLock2;