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

    一 程序介绍和核心功能

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

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

      

    二 程序结构 

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

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

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

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

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

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

     三 服务器

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

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

    3.1 引入生产者消费者模型 

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

    RingQueue.hpp 头文件

    1. #pragma once
    2.  
    3. #include 
    4. #include 
    5.  
    6. namespace My_RingQueue
    7. {
    8. const int DEF_CAP=10;
    9.  
    10.     template<class T>
    11.     class RingQueue
    12.     {
    13.     public:
    14.         RingQueue(size_t cap = DEF_CAP)
    15.             :_cap(cap)
    16.             ,_pro_step(0)
    17.             ,_con_step(0)
    18.         {
    19.             _queue.resize(_cap);
    20.             // 初始化信号量
    21.             sem_init(&_pro_sem, 0, _cap);
    22.             sem_init(&_con_sem, 0, 0);
    23.         }
    24.  
    25.         ~RingQueue(){
    26.             // 销毁信号量
    27.             sem_destroy(&_pro_sem);
    28.             sem_destroy(&_con_sem);
    29.         }
    30.  
    31.         // 生产商品
    32.         void Push(const T &inData){
    33.             // 申请信号量
    34.             P(&_pro_sem);
    35.             // 生产
    36.             _queue[_pro_step++] = inData;
    37.             _pro_step %= _cap;
    38.             // 释放信号量
    39.             V(&_con_sem);
    40.         }
    41.  
    42.         // 消费商品
    43.         void Pop(T *outData){
    44.             // 申请信号量
    45.             P(&_con_sem);
    46.             // 消费
    47.             *outData = _queue[_con_step++];
    48.             _con_step %= _cap;
    49.             // 释放信号量
    50.             V(&_pro_sem);
    51.         }
    52.  
    53.     private:
    54.         void P(sem_t *sem){
    55.             sem_wait(sem);
    56.         }
    57.  
    58.         void V(sem_t *sem){
    59.             sem_post(sem);
    60.         }
    61.  
    62.     private:
    63.         std::vector _queue; //这个环形队列我们直接使用数组实现
    64.         size_t _cap;
    65.         sem_t _pro_sem; //生产者信号量
    66.         sem_t _con_sem;  //消费者信号量
    67.         size_t _pro_step; // 生产者下标
    68.         size_t _con_step; // 消费者下标
    69.     };
    70. }

    3.2 客户端代码

    3.2.1 引入用户信息

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

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

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

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

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

    3.2.2 LockGuard小组件

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

    1. #pragma once
    2.  
    3. #include
    4.  
    5. class LockGuard{
    6.     public:
    7.  
    8.      LockGuard(pthread_mutex_t *pmtx)
    9.       :_mtx(pmtx)
    10.      {
    11.         pthread_mutex_lock(_mtx);
    12.      }
    13.  
    14.      ~LockGuard(){
    15.         pthread_mutex_unlock(_mtx);
    16.      }
    17.     private:
    18.     pthread_mutex_t *_mtx;
    19. };

    3.2.3 Thread.hpp头文件

    用自己的线程库

    1. #pragma once
    2.  
    3. #include
    4. #include
    5. #include
    6. #include
    7.  
    8. enum class Status{
    9.     NEW=0,//代表新建线程
    10.     RUNNING,//代表运行
    11.     EXIT //已退出线程
    12. };
    13. // 参数、返回值为 void 的函数类型
    14. //typedef void (*func_t)(void*);
    15. using func_t = std::function<void(void*)>;  // 使用包装器设定函数类型
    16.  
    17. class Thread{
    18. public:
    19.    Thread(int num=0,func_t func=nullptr,void *args=nullptr)
    20.      :_tid(0)
    21.      ,_status(Status::NEW)
    22.      ,_func(func)
    23.      ,_args(args)
    24.    {
    25.     //写入线程名字
    26.     char name[128];
    27.     snprintf(name,sizeof name,"thraed-%d",num);
    28.     _name=name;
    29.    }
    30.  
    31.    ~Thread(){}
    32.  
    33.    //获取线程id
    34.    pthread_t getTID() const{
    35.     return _tid;
    36.    }
    37.     
    38.    //获取线程名字
    39.    std::string getName() const{
    40.      return _name;
    41.    }
    42.    
    43.  
    44.    //获取线程状态
    45.    Status getStatus() const{
    46.     return _status;
    47.    }
    48.  
    49.      // 回调方法
    50.     static void* runHelper(void* args){
    51.         Thread* myThis = static_cast(args);
    52.         // 很简单，回调用户传进来的 func 函数即可
    53.         myThis->_func(myThis->_args);
    54.  
    55.         return nullptr;
    56.     }
    57.  
    58.      // 启动线程
    59.     void run(){
    60.         int ret = pthread_create(&_tid, nullptr, runHelper, this);
    61.         if(ret != 0){
    62.             std::cerr << "create thread fail!" << std::endl;
    63.             exit(1); // 创建线程失败，直接退出
    64.         }
    65.         _status =  Status::RUNNING; // 更改状态为 运行中
    66.     }
    67.  
    68.     // 线程等待
    69.     void join(){
    70.         int ret = pthread_join(_tid, nullptr);
    71.         if(ret != 0){
    72.             std::cerr << "thread join fail!" << std::endl;
    73.             exit(1); // 等待失败，直接退出
    74.         }
    75.         _status = Status::EXIT; // 更改状态为 退出
    76.     }
    77. private:
    78.     pthread_t _tid; // 线程 ID
    79.     std::string _name; // 线程名
    80.     Status _status; // 线程状态
    81.     func_t _func; // 线程回调函数
    82.     void* _args; // 传递给回调函数的参数
    83. };

    3.2.4 server.hpp 代码

    1. #include
    2. #include
    3. #include
    4. #include
    5. #include
    6. #include
    7. #include
    8. #include
    9. #include
    10. #include
    11. #include"err.hpp"
    12. #include"RingQueue.hpp"
    13. #include
    14. #include"Thread.hpp"
    15. #include"LockGuard.hpp"
    16. #include
    17.  
    18.  
    19. namespace My_server{
    20.  
    21.    //端口号默认值
    22.    const uint16_t default_port=8888;
    23.  
    24.    class server
    25.    {
    26.    private:
    27.     /* data */
    28.     int _sock;// 服务端套接字
    29.     uint16_t _port;//端口号
    30.     My_RingQueue::RingQueue _rq; //阻塞队列
    31.     std::unordered_mapstruct sockaddr_in> _userTable; // <用户标识符, sockaddr_in 结构体>
    32.  
    33.     pthread_mutex_t _mtx; // 互斥锁，保护哈希表
    34.  
    35.     Thread* _producer;//生产者线程
    36.     Thread* _consumer;//消费者线程
    37.  
    38.    public:
    39.     server(uint16_t port=default_port)
    40.      :_port(port)
    41.     {
    42.       pthread_mutex_init(&_mtx,nullptr);
    43.  
    44.        //创建线程，因为类内成员有隐含的this指针，需要bind固定该参数
    45.       _producer = new Thread(1,std::bind(&server::RecvMessage,this));
    46.       _consumer = new Thread(2,std::bind(&server::BroadcastMessage,this));
    47.     }
    48.     ~server(){
    49.  
    50.       //等待线程结束
    51.       _producer->join();
    52.       _consumer->join();
    53.  
    54.       //销毁互斥锁
    55.       pthread_mutex_destroy(&_mtx);
    56.      
    57.      //释放对象
    58.       delete _producer;
    59.       delete _consumer;
    60.     }
    61.  
    62.     //初始化服务器
    63.     void StartServer(){
    64.  
    65.         //1 创建套接字
    66.         _sock = socket(AF_INET,SOCK_DGRAM,0);
    67.         if(_sock==-1){
    68.             std::cout<<"Create Socket Fail:: "<<strerror(errno)<
    69.             exit(SOCKET_ERR);
    70.         }
    71.  
    72.         //创建成功
    73.         std::cout<<"Create Success Socket: "<<_sock<
    74.  
    75.         //2. 绑定IP地址和端口号
    76.         struct sockaddr_in local;
    77.         bzero(&local,sizeof(local));// 将结构体内容置0
    78.  
    79.         //填充字段
    80.         local.sin_family= AF_INET; //设置为网络通信
    81.         local.sin_port=htons(_port);//主机序列转换为网络序列
    82.         local.sin_addr.s_addr=INADDR_ANY; //服务器端要绑定任何可用IP
    83.  
    84.         //绑定 IP 地址和端口号
    85.         if(bind(_sock,(const sockaddr*)&local,sizeof(local))){
    86.             std::cout<<"Bind IP&&Port Fail: "<<strerror(errno)<
    87.             exit(BIND_ERR);
    88.         }
    89.         
    90.         //绑定成功
    91.         std::cout<<" Bind IP&&Port Success"<< std::endl;
    92.  
    93.         _producer->run();
    94.         _consumer->run();
    95.     }
    96.     
    97.    //接收信息
    98.     void RecvMessage(){
    99.  
    100.         //服务器不断运行，使用需要使用 一个whilc(true) 死循环
    101.         char buff[1024];
    102.         while(true){
    103.             //1 作为客户端 要接收信息 
    104.             struct sockaddr_in peer;// 客户端结构体
    105.             socklen_t len = sizeof(peer); //客户端结构体大小
    106.  
    107.             ssize_t n=recvfrom(_sock,buff,sizeof(buff)-1,0,(struct sockaddr*)&peer,&len);
    108.  
    109.             if(n>0){
    110.                 buff[n]='\0';
    111.             }
    112.             else{
    113.                 continue;
    114.             }
    115.  
    116.             //2. 处理数据
    117.             std::string clientIp=inet_ntoa(peer.sin_addr);// 获取服务端IP地址
    118.             uint16_t clientPort = ntohs(peer.sin_port);// 获取端口号
    119.             printf("Server get message from [%s:%d]$ %s\n",clientIp.c_str(),clientPort,buff);
    120.  
    121.             //3 判断是否在聊天室加入该用户
    122.             std::string user = clientIp + "-" + std::to_string(clientPort);
    123.             
    124.             //花括号作用域内使用锁 限定RAII锁的作用域
    125.             {
    126.               LockGuard lockguard(&_mtx);
    127.               if(_userTable.count(user)==0){ //首次出现，加入用户表
    128.                  _userTable[user]=peer;
    129.               }
    130.             }
    131.  
    132.             //4 将信息添加至环形队列
    133.             std::string msg="["+ clientIp +":"+std::to_string(clientPort)+"] say#" + buff;
    134.             _rq.Push(msg);
    135.     }
    136.     }
    137.       // 广播消息
    138.         void BroadcastMessage(){
    139.  
    140.             while(true) {
    141.                 // 1.从环形队列中获取消息
    142.                 std::string msg;
    143.                 _rq.Pop(&msg);
    144.  
    145.                 // 2.将消息发给用户
    146.                 // TODO
    147.                 std::vector arr;
    148.  
    149.                 {
    150.                  LockGuard lockguard(&_mtx);
    151.                  for(auto &user:_userTable){
    152.                     arr.push_back(user.second);
    153.                  }
    154.                 }
    155.  
    156.                 for(auto &add:arr){
    157.                     //向客户端发送信息
    158.                     sendto(_sock,msg.c_str(),msg.size(),0,(const sockaddr*)&add,sizeof(add));
    159.                 }
    160.             } 
    161.         }
    162.    };
    163.    
    164. }

    3.2.5 server.cc源文件

    几乎不需要更改

    1. #include
    2. #include"server.hpp"
    3.  
    4. using namespace My_server;
    5.  
    6. int main()
    7. {
    8.     std::unique_ptr msvr(new server());
    9.  
    10.     //初始化服务器
    11.     msvr->StartServer();
    12.  
    13.  
    14.     return 0;
    15. }

    四 客户端

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

    4.1 client.hpp头文件

    1. #pragma once
    2.  
    3. #include
    4. #include 
    5. #include "err.hpp"
    6. #include 
    7. #include 
    8. #include 
    9. #include 
    10. #include 
    11. #include 
    12.  
    13. namespace My_client{
    14.     
    15.     class client{
    16.       private:
    17.       /* data */
    18.       std::string server_ip;//服务端 IP 地址
    19.       uint16_t server_port;//服务器端口号
    20.       int _sock;
    21.       struct sockaddr_in _svr;
    22.       public:
    23.       //构造函数
    24.       client(const std::string& ip,uint16_t port)
    25.       :server_ip(ip)
    26.       ,server_port(port)
    27.       {}
    28.       //析构函数
    29.       ~client(){
    30.       }
    31.       // 初始化客户端
    32.       void InitClient() {
    33.         
    34.          //1. 创建套接字
    35.          _sock=socket(AF_INET,SOCK_DGRAM,0);
    36.          if(_sock==-1){
    37.            std::cout << "Create Socket Fail: " << strerror(errno) << std::endl;
    38.             exit(SOCKET_ERR);
    39.          }
    40.  
    41.          std::cout<<"Create Success Socket:"<<_sock<
    42.  
    43.          //2. 构建服务器的sockaddr_in 结构体信息
    44.          bzero(&_svr,sizeof(_svr));
    45.          _svr.sin_family=AF_INET;
    46.           // 绑定服务器IP地址
    47.          _svr.sin_addr.s_addr=inet_addr(server_ip.c_str());
    48.          //绑定服务器端口号
    49.          _svr.sin_port=htons(server_port);
    50.       }
    51.       // 启动客户端
    52.       void StartClient() {
    53.        
    54.        char buff[1024];
    55.         // 1. 启动客户端
    56.         while(true){
    57.         std::string msg;
    58.         std::cout<<"Input Message# ";
    59.         std::getline(std::cin,msg);
    60.  
    61.         ssize_t n=sendto(_sock,msg.c_str(),msg.size(),0,(const struct sockaddr*)&_svr, sizeof(_svr));
    62.  
    63.         if(n==-1){
    64.           std::cout<<"Send Message Fail: "<<strerror(errno)<
    65.            continue;
    66.         }
    67.  
    68.         //2 因为是回响 使用也要接收信息
    69.         socklen_t len = sizeof(_svr);
    70.          n = recvfrom(_sock,buff,sizeof(buff)-1,0,(struct sockaddr *)&_svr,&len);
    71.  
    72.          if(n>0){
    73.             buff[n]='\0';
    74.          }
    75.          else{
    76.             continue;
    77.          }
    78.          
    79.          //可以再次获取 IP地址和 端口号
    80.          std::string ip=inet_ntoa(_svr.sin_addr);
    81.          uint16_t port=ntohs(_svr.sin_port);
    82.  
    83.           printf("Client get message from [%s:%d]# %s\n",ip.c_str(), port, buff);
    84.       }
    85.       }
    86.     };
    87. }

    4.2 client.cc 客户端源文件

    1. #include
    2. #include"client.hpp"
    3. #include"err.hpp"
    4.  
    5.  
    6. using namespace My_client;
    7.  
    8. void Usage(const char* program){
    9.     std::cout<<"Usage:"<
    10.     std::cout<<"\t"<"ServerIP ServerPort" << std::endl;
    11. }
    12.  
    13. int main(int argc,char *argv[]){
    14.  
    15.     if(argc!=3){
    16.         //启动方式是错误的，提升错误信息
    17.         Usage(argv[0]);
    18.         return USAGE_ERR; 
    19.     }
    20.  
    21.     std::string ip = argv[1];
    22.     uint16_t port = std::stoi(argv[2]);
    23.  
    24.     std::unique_ptr mcit(new client(ip,port));
    25.  
    26.     //启动客户端
    27.     mcit->StartClient();
    28.  
    29.     return 0;
    30. }

    示例：

  • 相关阅读:
    docsify项目部署（华为云+宝塔+centos+docker+nginx）踩坑指南
    电子商务、搜索引擎
    Dell清除BIOS密码及硬盘锁
    JavaScript对象
    Go语学习笔记 - gorm使用 - 表增删改查 Web框架Gin（八）
    CUDA和cuDNN安装配置
    dubbo(四)异常处理
    JavaScript基础教程笔记（一）
    二十二、商城 - 商品录入-FastDFS（10）
    结冰过程渲染-Ovito实现
  • 原文地址：https://blog.csdn.net/sushhsishdgsusk/article/details/137990775