Linux socket編程（三）簡單的多線程聊天室

本文轉載自查看原文 2012-09-06 11:32 11417 Linux

要用到多線程以及線程的讀寫鎖，之前寫的Socket類、ServerSocket要做相應的改變

因為服務器端要維持着一個存儲客戶端Socket信息到數據結構，當多個線程同時訪問這個結構時，要做同步處理，所以要在適當的時候加上讀鎖或寫鎖。

新的ServerSocket類

#ifndef SERVERSOCKET_H
#define SERVERSOCKET_H

#include "Socket.h"
#include <list>
#include <semaphore.h>
#include "ThreadReadWriteLock.h"

using std::list;

class ServerSocket:public Socket
{
    public:
        ServerSocket(const int port);
        ServerSocket();
        virtual ~ServerSocket();

        void Accept(Socket& socket);
        //run server to connect multi-clients
        void Run();

    private:
        //accept multi-clients
        bool Accept();
        void AddClient(Socket* clientSocket);
        static void DeleteClient(Socket* clientSocket);
        static void* ProcessMessage(void* arg);
        static void SendMsgToAllUsers(const std::string& message);

        static list<Socket*> clientSockets;
        static bool serviceFlag;
        //use thread-read-write-lock to synchronize threads
        static ThreadReadWriteLock readWriteLock;
};

#endif

其中有static成員函數，因為創建一個新的線程時，要傳遞一個函數指針，不過類普通的成員函數的函數指針與一般的函數指針是不兼容的，所以要傳遞static成員函數的函數指針。

以下是ServerSocket的新實現：

ServerSocket.cpp

#include "ServerSocket.h"
#include "SocketException.h"
#include <pthread.h>
#include <iostream>

list<Socket*> ServerSocket::clientSockets;
ThreadReadWriteLock ServerSocket::readWriteLock;
bool ServerSocket::serviceFlag=true;

ServerSocket::ServerSocket(const int port)
{
      if ( ! Socket::Create() )
        {
          throw SocketException ( "Could not create server socket." );
        }

      if ( ! Socket::Bind ( port ) )
        {
          throw SocketException ( "Could not bind to port." );
        }

      if ( ! Socket::Listen() )
        {
          throw SocketException ( "Could not listen to socket." );
        }
}

ServerSocket::~ServerSocket()
{
    list<Socket*>::iterator iter;
    for(iter=clientSockets.begin();iter!=clientSockets.end();iter++)
        delete (*iter);
}

void ServerSocket::Accept(Socket& socket)
{
      if ( ! Socket::Accept ( socket ) )
        {
          throw SocketException ( "Could not accept socket." );
        }
}

bool ServerSocket::Accept()
{
    Socket* clientSocket=new Socket;
    Accept(*clientSocket);
    AddClient(clientSocket);

    //create new thread for a new client
    pthread_t newThread;
    int result=pthread_create(&newThread,NULL,ProcessMessage,static_cast<void*>(clientSocket));
    if(result!=0)
        return false;

    //detach the newThread
    //so when newThread exits it can release it's resource
    result=pthread_detach(newThread);
    if(result!=0)
        perror("Failed to detach thread");

    return true;
}

void ServerSocket::Run()
{
    while(serviceFlag)
    {
        if(clientSockets.size()>=static_cast<unsigned int>(MAXCONNECTION))
            serviceFlag=false;
        else
            serviceFlag=Accept();
        sleep(1);
    }
}

void* ServerSocket::ProcessMessage(void* arg)
{
    std::string message;
    Socket* clientSocket=static_cast<Socket*>(arg);

    Send(*clientSocket,"Welcome!");

    while(serviceFlag)
    {
        Receive(*clientSocket,message);
        if(message=="exit")
        {
            Send(*clientSocket,"user_exit");
            DeleteClient(clientSocket);
            break;
        }
        else
            SendMsgToAllUsers(message);
        sleep(1);
    }
    pthread_exit(NULL);
    return NULL;
}


void ServerSocket::AddClient(Socket* socket)
{
    if(readWriteLock.SetWriteLock())
    {
        clientSockets.push_back(socket);

        std::cout<<"Now "<<clientSockets.size()<<" users..";
        std::cout<<"New User: "<<socket->GetAddress()<<" "<<socket->GetPort()<<"\n";

        readWriteLock.UnLock();
    }
    else
        serviceFlag=false;
}

void ServerSocket::DeleteClient(Socket* socket)
{
    if(readWriteLock.SetWriteLock())
    {
        list<Socket*>::iterator iter;
        for(iter=clientSockets.begin();iter!=clientSockets.end();iter++)
            if((*iter)->GetAddress()==socket->GetAddress()
               && (*iter)->GetPort()==socket->GetPort())
            {
                //delete socket* in list
                delete (*iter);
                clientSockets.erase(iter);
                std::cout<<"Now "<<clientSockets.size()<<" users..\n";
                break;
            }
        readWriteLock.UnLock();
    }
    else
        serviceFlag=false;
}

接下來是讀寫鎖操作的封裝 ThreadReadWriteLock.h

這個類封裝了對線程讀寫鎖pthread_rwlock_t的操作，這些操作包括pthread_rwlock_init,pthread_rwlock_rlock,pthread_rwlock_wrlock,pthread_rwlock_unlock等

#ifndef THREADREADWRITELOCK_H
#define THREADREADWRITELOCK_H

#include <pthread.h>

class ThreadReadWriteLock
{
    public:
        ThreadReadWriteLock();
        ~ThreadReadWriteLock();

        bool SetReadLock();
        bool SetWriteLock();
        void UnLock();

    private:
        pthread_rwlock_t readWriteLock;
};

#endif