计算机网络 -- 多人聊天室

一程序介绍和核心功能

这是基于 UDP 协议实现的一个网络程序，主要功能是 构建一个多人聊天室，当某个用户发送消息时，其他用户可以立即收到，形成一个群聊。

这个程序由一台服务器和n个客户端组成，服务器扮演了一个接受信息和分发信息的角色，将信息发送给所有已知的用户主机。

二程序结构

将服务器接收消息看作生产商品、分发消息看作消费商品，这不就是一个生动形象的 「生产者消费者模型」 吗？

「生产者消费者模型」 必备 321

3：三组关系
2：两个角色
1：一个交易场所

其中两个角色可以分别创建两个线程，一个负责接收消息，放入 「生产者消费者模型」，另一个则是负责从 「生产者消费者模型」 中拿去消息，分发给用户主机。

这对我们客户端也有相似的地方，但是与服务器不同，我们每个客户端都认为自己只需要与服务器 1对1 连接就可以了，因此我们 每个客户端都只需要即使接收和发送资源就可以了，只需要创建两个线程即可。

这里的交易场所可以选则 阻塞队列，也可以选择 环形队列。

三服务器

在引入 「生产者消费者模型」 后，服务器头文件结构将会变成下面这个样子

启动服务器，原初始化服务器、启动线程
接收消息，将收到的消息存入环形队列
发送消息，从环形队列中获取消息，并派发给线程

3.1 引入生产者消费者模型

这里我们直接使用一个vector数组模拟实现环形队列，同时借用信号量实现生产者消费者模型。

RingQueue.hpp 头文件

#pragma once#include <vector>
#include <semaphore.h>namespace My_RingQueue
{
const int DEF_CAP=10;template<class T>class RingQueue{public:RingQueue(size_t cap = DEF_CAP):_cap(cap),_pro_step(0),_con_step(0){_queue.resize(_cap);// 初始化信号量sem_init(&_pro_sem, 0, _cap);sem_init(&_con_sem, 0, 0);}~RingQueue(){// 销毁信号量sem_destroy(&_pro_sem);sem_destroy(&_con_sem);}// 生产商品void Push(const T &inData){// 申请信号量P(&_pro_sem);// 生产_queue[_pro_step++] = inData;_pro_step %= _cap;// 释放信号量V(&_con_sem);}// 消费商品void Pop(T *outData){// 申请信号量P(&_con_sem);// 消费*outData = _queue[_con_step++];_con_step %= _cap;// 释放信号量V(&_pro_sem);}private:void P(sem_t *sem){sem_wait(sem);}void V(sem_t *sem){sem_post(sem);}private:std::vector<T> _queue; //这个环形队列我们直接使用数组实现size_t _cap;sem_t _pro_sem; //生产者信号量sem_t _con_sem;  //消费者信号量size_t _pro_step; // 生产者下标size_t _con_step; // 消费者下标};
}

3.2 客户端代码

3.2.1 引入用户信息

在首次接收到某个用户的信息时，需要将其进行标识，以便后续在进行消息广播时分发给他

有点类似于用户首次发送消息，就被拉入了 “群聊”。

目前可以使用 IP + Port 的方式标识用户，确保用户的唯一性，这里选取 unordered_map 这种哈希表结构，方便快速判断用户是否已存在

key：用户标识符
value：用户客户端的 sockaddr_in 结构体

注意： 这里的哈希表后面会涉及多线程的访问，需要加锁保护。

3.2.2 LockGuard小组件

利用RAII思想实现锁的自动化

#pragma once#include<pthread.h>class LockGuard{public:LockGuard(pthread_mutex_t *pmtx):_mtx(pmtx){pthread_mutex_lock(_mtx);}~LockGuard(){pthread_mutex_unlock(_mtx);}private:pthread_mutex_t *_mtx;
};

3.2.3 Thread.hpp头文件

用自己的线程库

#pragma once#include<iostream>
#include<string>
#include<pthread.h>
#include<functional>enum class Status{NEW=0,//代表新建线程RUNNING,//代表运行EXIT //已退出线程
};
// 参数、返回值为 void 的函数类型
//typedef void (*func_t)(void*);
using func_t = std::function<void(void*)>;  // 使用包装器设定函数类型class Thread{
public:Thread(int num=0,func_t func=nullptr,void *args=nullptr):_tid(0),_status(Status::NEW),_func(func),_args(args){//写入线程名字char name[128];snprintf(name,sizeof name,"thraed-%d",num);_name=name;}~Thread(){}//获取线程idpthread_t getTID() const{return _tid;}//获取线程名字std::string getName() const{return _name;}//获取线程状态Status getStatus() const{return _status;}// 回调方法static void* runHelper(void* args){Thread* myThis = static_cast<Thread*>(args);// 很简单，回调用户传进来的 func 函数即可myThis->_func(myThis->_args);return nullptr;}// 启动线程void run(){int ret = pthread_create(&_tid, nullptr, runHelper, this);if(ret != 0){std::cerr << "create thread fail!" << std::endl;exit(1); // 创建线程失败，直接退出}_status =  Status::RUNNING; // 更改状态为 运行中}// 线程等待void join(){int ret = pthread_join(_tid, nullptr);if(ret != 0){std::cerr << "thread join fail!" << std::endl;exit(1); // 等待失败，直接退出}_status = Status::EXIT; // 更改状态为 退出}
private:pthread_t _tid; // 线程 IDstd::string _name; // 线程名Status _status; // 线程状态func_t _func; // 线程回调函数void* _args; // 传递给回调函数的参数
};

3.2.4 server.hpp 代码

#include<iostream>
#include<string>
#include<functional>
#include<cerrno>
#include<cstring>
#include<cstdlib>
#include<sys/types.h>
#include<sys/socket.h>
#include<netinet/in.h>
#include<arpa/inet.h>
#include"err.hpp"
#include"RingQueue.hpp"
#include<unordered_map>
#include"Thread.hpp"
#include"LockGuard.hpp"
#include<cstdio>namespace My_server{//端口号默认值const uint16_t default_port=8888;class server{private:/* data */int _sock;// 服务端套接字uint16_t _port;//端口号My_RingQueue::RingQueue<std::string> _rq; //阻塞队列std::unordered_map<std::string, struct sockaddr_in> _userTable; // <用户标识符, sockaddr_in 结构体>pthread_mutex_t _mtx; // 互斥锁，保护哈希表Thread* _producer;//生产者线程Thread* _consumer;//消费者线程public:server(uint16_t port=default_port):_port(port){pthread_mutex_init(&_mtx,nullptr);//创建线程，因为类内成员有隐含的this指针，需要bind固定该参数_producer = new Thread(1,std::bind(&server::RecvMessage,this));_consumer = new Thread(2,std::bind(&server::BroadcastMessage,this));}~server(){//等待线程结束_producer->join();_consumer->join();//销毁互斥锁pthread_mutex_destroy(&_mtx);//释放对象delete _producer;delete _consumer;}//初始化服务器void StartServer(){//1 创建套接字_sock = socket(AF_INET,SOCK_DGRAM,0);if(_sock==-1){std::cout<<"Create Socket Fail:: "<<strerror(errno)<<std::endl;exit(SOCKET_ERR);}//创建成功std::cout<<"Create Success Socket: "<<_sock<<std::endl;//2. 绑定IP地址和端口号struct sockaddr_in local;bzero(&local,sizeof(local));// 将结构体内容置0//填充字段local.sin_family= AF_INET; //设置为网络通信local.sin_port=htons(_port);//主机序列转换为网络序列local.sin_addr.s_addr=INADDR_ANY; //服务器端要绑定任何可用IP//绑定 IP 地址和端口号if(bind(_sock,(const sockaddr*)&local,sizeof(local))){std::cout<<"Bind IP&&Port Fail: "<<strerror(errno)<<std::endl;exit(BIND_ERR);}//绑定成功std::cout<<" Bind IP&&Port Success"<< std::endl;_producer->run();_consumer->run();}//接收信息void RecvMessage(){//服务器不断运行，使用需要使用 一个whilc(true) 死循环char buff[1024];while(true){//1 作为客户端 要接收信息 struct sockaddr_in peer;// 客户端结构体socklen_t len = sizeof(peer); //客户端结构体大小ssize_t n=recvfrom(_sock,buff,sizeof(buff)-1,0,(struct sockaddr*)&peer,&len);if(n>0){buff[n]='\0';}else{continue;}//2. 处理数据std::string clientIp=inet_ntoa(peer.sin_addr);// 获取服务端IP地址uint16_t clientPort = ntohs(peer.sin_port);// 获取端口号printf("Server get message from [%s:%d]$ %s\n",clientIp.c_str(),clientPort,buff);//3 判断是否在聊天室加入该用户std::string user = clientIp + "-" + std::to_string(clientPort);//花括号作用域内使用锁 限定RAII锁的作用域{LockGuard lockguard(&_mtx);if(_userTable.count(user)==0){ //首次出现，加入用户表_userTable[user]=peer;}}//4 将信息添加至环形队列std::string msg="["+ clientIp +":"+std::to_string(clientPort)+"] say#" + buff;_rq.Push(msg);}}// 广播消息void BroadcastMessage(){while(true) {// 1.从环形队列中获取消息std::string msg;_rq.Pop(&msg);// 2.将消息发给用户// TODOstd::vector<sockaddr_in> arr;{LockGuard lockguard(&_mtx);for(auto &user:_userTable){arr.push_back(user.second);}}for(auto &add:arr){//向客户端发送信息sendto(_sock,msg.c_str(),msg.size(),0,(const sockaddr*)&add,sizeof(add));}} }};}

3.2.5 server.cc源文件

几乎不需要更改

#include<memory>
#include"server.hpp"using namespace My_server;int main()
{std::unique_ptr<server> msvr(new server());//初始化服务器msvr->StartServer();return 0;
}

四客户端

有了之前 server.hpp 服务器头文件多线程化的经验后，改造 client.hpp 客户端头文件就很简单了，同样是创建两个线程，一个负责发送消息，一个负责接收消息

4.1 client.hpp头文件

#pragma once#include<iostream>
#include <string>
#include "err.hpp"
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <cstdlib>
#include <cstring>namespace My_client{class client{private:/* data */std::string server_ip;//服务端 IP 地址uint16_t server_port;//服务器端口号int _sock;struct sockaddr_in _svr;public://构造函数client(const std::string& ip,uint16_t port):server_ip(ip),server_port(port){}//析构函数~client(){}// 初始化客户端void InitClient() {//1. 创建套接字_sock=socket(AF_INET,SOCK_DGRAM,0);if(_sock==-1){std::cout << "Create Socket Fail: " << strerror(errno) << std::endl;exit(SOCKET_ERR);}std::cout<<"Create Success Socket:"<<_sock<<std::endl;//2. 构建服务器的sockaddr_in 结构体信息bzero(&_svr,sizeof(_svr));_svr.sin_family=AF_INET;// 绑定服务器IP地址_svr.sin_addr.s_addr=inet_addr(server_ip.c_str());//绑定服务器端口号_svr.sin_port=htons(server_port);}// 启动客户端void StartClient() {char buff[1024];// 1. 启动客户端while(true){std::string msg;std::cout<<"Input Message# ";std::getline(std::cin,msg);ssize_t n=sendto(_sock,msg.c_str(),msg.size(),0,(const struct sockaddr*)&_svr, sizeof(_svr));if(n==-1){std::cout<<"Send Message Fail: "<<strerror(errno)<<std::endl;continue;}//2 因为是回响 使用也要接收信息socklen_t len = sizeof(_svr);n = recvfrom(_sock,buff,sizeof(buff)-1,0,(struct sockaddr *)&_svr,&len);if(n>0){buff[n]='\0';}else{continue;}//可以再次获取 IP地址和 端口号std::string ip=inet_ntoa(_svr.sin_addr);uint16_t port=ntohs(_svr.sin_port);printf("Client get message from [%s:%d]# %s\n",ip.c_str(), port, buff);}}};
}

`4.2 client.cc` 客户端源文件

#include<memory>
#include"client.hpp"
#include"err.hpp"using namespace My_client;void Usage(const char* program){std::cout<<"Usage:"<<std::endl;std::cout<<"\t"<<program<<"ServerIP ServerPort" << std::endl;
}int main(int argc,char *argv[]){if(argc!=3){//启动方式是错误的，提升错误信息Usage(argv[0]);return USAGE_ERR; }std::string ip = argv[1];uint16_t port = std::stoi(argv[2]);std::unique_ptr<client> mcit(new client(ip,port));//启动客户端mcit->StartClient();return 0;
}

示例：