c++ - 具有多个客户端的服务器 : how to forward message from one client to another

Question

我有一个可以处理多个客户端(每个客户端一个线程)的工作服务器 - 改编自 here .目前它是这样工作的:

1. 客户端连接到服务器
2. 用户在发送到服务器的客户端控制台上键入消息(并显示在服务器端控制台上。
3. 用户在服务器控制台上键入消息，该消息被发送回同一客户端。

但我想做的是从例如客户端1，到服务器处理，然后可能转发给客户端3(将消息转发给哪个客户端由服务器决定)。

我的猜测是我需要修改clientHandleThread，但我不知道我需要做什么。我也不确定是否可以从当前线程访问单独的线程。

我是套接字编程和线程的新手，正在努力学习，因此欢迎任何帮助!我附上包含 main() 的代码(如果我应该附上其他代码，请告诉我!)

myLog winLog;

DWORD WINAPI clientHandleThread(LPVOID threadInfo)
{
    //this structure will contain all the data this callback will work on
    myThreadArgument* clientArgument = (myThreadArgument*)threadInfo;

    //the semamphore to protect the access to the std output
    mySemaphore* coutSemaphore = clientArgument->getCoutSemaphore();

    /*get the client connection: receiving messages from client and
    sending messages to the client will all be done by using
    this client connection*/
    myTcpSocket* clientConnection = clientArgument->getClientConnect();
    string clientName = clientArgument->getHostName();

    //the server is communicating with this client here
    while(1)
    {
        string messageFromClient = "";

        //receive from the client

        int numBytes = clientConnection->receiveMessage(messageFromClient);
        if ( numBytes == -99 ) break;

        //write to the console and the log file, so lock the semaphore
        coutSemaphore->lock();

        cout   << "[RECV fr " << clientName << "]: " << messageFromClient << endl;
        winLog << "[RECV fr " << clientName << "]: " << messageFromClient << endl;

        msgInfo proMsgFrCli = msgClassification(messageFromClient);

        //if the client wants to discount
        if ( messageFromClient.compare("quit") == 0 || messageFromClient.compare("Quit") == 0 )
        {
            coutSemaphore->unlock();
            break;
        }
        else // send to the client
        {
            char messageToClient[MAX_MSG_LEN+1];
            memset(messageToClient,0,sizeof(messageToClient));
            cout << "[SEND to " << clientName << "]: ";
            cin.getline(messageToClient,MAX_MSG_LEN);
            winLog << "[SEND to " << clientName << "]: " << messageToClient << endl;
            clientConnection->sendMessage(string(messageToClient));
            coutSemaphore->unlock();
        }
    }

    // if we reach here, this session with the client is done,
    // so we set the event on this thread to inform the main
    // control that this session is finished
    clientArgument->getExitEvent()->setEvent();
    return 1;
}

DWORD WINAPI serverHandleThread(LPVOID threadInfo) //server thread
{
    //this structure will contain all the data this callback will work on
    myThreadArgument* serverArgument = (myThreadArgument*)threadInfo;

    //the semamphore to protect the access to the std output
    mySemaphore* coutSemaphore = serverArgument->getCoutSemaphore();

    //get the server
    myTcpSocket* myServer = serverArgument->getClientConnect();
    string serverName = serverArgument->getHostName();

    //bind the server to the socket
    myServer->bindSocket();
    cout   << endl << "server finishes binding process... " << endl;
    winLog << endl << "server finishes binding process... " << endl;

    //server starts to wait for client calls
    myServer->listenToClient();
    cout   << "server is waiting for client calls ... " << endl;
    winLog << "server is waiting for client calls ... " << endl;

    //server starts to listen, and generates a thread to handle each client
    myThreadArgument* clientArgument[MAX_NUM_CLIENTS];
    myThread* clientHandle[MAX_NUM_CLIENTS];
    for ( int i = 0; i < MAX_NUM_CLIENTS; i++ )
    {
        clientArgument[i] = NULL;
        clientHandle[i] = NULL;
    }
    int currNumOfClients = 0;
    char buffer [100]; //temp buffer to convert currNumOfClients to char

    while ( 1 )
    {
        //wait to accept a client connection.  
        //processing is suspended until the client connects
        myTcpSocket* client; //connection dedicated for client communication
        string clientName; //client name 
        client = myServer->acceptClient(clientName);    
        clientName = clientName + "-" + itoa(currNumOfClients, buffer, 10);//char(65+currNumOfClients);

        //lock the std out so we can write to the console
        coutSemaphore->lock();
        cout   << endl << endl << "==> a client from [" << clientName << "] is connected!" << endl;
        winLog << endl << "==> a client from [" << clientName << "] is connected!" << endl << endl;

        coutSemaphore->unlock();

        //for this client, generate a thread to handle it
        if ( currNumOfClients < MAX_NUM_CLIENTS-1 )
        {
            clientArgument[currNumOfClients] = new myThreadArgument(client,coutSemaphore,clientName);
            clientHandle[currNumOfClients] = new myThread(clientHandleThread,(void*)clientArgument[currNumOfClients]);
            serverArgument->addClientArgument(clientArgument[currNumOfClients]);
            clientHandle[currNumOfClients]->execute();
            currNumOfClients++;
        }
    }

    return 1;
}

int main()
{
    /*build a semaphore so we can synchronize the access to std cout
    also includes the log file*/
    mySemaphore coutSemaphore(string(""),1);

    //initialize the winsock library
    myTcpSocket::initialize();

    /*create the server: local host will be used as the server, let us 
    first use myHostInfo class to show the name and IP address 
    of the local host*/
    winLog << endl;
    winLog << "retrieve the local host name and address:" << endl;

    myHostInfo serverInfo;
    string serverName = serverInfo.getHostName();
    string serverIPAddress = serverInfo.getHostIPAddress();
    cout << "my localhost (server) information:" << endl;
    cout << "   name:    " << serverName << endl;
    cout << "   address: " << serverIPAddress << endl;
    winLog << "     ==> name: " << serverName << endl;
    winLog << "     ==> address: " << serverIPAddress << endl;

    //open socket on the local host(server) and show its configuration
    myTcpSocket myServer(PORTNUM);
    cout   << myServer;
    winLog << myServer;

    //read connectivityFile
    neighbourInfo = connFrFile(numberOfFiles, intBtwnChange);
    //read routingFile
    nextHopInfo = routFrFile(numberOfFiles, intBtwnChange);

    /*create a thread to implement server process: listening to the socket,
    accepting client calls and communicating with clients. This will free the 
    main control (see below) to do other process*/
    myThreadArgument* serverArgument = new myThreadArgument(&myServer,&coutSemaphore,serverName);
    myThread* serverThread = new myThread(serverHandleThread,(void*)serverArgument);
    serverThread->execute();

    // main control: since the serverThread is handling the server functions,
    // this main control is free to do other things.
    while ( 1 )
    {
        /*do whatever you need to do here, I am using Sleep(x) 
        to make a little delay, pretending to be the other 
        possible processings*/
        Sleep(50000);

        //report the server status
        coutSemaphore.lock();
        cout   << endl << "-----------------------------------------------------------------" << endl;
        winLog << endl << "-----------------------------------------------------------------" << endl;
        cout   << "server (name:" << serverName << ") status report:" << endl;
        winLog << "server (name:" << serverName << ") status report:" << endl;
        cout   << "   the following clients have successfully connected with server: " << endl;
        winLog << "   the following clients have successfully connected with server: " << endl;
        for ( int i = 0; i < MAX_NUM_CLIENTS; i ++ )
        {
            myThreadArgument* clientInfo = serverArgument->getClientArgument(i);
            if ( clientInfo ) 
            {
                cout   << "         " << clientInfo->getHostName() << endl;
                winLog << "         " << clientInfo->getHostName() << endl;
            }
        }
        cout   << "   the following clients have shutdown the connection: " << endl;
        winLog << "   the following clients have shutdown the connection: " << endl;
        for ( int i = 0; i < MAX_NUM_CLIENTS; i ++ )
        {
            myThreadArgument* clientInfo = serverArgument->getClientArgument(i);
            if ( clientInfo && clientInfo->getExitEvent()->waitForEvent(0) )
            {
                clientInfo->setSignalToEnd(true);
                cout   << "         " << clientInfo->getHostName() << endl;
                winLog << "         " << clientInfo->getHostName() << endl;
            }
        }
        cout   << "-----------------------------------------------------------------" << endl << endl;
        winLog << "-----------------------------------------------------------------" << endl << endl;
        coutSemaphore.unlock();
    }

    return 1;
}

Answer 1

对于每个连接，您可以有两个队列:一个用于输入，一个用于输出。连接线程(如果每个连接有一个专用线程)从客户端读取输入并将其放入输入队列。连接线程还从输出队列中获取消息并将其发送到连接的客户端。

服务器然后有另一个线程，它遍历所有连接输入队列，提取消息，决定如何处理输入，然后可能将其放入其他连接的输出队列。

---

伪代码示例:

struct message_struct
{
    int source;       // Source where the message came from
    int destination;  // Destination client to send message on to
}

void client_thread()
{
    while (!exit_thread)
    {
        if (is_there_anything_to_recv())
        {
            // Receive and parse a message from the client
            message = receive();

            // Add to the threads input queue
            add_to_queue(input_queue, message);
        }

        // As long as there are messages in the output queue
        while (!queue_is_empty(output_queue))
        {
            // Remove one message from the queue
            message = remove_from_queue(output_queue);

            // And send it to the connected client
            send(message);
        }
    }
}

void server_thread()
{
    while (!exit_thread)
    {
        // Check for new connections
        // ...

        // Assuming the threads are on array (or array-like structure)
        for (i = 0; i < number_of_client_threads; i++)
        {
            // While the current threads (`i`)  input queue is not empty
            while (!queue_is_empty(client_threads[i].input_queue))
            {
                // Remove the message from the threads input queue
                message = remove_from_queue(client_threads[i].input_queue);

                // And add it to the destinations output queue
                add_to_queue(client_threads[message.destination].output_queue);
            }
        }
    }
}