面向連接的Socket Server的簡單實現（簡明易懂）

阿新 • • 發佈：2017-09-30

nta on() img == con 線程關閉 sse 每一個 sockaddr

http://www.cnblogs.com/worldtraveler/p/4685977.html

一、基本原理

有時候我們需要實現一個公共的模塊，需要對多個其他的模塊提供服務，最常用的方式就是實現一個Socket Server，接受客戶的請求，並返回給客戶結果。

這經常涉及到如果管理多個連接及如何多線程的提供服務的問題，常用的方式就是連接池和線程池，基本流程如下：

技術分享

首先服務器端有一個監聽線程，不斷監聽來自客戶端的連接。

當一個客戶端連接到監聽線程後，便建立了一個新的連接。

監聽線程將新建立的連接放入連接池進行管理，然後繼續監聽新來的連接。

線程池中有多個服務線程，每個線程都監聽一個任務隊列，一個建立的連接對應一個服務任務，當服務線程發現有新的任務的時候，便用此連接向客戶端提供服務。

一個Socket Server所能夠提供的連接數可配置，如果超過配置的個數則拒絕新的連接。

當服務線程完成服務的時候，客戶端關閉連接，服務線程關閉連接，空閑並等待處理新的任務。

連接池的監控線程清除其中關閉的連接對象，從而可以建立新的連接。

二、對Socket的封裝

Socket的調用主要包含以下的步驟：

技術分享

調用比較復雜，我們首先區分兩類Socket，一類是Listening Socket，一類是Connected Socket.

Listening Socket由MySocketServer負責，一旦accept，則生成一個Connected Socket，又MySocket負責。

MySocket主要實現的方法如下：

int MySocket::write(const char * buf, int length)
{
int ret = 0;
int left = length;
int index = 0;
while(left > 0)
{
ret = send(m_socket, buf + index, left, 0);
if(ret == 0)
break;
else if(ret == -1)
{
break;
}
left -= ret;
index += ret;
}
if(left > 0)
return -1;
return 0;
}

int MySocket::read(char * buf, int length)
{
int ret = 0;
int left = length;
int index = 0;
while(left > 0)
{
ret = recv(m_socket, buf + index, left, 0);
if(ret == 0)
break;
else if(ret == -1)
return -1;
left -= ret;
index += ret;
}

return index;
}

int MySocket::status()
{
int status;
int ret;
fd_set checkset;
struct timeval timeout;

FD_ZERO(&checkset);
FD_SET(m_socket, &checkset);

timeout.tv_sec = 10;
timeout.tv_usec = 0;

status = select((int)m_socket + 1, &checkset, 0, 0, &timeout);
if(status < 0)
ret = -1;
else if(status == 0)
ret = 0;
else
ret = 0;
return ret;
}

int MySocket::close()
{
struct linger lin;
lin.l_onoff = 1;
lin.l_linger = 0;
setsockopt(m_socket, SOL_SOCKET, SO_LINGER, (const char *)&lin, sizeof(lin));
::close(m_socket);
return 0;
}

MySocketServer的主要方法實現如下：

int MySocketServer::init(int port)
{
if((m_socket = socket(AF_INET, SOCK_STREAM, 0)) == -1)
{
return -1;
}

struct sockaddr_in serverAddr;
memset(&serverAddr, 0, sizeof(struct sockaddr_in));
serverAddr.sin_addr.s_addr = htonl(INADDR_ANY);
serverAddr.sin_family = AF_INET;
serverAddr.sin_port = htons(port);

if(bind(m_socket, (struct sockaddr *)&serverAddr, sizeof(serverAddr)) == -1)
{
::close(m_socket);
return -1;
}

if(listen(m_socket, SOMAXCONN) == -1)
{
::close(m_socket);
return -1;
}

struct linger lin;
lin.l_onoff = 1;
lin.l_linger = 0;

setsockopt(m_socket, SOL_SOCKET, SO_LINGER, (const char *)&lin, sizeof(lin));
m_port = port;
m_inited = true;
return 0;
}

MySocket * MySocketServer::accept()
{
int sock;
struct sockaddr_in clientAddr;
socklen_t clientAddrSize = sizeof(clientAddr);
if((sock = ::accept(m_socket, (struct sockaddr *)&clientAddr, &clientAddrSize)) == -1)
{
return NULL;
}
MySocket* socket = new MySocket(sock);
return socket;
}

MySocket * MySocketServer::accept(int timeout)
{
struct timeval timeout;
timeout.tv_sec = timeout;
timeout.tv_usec = 0;

fd_set checkset;
FD_ZERO(&checkset);
FD_SET(m_socket, &checkset);

int status = (int)select((int)(m_socket + 1), &checkset, NULL, NULL, &timeout);
if(status < 0)
return NULL;
else if(status == 0)
return NULL;

if(FD_ISSET(m_socket, &checkset))
{
return accept();
}
}

三、線程池的實現

一個線程池一般有一個任務隊列，啟動的各個線程從任務隊列中競爭任務，得到的線程則進行處理：list<MyTask *> m_taskQueue;

任務隊列由鎖保護，使得線程安全：pthread_mutex_t m_queueMutex

任務隊列需要條件變量來支持生產者消費者模式：pthread_cond_t m_cond

如果任務列表為空，則線程等待，等待中的線程個數為：m_numWaitThreads

需要一個列表來維護線程池中的線程：vector<MyThread *> m_threads

每個線程需要一個線程運行函數：

void * __thread_new_proc(void *p)
{
((MyThread *)p)->run();
return 0;
}

每個線程由MyThread類負責，主要函數如下：

int MyThread::start()
{

pthread_attr_t attr;
pthread_attr_init(&attr);
pthread_attr_setschedpolicy(&attr, SCHED_FIFO);

int ret = pthread_create(&m_thread, &attr, thread_func, args);
pthread_attr_destroy(&attr);

if(ret != 0)
return –1;

}

int MyThread::stop()
{

int ret = pthread_kill(m_thread, SIGINT);

if(ret != 0)
return –1;
}

int MyThread::join()

{

int ret = pthread_join(m_thread, NULL);

if(ret != 0)

return –1;

}

void MyThread::run()

{

while (false == m_bStop)

{

MyTask *pTask = m_threadPool->getNextTask();

if (NULL != pTask)

{

pTask->process();

}

線程池由MyThreadPool負責，主要函數如下：

int MyThreadPool::init()
{

pthread_condattr_t cond_attr;
pthread_condattr_init(&cond_attr);
pthread_condattr_setpshared(&cond_attr, PTHREAD_PROCESS_SHARED);
int ret = pthread_cond_init(&m_cond, &cond_attr);
pthread_condattr_destroy(&cond_attr);

if (ret_val != 0)
return –1;

pthread_mutexattr_t attr;
pthread_mutexattr_init(&attr);
pthread_mutexattr_setpshared(&attr, PTHREAD_PROCESS_SHARED);
ret = pthread_mutex_init(&m_queueMutex, &attr);
pthread_mutexattr_destroy(&attr);

if (ret_val != 0)
return –1;

for (int i = 0; i< m_poolSize; ++i)
{
MyThread *thread = new MyThread(i+1, this);
m_threads.push_back(thread);
}

return 0;
}

int MyThreadPool::start()
{
int ret;
for (int i = 0; i< m_poolSize; ++i)
{
ret = m_threads[i]->start();
if (ret != 0)
break;
}

ret = pthread_cond_broadcast(&m_cond);

if(ret != 0)
return –1;
return 0;
}

void MyThreadPool::addTask(MyTask *ptask)
{
if (NULL == ptask)
return;

pthread_mutex_lock(&m_queueMutex);

m_taskQueue.push_back(ptask);

if (m_waitingThreadCount > 0)
pthread_cond_signal(&m_cond);

pthread_mutex_unlock(&m_queueMutex);
}

MyTask * MyThreadPool::getNextTask()
{
MyTask *pTask = NULL;

pthread_mutex_lock(&m_queueMutex);

while (m_taskQueue.begin() == m_taskQueue.end())
{
++m_waitingThreadCount;

pthread_cond_wait(&n_cond, &m_queueMutex);

--m_waitingThreadCount;
}

pTask = m_taskQueue.front();

m_taskQueue.pop_front();

pthread_mutex_unlock(&m_queueMutex);

return pTask;
}

其中每一個任務的執行由MyTask負責，其主要方法如下：

void MyTask::process()

{

//用read從客戶端讀取指令

//對指令進行處理

//用write向客戶端寫入結果

}

四、連接池的實現

每個連接池保存一個鏈表保存已經建立的連接：list<MyConnection *> * m_connections

當然這個鏈表也需要鎖來進行多線程保護：pthread_mutex_t m_connectionMutex;

此處一個MyConnection也是一個MyTask，由一個線程來負責。

線程池也作為連接池的成員變量：MyThreadPool * m_threadPool

連接池由類MyConnectionPool負責，其主要函數如下：

void MyConnectionPool::addConnection(MyConnection * pConn)
{

pthread_mutex_lock(&m_connectionMutex);

m_connections->push_back(pConn);

pthread_mutex_unlock(&m_connectionMutex);

m_threadPool->addTask(pConn);
}

MyConnectionPool也要啟動一個背後的線程，來管理這些連接，移除結束的連接和錯誤的連接。

void MyConnectionPool::managePool()
{

pthread_mutex_lock(&m_connectionMutex);

for (list<MyConnection *>::iterator itr = m_connections->begin(); itr!=m_connections->end(); )
{
MyConnection *conn = *itr;
if (conn->isFinish())
{
delete conn;
conn = NULL;
list<MyConnection *>::iterator pos = itr++;
m_connections->erase(pos);
}
else if (conn->isError())
{

//處理錯誤的連接
++itr;
}
else
{
++itr;
}
}

pthread_mutex_unlock(&m_connectionMutex);

}

五、監聽線程的實現

監聽線程需要有一個MySocketServer來監聽客戶端的連接，每當形成一個新的連接，查看是否超過設置的最大連接數，如果超過則關閉連接，如果未超過設置的最大連接數，則形成一個新的MyConnection，將其加入連接池和線程池。

MySocketServer *pServer = new MySocketServer(port);

MyConnectionPool *pPool = new MyConnectionPool();

while (!stopFlag)

{

MySocket * sock = pServer->acceptConnection(5);

if(sock != null)

{

if(m_connections.size > maxConnectionSize)

{

sock.close();

}

MyTask *pTask = new MyConnection();

pPool->addConnection(pTask);

}

面向連接的Socket Server的簡單實現（簡明易懂）

nta on() img == con 線程關閉 sse 每一個 sockaddr http://www.cnblogs.com/worldtraveler/p/4685977.html 一、基本原理有時候我們需要實現一個公共的模塊，需要對多個其他的模塊提供服務，最常用

面向連接的Socket Server的簡單實現（簡明易懂）

http://www.cnblogs.com/worldtraveler/p/4685977.html

一、基本原理

二、對Socket的封裝

三、線程池的實現

四、連接池的實現

五、監聽線程的實現

面向連接的Socket Server的簡單實現（簡明易懂）

使用自定義端口連接SQL Server的方法（轉載）

Python 3.5 連接Mysql數據庫（pymysql 方式）

js隨筆—日曆的簡單實現（無css）

Unity連接Sql數據庫（PC端）

利用spring+springMvc對單點登入(SSO)的簡單實現（含原始碼）

佇列的簡單實現（C語言）

歸併排序的簡單實現（c++ 版本）

viewpager輪播圖的實現（簡潔易懂）

Adaboost演算法原理分析和例項+程式碼（簡明易懂）

Unity3D UGUI攝像機跟隨主角（簡明易懂）NGUI下也可以參考使用

vSphere Web Client無法連接vCenter Server問題（5.5）

java數據庫連接池技術簡單使用

遇到的vSphere Client無法連接vSphere server的問題

Java連接sql server失敗

關於Android手機MTP模式連接的一些設置（win7和ubuntu下，以紅米1s為例）

在windows 上的RedisClient 上連接遠程linux redis （"jave.net.ConnectException: Connection refused:connect"）

php5.6 連接SQL SERVER

C#連接SQL Server數據庫

從頭認識Spring-3.8 簡單的AOP日誌實現（註解版）-擴展添加檢查訂單功能，以便記錄並檢測輸入的參數

面向連接的Socket Server的簡單實現（簡明易懂）

http://www.cnblogs.com/worldtraveler/p/4685977.html

一、基本原理

二、對Socket的封裝

三、線程池的實現

四、連接池的實現

五、監聽線程的實現

相關推薦