• 【Linux网络编程】Socket-UDP实例

    目录

    知识汇总：

    1.IP地址与端口号

    2.主机序列与网络序列

    3.多网卡/多IP

    接口函数：

    socket:

    bind：

    recvfrom:

    sendto:

    inet_addr:

    inet_ntoa:

    htons:

    popen：

    fgets:文件I/O与标准I/O

    文件I/O与标准I/O部分：

    代码：

    udp_client.cc

    udp_server.cc

    udp_server.hpp

    RingQueue.hpp

    lockGuard.hpp

    Thread.hpp

    makefile

    这份代码利用下面所有知识编写了一个简易聊天室（基于Linux操作系统）。虽然字数挺多其实并不复杂，这里如果能够看完或许会对你的知识进行一下串联，这篇文章比较杂并且网络编程这块知识需要用到系统编程的知识，希望能帮助到您。

    知识汇总：

    1.IP地址与端口号

    我们知道同一台主机的进程间通信有system V共享内存，消息队列，信号量这些方式，而跨主机的进程间通信怎么搞呢？使用IP地址与端口号！

    IP地址用来网络中标识唯一一台主机，是一个32位无符号整数，常常用192.163.1.1这样点分十进制的字符串形式表示。

    端口号用来表示一台主机中的一个进程，它是一个16位无符号整数，所以端口号最小是0，最大是65536。那么端口号如何表示一个进程呢？如下图​​​​，端口号作为数组的下标，数组中存放的是进程PID。它相当于一个哈希表，根据下标即端口号就可以找到对应的进程。

    这里有一个问题，为什么不直接用进程PID呢，非要多走一步端口号，感觉有点多此一举。

    我是这样理解的，我们使用的应用程序都是有对应的服务器维护的，我们作为一个客户端需要和服务器进行数据交互，那么就必须明白两个问题，一是服务器在哪，二是与服务器的哪个进程进行通信。当我们通信之前，就必须知道服务器的IP地址与进程PID，那么我们怎么知道呢？IP地址我们可以视为客户端提前知晓且并不变更，那进程PID呢？服务器每重新打开一次进程，PID会一样吗？显然不会，那么我怎么找到服务器的对应进程呢？这里就陷入了一个死循环。

    网络：请问您是要和服务器123.123.123.123通信吗？

    客户端：对的。

    网络：请告诉我你是要和服务器的哪个进程通信呢？

    客户端：不知道啊？它的进程每次重新启动，进程号都会变更。

    网络：对不起先生，没有进程号我们没法帮您通信。

    客户端：我不跟服务器通信我怎么知道服务器的进程号。

    当然只有ip地址也是可以接收到数据的，但是交由哪个进程处理，这些数据是什么意思用来干什么的，就成了问题。

    为了避免这个问题，就有了端口号的概念。服务器的相应进程会放到一个固定的端口号上，客户端都是提前知晓这个端口号的，所以在通信时，客户端只需要端口号就可以找到对应进程。这也使得许多端口号约定成俗，比如常见的8080端口。

    2.主机序列与网络序列

    每台计算机的存储顺序不同，分为大端存储和小端存储。大端存储就是低字节放到高地址，小端存储就是高字节放到低地址。如下图，定义一个int num=1；

    可以看到01放到高地址处的是大端存储，放到低地址处的是小端存储。 

    既然有这种主机存储顺序的不同，那么在进行网络通信时如果两个终端存储顺序不同，那么数据就会被错误解读。为了解决这个问题，就定义了一个共同的标准，在传输网络数据的时候都以大端存储为标准。

     因为客户端发送数据，携带的目的ip与目的端口都是网络序列的，服务器端要对比数据是给哪个端口，所以本地ip和端口必须转为网络序列。

    3.多网卡/多IP

    这块是关于创建套接字后，使用bind函数绑定端口号与ip的一个细节。

    云服务器，或者一款服务器不要bind一个具体的ip，因为服务器可能有多个网卡多个ip地址，这些ip都有可能接收指定端口的数据，所以需要在服务器启动的时候bind任意一个ip地址，这就要求在对sockaddr里面的sin_addr里面的s_addr初始化时，使用INADDR_ANY进行初始化。

    接口函数：

    socket:

    1. #include          
    2. #include 
    3.  
    4. int socket(int domain, int type, int protocol);

    socket函数用来创建一个套接字。domain选择协议家族来进行通信，ipv4网络通信使用AF_INET,ipv6使用AF_INET6。type是用来选择套接字类型的，  SOCK_STREAM就是面向连接，可靠的，SOCK_DGRAM就是无连接，不可靠的。protocol用0即可，选择默认合适的协议。socket创建成功会返回一个文件描述符，创建失败返回-1。 

    bind：

    1. #include          
    2. #include 
    3.  
    4. int bind(int sockfd, const struct sockaddr *addr,
    5.                 socklen_t addrlen);

    bind函数用来绑定本地主机ip与端口号。sockfd就是创建套接字成功返回的文件描述符。

    我们可以看到sockaddr是个结构体，那这个结构体的成员有哪些呢？

    sockaddr结构体:

    __SOCKADDR_COMMON (sa_)就是#define  __SOCKADDR_COMMON(sa_prefix) \

      sa_family_t sa_prefix##family，其实绕来绕去就是sa_family_t  sa_family,一个16位短整型变量（下面sockaddr_in结构体的第一个成员也大体一样，sa_family_t  sin_family，一个16位短整型变量）,用来表示地址类型如AF_INET。char sa_data[14]就是14字节的地址数据。

    不过我们在进行网络通信时，使用的是sockaddr_in类型的结构体

    sockaddr_in结构体：

    由上图可以看出sockaddr结构体里面的sin_port是一个16位无符号整数，in_addr结构体里面有唯一一个成员---32位无符号整数。他们分别代表一个端口号和IP地址。sin_zero结构体就是填充字段，可以看到用sockaddr结构体大小减去了sockaddr_in结构体里面的三个成员的大小，最后自然sockaddr和sockaddr_in结构体的大小就一样了。这不明摆着是让sockaddr和sockaddr_in适配么。使用时直接取地址然后强转就可以了。

    所以得出下面的结论：

    IPv4和IPv6的地址格式定义在netinet/in.h中,IPv4地址用sockaddr_in结构体表示,包括16位地址类型, 16位端口号和32位IP地址.

    IPv4、IPv6地址类型分别定义为常数AF_INET、AF_INET6. 这样,只要取得某种sockaddr结构体的首地址,不需要知道具体是哪种类型的sockaddr结构体,就可以根据地址类型字段确定结构体中的内容.

    socket API可以都用struct sockaddr *类型表示, 在使用的时候需要强制转化成sockaddr_in; 这样的好处是程序的通用性, 可以接收IPv4, IPv6, 以及UNIX Domain Socket各种类型的sockaddr结构体指针做为参数;

    addrlen就是一个无符号整形，指明sockaddr结构体大小的。

    recvfrom:

    1. #include 
    2. #include 
    3.  
    4. ssize_t recvfrom(int sockfd, void *buf, size_t len, int flags,
    5.                         struct sockaddr *src_addr, socklen_t *addrlen);

    利用创建的套接字把接收到的最大为len字节长度的数据放到buf中，flags标志位表示是否阻塞接收（设为0即可）,src_addr指针和addrlen指针分别指向一个输入性参数，用来接收发送方的IP地址端口号以及结构体大小。数据成功则返回实际接收到的字符数，失败返回-1。

    sendto:

    1. #include 
    2. #include 
    3.  
    4.  
    5. ssize_t sendto(int sockfd, const void *buf, size_t len, int flags,
    6.                       const struct sockaddr *dest_addr, socklen_t addrlen);

    利用创建的套接字发送最大为len字节长度的数据，flags标志位表示是否阻塞发送（设为0即可）,dest_addr指针指向一个sockaddr_in结构体（里面有目的ip和目的端口号），addrlen为该结构体大小。成功则返回实际传送出去的字符数，失败返回-1。

    inet_addr:

    1. #include 
    2. #include 
    3. #include 
    4.  
    5. in_addr_t inet_addr(const char *cp);

    inet_addr（） 函数将互联网主机地址 cp 字符串从 IPv4 数字和点表示法转换为按网络字节顺序的二进制数据。 

    inet_ntoa:

    1. #include 
    2. #include 
    3. #include 
    4.  
    5.      
    6. char *inet_ntoa(struct in_addr in);

    inet_ntoa（） 函数将按网络字节顺序给出的互联网主机地址转换为 IPv4 点分十进制表示法的字符串。 字符串以静态分配的缓冲区，后续调用将覆盖该缓冲区。不过这里的in_addr是sockaddr_in结构体里面的一个结构体成员，这个in_addr结构体里面存放的是一个32位无符号整数（IP地址）。

    htons:

    1. #include 
    2.  
    3.        uint32_t htonl(uint32_t hostlong);
    4.  
    5.        uint16_t htons(uint16_t hostshort);
    6.  
    7.        uint32_t ntohl(uint32_t netlong);
    8.  
    9.        uint16_t ntohs(uint16_t netshort);

    已知端口号是16位无符号整数，ip地址是32位无符号整数。所以这里四个函数就是把主机字节序转换成网络字节序or网络字节序转换成主机字节序，IP地址用uint32_t,端口号用uint16_t。

    popen：

    1.  #include 
    2.  
    3.        FILE *popen(const char *command, const char *type);
    4.  
    5.        int pclose(FILE *stream);

    command： 是一个指向以 NULL 结束的 shell 命令字符串的指针。这行命令将被传到 bin/sh 并使用 -c标志，shell 将执行这个命令。
    type： 只读或只写类型。如果 type 是 “r” 则文件指针连接到 command 的标准输出；如果 type 是 “w” 则文件指针连接到 command 的标准输入。 

    如果调用成功，则返回一个读或者打开文件的指针，如果失败，返回NULL。

    man手册中关于popen函数的解释：popen（） 函数通过创建管道、分叉和调用 shell 来打开进程。 由于管道根据定义是单向的，因此类型参数可以指定只有阅读或写作，而不是两者兼而有之;生成的流相应地是只读或只写的。

    popen（） 的返回值在所有方面都是正常的标准 I/O 流，除了它必须使用 pclose（） 而不是 fclose（3） 关闭。 写入这样的流写入命令的标准输入;该命令的标准输出与调用 popen（） 的进程的标准输出相同，除非命令对此进行了更改本身。相反，从“打开的”流中读取会读取命令的标准输出，并且命令的标准输入与进程的标准输入相同称为 popen（）。

    fopen函数：

    可以看到popen函数与fopen函数极其相似，都是标准I/O库函数，且返回值都是一个文件流指针（FILE*），都需要用close函数关闭。但是fopen函数是用于打开一个文件，而popen函数作用是创建管道并创建子进程，并利用子进程处理command命令，处理结果返回到一个文件。调用popen函数的进程就是父进程。

    fgets:文件I/O与标准I/O

    1.  #include 
    2.  
    3.  
    4.        char *fgets(char *s, int size, FILE *stream);

    fgets（） 从流中最多读取一个小于size大小的字符，并将它们存储到 S 指向的缓冲区中。 读取在 EOF 或换行符后停止。 如果是新的行被读取，它被存储到缓冲区中。 终止空字节 （'0'） 存储在缓冲区中最后一个字符之后。

    s 代表要保存到的内存空间的首地址，可以是字符数组名，也可以是指向字符数组的字符指针变量名。size 代表的是读取字符串的长度。stream 表示从何种流中读取，可以是标准输入流 stdin，也可以是文件流，即从某个文件中读取。

    可以看到fgets函数与gets函数相似，但fgets函数更为安全，并且可以从文件中读取字符，而gets（）只能从标准输入中获取。fegts还能检查预留存储区的大小，保证字符串不会超出预留空间。gets（） 将一行从 stdin 读取到 s 指向的缓冲区中，直到终止换行符或 EOF，它用空字节 （'0'） 替换它，但并不检查缓冲区是否溢出。

    文件I/O与标准I/O部分：

    写到这里有一个小问题，为什么stdin可以传入FILE*类型参数，stdin是什么?明白的可以自动跳过这里 。

    下面这个链接主要是文件I/O与标准I/O的知识。。。。

    文件I/O与标准I/O



     

    代码：

    简介：下面的代码包括一个封装好的环形队列作为服务器接受客户端发送消息的容器、只需要传入互斥量指针就自动加锁自动解锁的类、封装好的线程类以及客户端服务器主程序。其实代码逻辑很简单，从udp_server.hpp的UdpServer类里面的私有成员变量入手就好。

    服务器启动需要绑定一个端口号，端口号以命令行参数形式传入。

    客户端启动需要在命令行输入服务器ip与端口号

    udp_client.cc

    1. #include 
    2. using namespace std;
    3. #include 
    4. #include 
    5. #include 
    6. #include 
    7. #include 
    8. #include 
    9.  
    10. //sockaddr_in结构体的头文件,当然也包含一些主机转网络序列的函数比如htons
    11. #include 
    12. #include 
    13.  
    14. #include "error.hpp"
    15.  
    16.  
    17.  
    18.  
    19. static void* rfo(void *args)
    20. {
    21.     int sock=*(static_cast<int*>(args));
    22.     while(true)
    23.     {
    24.           //收
    25.         char buffer[4096];
    26.         struct sockaddr_in tmp;//输入型参数；
    27.         socklen_t len=sizeof(tmp);//要初始化，不然没法修改；
    28.         
    29.         //阻塞式接收
    30.         int n=recvfrom(sock,buffer,sizeof(buffer)-1,0,(struct sockaddr*)&tmp,&len);
    31.       
    32.         if(n>0)//接收服务器数据成功
    33.         {
    34.             buffer[n]=0;
    35.             cout<
    36.         }
    37.     }
    38.  
    39.  
    40. }
    41.  
    42. //当传入程序参数个数不对时，调用这个Usage函数告诉他什么是他妈的惊喜！
    43. static void Usage(string proc)
    44. {
    45.     cout<<"Usage:\n\t"<" serverip "<<" serverport\n"<
    46. }
    47.  
    48.  
    49.  
    50.  
    51.  
    52.  
    53.  
    54. // ./udp_client serverip serverport
    55. int main(int argc,char* argv[])
    56. {
    57.     if(argc!=3)
    58.     {
    59.         Usage(argv[0]);
    60.         exit(USAGE_ERR);
    61.     }
    62.  
    63.     //保留输入的服务器的IP地址与端口号
    64.     string serverip=argv[1];
    65.     uint16_t serverport=atoi(argv[2]);
    66.  
    67.     int sock = socket(AF_INET, SOCK_DGRAM, 0);
    68.     if (sock < 0)
    69.     {
    70.         cerr << " create socket error " << strerror(errno) << endl;
    71.         exit(SOCKET_ERR);
    72.     }
    73.  
    74.     //client要不要bind呢？要的！socket通信的本质[clientip,clientport ::serverip,serverport]
    75.     //要不要自己bind呢？不需要自己bind，也不要自己bind，OS自动bind--  客户端的端口号要操作系统随机分配，防止客户端出现启动冲突。
    76.     
    77.     //创建线程去接收；
    78.     pthread_t tid;
    79.     pthread_create(&tid,nullptr,rfo,(void*)&sock);
    80.  
    81.  
    82.     //明确server是谁
    83.     struct sockaddr_in server;
    84.     memset((void*)&server,0,sizeof(server));
    85.     server.sin_family=AF_INET;
    86.     server.sin_port=htons(serverport);//主机序列转网络序列
    87.     server.sin_addr.s_addr=inet_addr(serverip.c_str());//点分十进制字符串ip转成32位无符号整数并转为网络序列，这个函数有两个功能；
    88.  
    89.  
    90.     while(true)
    91.     {
    92.        string message;
    93.        cout<<"please Enter# ";
    94.        getline(cin,message);
    95.  
    96.  
    97.         //在首次调用sendto函数时，操作系统自动给本程序绑定IP地址和端口号，客户端不能自己绑定端口号和ip地址，因为端口号和IP地址会变。
    98.         sendto(sock,message.c_str(),message.size(),0,(const struct sockaddr*)&server,sizeof(server));
    99.     }
    100.  
    101.  
    102.     return 0;
    103. }

    udp_server.cc

    1.  
    2. #include 
    3. #include "udp_server.hpp"
    4.  
    5. #include 
    6. #include 
    7.  
    8. using namespace ns_server;
    9.  
    10. // 上层的业务处理，不关心网络发送，只负责信息处理即可
    11. // 客户端输入命令，服务器执行命令，结果返回给客户端；
    12.  
    13. // 业务1（字符串全部转大写）
    14. string transaction(string request)
    15. {
    16.  
    17.     string result;
    18.     char c;
    19.  
    20.     for (auto &e : request)
    21.     {
    22.         if (islower(e))
    23.         {
    24.             c = toupper(e);
    25.             result.push_back(c);
    26.         }
    27.         else
    28.         {
    29.             result.push_back(e);
    30.         }
    31.     }
    32.  
    33.     return result;
    34. }
    35.  
    36. bool notsecure(string &command)
    37. {
    38.     bool ret = false;
    39.  
    40.     int pos;
    41.  
    42.     pos = command.find("rm");
    43.     if (pos != string::npos)
    44.         ret = true;
    45.  
    46.     pos = command.find("while");
    47.     if (pos != string::npos)
    48.         ret = true;
    49.  
    50.     pos = command.find("mv");
    51.     if (pos != string::npos)
    52.         ret = true;
    53.  
    54.     pos = command.find("kill");
    55.     if (pos != string::npos)
    56.         ret = true;
    57.  
    58.     return ret;
    59. }
    60. // 业务二（服务器端获取命令字符串，服务器执行完成后给客户端返回结果）
    61. string excuteCommand(string command)
    62. {
    63.     // 1.安全检查
    64.     if (notsecure(command))
    65.         return "Sorry,you can do that!";
    66.  
    67.     // 2.业务逻辑处理
    68.     FILE *fp = popen(command.c_str(), "r");//popen函数是创建管道在创建子进程，利用子进程来处理命令，并把结果输出到一个文件的，返回值是文件指针。
    69.     if (fp == nullptr)
    70.         return "None";
    71.  
    72.     // 3.获取结果
    73.     char line[1024];
    74.     string result;
    75.  
    76.     // 这里用while的原因是fgets函数遇到换行符或EOF读取结束,也就是说一次读一行，使用while循环读到文件结尾；
    77.     while (fgets(line, sizeof(line), fp)!=nullptr)
    78.     {
    79.         result += line;
    80.     }
    81.  
    82.     pclose(fp);
    83.     return result;
    84. }
    85.  
    86. // 当传入程序参数个数不对时，调用这个Usage函数告诉他什么是他妈的惊喜！
    87. static void Usage(string proc)
    88. {
    89.     cout << "Usage:\n\t" << proc << " port\n"
    90.          << endl;
    91. }
    92.  
    93. // ./udp_server     serverport（服务器自己设置端口号）
    94. int main(int argc, char *argv[])
    95. {
    96.     if (argc != 2) // 命令行传入参数不够
    97.     {
    98.         Usage(argv[0]);
    99.         exit(USAGE_ERR);
    100.     }
    101.     // 把字符串port转换成16位整数
    102.     uint16_t port = atoi(argv[1]);
    103.  
    104.     // 智能指针构造UdpServer对象，构造函数需要传入自己想定义的port
    105.     //unique_ptr usvr(new UdpServer(excuteCommand, port));
    106.  
    107.     unique_ptr usvr(new UdpServer(port));
    108.     //usvr->InitServer();  // 服务器初始化
    109.     usvr->StartServer(); // 服务器开始服务
    110.  
    111.     return 0;
    112. }

    udp_server.hpp

    1. #pragma once
    2. #include 
    3. #include  /* See NOTES */
    4. #include 
    5. #include 
    6. #include 
    7. #include 
    8. #include 
    9. #include 
    10. #include 
    11. #include 
    12. // sockaddr_in结构体的头文件
    13. #include 
    14. #include 
    15. #include "error.hpp"
    16. using namespace std;
    17. #include "RingQueue.hpp"
    18. #include "Thread.hpp"
    19. #include "lockGuard.hpp"
    20.  
    21. namespace ns_server
    22. {
    23.     // const uint16_t default_port = 8081;
    24.  
    25.     using func_t = function<string(string)>; // func_t是指代返回值为string，参数为string的函数指针；
    26.     class UdpServer
    27.     {
    28.  
    29.     public:
    30.         // // 构造服务器对象必须绑定端口号，指定服务器处理方法
    31.         // UdpServer(func_t cb, uint16_t port = default_port)
    32.         //     : _port(port), _service(cb)
    33.  
    34.         // {
    35.  
    36.         //     cout << " Server Port : " << _port << endl;
    37.         // }
    38.  
    39.         // 构造服务器对象必须绑定端口号，指定服务器处理方法
    40.         UdpServer(uint16_t port)
    41.             : _port(port), _p()
    42.  
    43.         {
    44.             cout << " Server Port : " << _port << endl;
    45.  
    46.             pthread_mutex_init(&_mutex, nullptr); // 初始化锁；
    47.             // 这里使用c++11 bind函数，相当于函数适配器，构建了一个可调用对象，函数参数顺序也可以占位符标定，_1,_2类似这样；
    48.             _p = new Thread(1, bind(&UdpServer::Recv, this));
    49.             _c = new Thread(1, bind(&UdpServer::Broadcast, this));
    50.         }
    51.  
    52.         void StartServer()
    53.         {
    54.  
    55.             // 1.创建socket接口，打开网络文件；
    56.             _socket = socket(AF_INET, SOCK_DGRAM, 0);
    57.             if (_socket < 0)
    58.             {
    59.                 cerr << " create socket error: " << strerror(errno) << endl;
    60.                 exit(SOCKET_ERR);
    61.             }
    62.  
    63.             cout << " create socket success: " << _socket << endl; // 3
    64.  
    65.             // 2.给服务器绑定本地IP和端口号（要知道是哪个IP哪个端口号接收数据）
    66.             struct sockaddr_in local;
    67.             bzero(&local, sizeof(local)); // 清零
    68.  
    69.             local.sin_family = AF_INET;
    70.             local.sin_port = htons(_port);             // 端口号
    71.             local.sin_addr.s_addr = htonl(INADDR_ANY); // IP地址
    72.  
    73.             if (bind(_socket, (const struct sockaddr *)&local, sizeof(local)) < 0) // 绑定本地Ip与端口号
    74.             {
    75.                 cerr << " bind socket error: " << strerror(errno) << endl;
    76.                 exit(BIND_ERR);
    77.             }
    78.             cout << " bind socket success: " << _socket << endl;
    79.  
    80.             _p->run();
    81.             _c->run();
    82.         }
    83.  
    84.         void addUser(const string &name, const struct sockaddr_in &peer)
    85.         {
    86.             lockGuard lock(&_mutex);
    87.  
    88.             auto it = _onlineUser.find(name);
    89.             if (it != _onlineUser.end())
    90.                 return;
    91.  
    92.             // 没有就插入
    93.             _onlineUser.insert(pairstruct sockaddr_in>(name, peer));
    94.         }
    95.  
    96.         // 接收client数据并记录用户ip和端口
    97.         void Recv()
    98.         {
    99.             char buffer[1024];
    100.             while (true)
    101.             {
    102.                 // 收
    103.                 struct sockaddr_in peer; // 输入性参数，获得客户端ip与端口号
    104.                 socklen_t len = sizeof(peer);
    105.                 int n = recvfrom(_socket, buffer, sizeof(buffer) - 1, 0, (struct sockaddr *)&peer, &len); // 接收客户端发送过来的消息和客户端ip与端口
    106.  
    107.                 if (n > 0)
    108.                     buffer[n] = '\0';
    109.                 else
    110.                     continue;
    111.  
    112.                 // 提取client信息---debug；
    113.                 string client_ip = inet_ntoa(peer.sin_addr);
    114.                 uint16_t client_port = ntohs(peer.sin_port);                        // 网络序列转为主机序列
    115.                 cout << client_ip << "-" << client_port << " # " << buffer << endl; // 显示客户端发来的数据
    116.  
    117.                 // 利用ip和端口构建一个用户名
    118.                 string name = client_ip;
    119.                 name += "-";
    120.                 name += to_string(client_port);
    121.  
    122.                 
    123.                 addUser(name, peer); // 存入用户ip和端口，后面把消息转发给所有用户
    124.                 
    125.                 string message=name;
    126.                 message+=">>";
    127.                 message+=buffer;
    128.                 _rq.push(message);    // 接收到的消息加工一下存入环形队列；
    129.  
    130.                 // 业务处理
    131.                 // string message = _service(buffer);
    132.                 // 发
    133.                 // sendto(_socket, message.c_str(), message.size(), 0, (struct sockaddr *)&peer, sizeof(peer));
    134.             }
    135.         }
    136.  
    137.         // 广播
    138.         void Broadcast()
    139.         {
    140.             while (true)
    141.             {
    142.                 string message;
    143.  
    144.                 // 因为封装好的环形队列里面有信号量和互斥量，所以这里不必担心线程安全问题；
    145.                 _rq.pop(&message);
    146.  
    147.                 vector<struct sockaddr_in> v;
    148.                 // 这里需要设置互斥量，因为两个线程访问了临界资源，结果具有不确定性；
    149.                 {
    150.                     lockGuard lock(&_mutex);
    151.                     for (auto &user : _onlineUser)
    152.                     {
    153.                         v.push_back(user.second);
    154.                     }
    155.                 }
    156.  
    157.                 for (auto &e : v) // 给所有用户发消息；
    158.                 {
    159.                     sendto(_socket, message.c_str(), message.size(), 0, (const struct sockaddr *)&e, sizeof(e));
    160.                     // 测试消息发送出去了没
    161.                     cout << "send done..." << message << endl;
    162.                 }
    163.             }
    164.         }
    165.  
    166.         ~UdpServer()
    167.         {
    168.             pthread_mutex_destroy(&_mutex);
    169.  
    170.             // 等待线程结束
    171.             _p->join();
    172.             _c->join();
    173.  
    174.             // 回收堆空间；
    175.             delete _p;
    176.             delete _c;
    177.         }
    178.  
    179.     private:
    180.         int _socket;
    181.         uint16_t _port;
    182.         // func_t _service; // 上一个版本只是简单的IO，现在要进行业务处理；
    183.         unordered_mapstruct sockaddr_in> _onlineUser; // 把所有用户ip和端口保存起来，后面要给所有人转发消息；
    184.         RingQueue _rq;                                 // 环形队列存放用户发的消息；
    185.         pthread_mutex_t _mutex;
    186.  
    187.         // 两个线程，一个收，一个发；
    188.         Thread *_p;
    189.         Thread *_c;
    190.     };
    191. }
    1. #pragma once
    2.        enum
    3.         {   
    4.             USAGE_ERR=1,
    5.             SOCKET_ERR,
    6.             BIND_ERR
    7.  
    8.         };

    RingQueue.hpp

    1. #pragma once
    2.  
    3. #include
    4. #include
    5.  
    6. #include
    7. #include
    8. #include
    9.  
    10.  
    11.  
    12. using namespace std;
    13.  
    14. const int N=50;
    15.  
    16. template<class T>
    17. class RingQueue
    18. {
    19.  
    20.     void P(sem_t* sem)
    21.     {
    22.         sem_wait(sem);
    23.     }
    24.  
    25.     void V(sem_t* sem)
    26.     {
    27.         sem_post(sem);
    28.     }
    29.  
    30.     void Lock(pthread_mutex_t& mutex)
    31.     {
    32.         pthread_mutex_lock(&mutex);
    33.     }
    34.      void UnLock(pthread_mutex_t& mutex)
    35.     {
    36.         pthread_mutex_unlock(&mutex);
    37.     }
    38. public:
    39.     RingQueue(int num=N):_ring(num),_cup(num),_consumer_step(0),_productor_step(0)
    40.     {
    41.        sem_init(&_data_sem,0,0);
    42.        sem_init(&_space_sem,0,_cup);
    43.        pthread_mutex_init(&_c_mutex,nullptr);
    44.        pthread_mutex_init(&_p_mutex,nullptr);
    45.  
    46.     }
    47.  
    48.     void push(const T& in)
    49.     {
    50.         P(&_space_sem);
    51.         Lock(_p_mutex);
    52.         _ring[_productor_step++]=in;
    53.         _productor_step %= _cup;
    54.         //消费者信号量加一；（数据）
    55.         V(&_data_sem);
    56.         UnLock(_p_mutex);
    57.     }
    58.  
    59.     void pop(T* out)
    60.     {
    61.         P(&_data_sem);
    62.         Lock(_c_mutex);
    63.         *out=_ring[_consumer_step++];
    64.         _consumer_step %= _cup;
    65.         //生产者信号量加一（空间）
    66.         V(&_space_sem);
    67.         UnLock(_c_mutex);
    68.     }
    69.  
    70.     ~RingQueue()
    71.     {
    72.         sem_destroy(&_data_sem);
    73.         sem_destroy(&_space_sem);
    74.  
    75.         pthread_mutex_destroy(&_c_mutex);
    76.         pthread_mutex_destroy(&_p_mutex);
    77.     }
    78.  
    79. private:
    80.  
    81.     vector _ring;//数组模拟环形队列
    82.     int  _cup;//容量
    83.  
    84.     sem_t _data_sem;//消费者信号量
    85.     sem_t _space_sem;//生产者信号量
    86.  
    87.     int _consumer_step;//消费者下标
    88.     int _productor_step;//生产者下标
    89.  
    90.  
    91.     // 单生产和单消费不存在竞争问题，只要有信号量即可；但是多生产和多消费的线程，可能都申请到了信号量，但是都在竞争同一块资源，无法保证原子性；
    92.     pthread_mutex_t _c_mutex;
    93.     pthread_mutex_t _p_mutex;
    94.  
    95. };

    lockGuard.hpp

    1. #pragma once
    2.  
    3. #include 
    4. #include 
    5.  
    6. class Mutex//成员：加锁函数和解锁函数
    7. {
    8. public:
    9.     Mutex(pthread_mutex_t* pmutex):_pmutex(pmutex)   {}
    10.     
    11.     void lock()
    12.     {
    13.         pthread_mutex_lock(_pmutex);
    14.     }
    15.    
    16.        void unlock()
    17.     {
    18.         pthread_mutex_unlock(_pmutex);
    19.     }
    20.    
    21.  
    22.     ~Mutex(){}
    23.  
    24. private:
    25.     pthread_mutex_t* _pmutex;//需要传入一个互斥量（锁）的指针；
    26. };
    27.  
    28.  
    29. //对Mutex进行二次封装；
    30. //创建该对象时自动加锁，析构时自动解锁；
    31. class lockGuard
    32. {   
    33. public:
    34.     lockGuard(pthread_mutex_t* pmutex):_mutex(pmutex)//利用锁的指针构建Mutex对象
    35.     {
    36.         _mutex.lock();
    37.     }
    38.  
    39.    ~lockGuard()
    40.     {
    41.         _mutex.unlock();
    42.     }
    43.  
    44. private:
    45.     Mutex _mutex;//类内创建对象
    46. };

    Thread.hpp

    1. #pragma once
    2.  
    3. #include 
    4. #include 
    5. #include 
    6. #include 
    7. #include 
    8.  
    9.  
    10.  
    11. class Thread
    12. {
    13. public:
    14.     //typedef void (*func_t) (void*);
    15.     using func_t=function<void()>;//fun_t：无返回值，无参数的函数指针；
    16.     typedef enum
    17.     {
    18.         NEW=0,
    19.         RUNNING,
    20.         EXITED
    21.     }ThreadStatus;
    22.  
    23.  
    24.  
    25. public:
    26.  
    27.    
    28.         Thread(int num,func_t func)
    29.         :_tid(0),
    30.         _status(NEW),
    31.         _func(func)
    32.        
    33.         {
    34.             char name[128];
    35.             snprintf(name,sizeof(name),"thread-%d",num);
    36.             _name=name;
    37.         }
    38.  
    39.         //状态：new，running，exited
    40.         int status()
    41.         {
    42.             return _status;
    43.         }
    44.         //线程名
    45.         std::string threadname()
    46.         {
    47.             return _name;
    48.         }
    49.         //线程ID（共享库中的进程地址空间的虚拟地址）
    50.         pthread_t threadid()
    51.         {
    52.             if(_status==RUNNING)//线程已经被创建，线程id已经输入到成员变量_tid中；
    53.                 return _tid;
    54.             else 
    55.             {
    56.                 std::cout<<"thread is not running,no tid!"<
    57.                 return 0;
    58.             }
    59.         }
    60.  
    61.         static void* runHelper(void *args)
    62.         {
    63.             //静态成员函数不能访问类内所有成员，因为没有this指针；
    64.             Thread* td=(Thread*)args;
    65.             (*td)();//该对象调用仿函数；
    66.             return nullptr; 
    67.         }
    68.  
    69.         void operator()()//仿函数
    70.         {
    71.             _func();
    72.         }
    73.  
    74.         //创建线程
    75.         void run()
    76.         {
    77.             //因为runHelper函数必须只能有一个void*参数，所以runHelper函数在类内必须定义为static，这样才没有this指针；
    78.             int n=pthread_create(&_tid,nullptr,runHelper,this);
    79.             if(n!=0) return exit(0);//线程创建失败，那么直接退出进程；
    80.             _status=RUNNING;
    81.         }
    82.  
    83.         //等待线程结束
    84.         void join()
    85.         {
    86.             int n=pthread_join(_tid,nullptr);
    87.             if(n!=0) 
    88.             {
    89.                 std::cerr<<"main thread join thread "<<_name<<" error "<
    90.                 return;
    91.             }
    92.             _status=EXITED;//线程退出；
    93.         }
    94. private:
    95.     pthread_t _tid;//线程ID（原生线程库中为该线程所创建的TCB起始虚拟地址）
    96.     std::string _name;//线程名
    97.     func_t _func;//线程要执行的回调
    98.     //void* _args;//线程回调函数参数
    99.     ThreadStatus _status;//枚举类型：状态
    100.  
    101. };
    102.

    makefile

    1. .PHONY:all
    2. all:udp_server udp_client
    3.  
    4. udp_server:udp_server.cc
    5. 	g++ $^ -o $@ -std=c++11 -lpthread
    6. udp_client:udp_client.cc
    7. 	g++ $^ -o $@ -std=c++11 -lpthread
    8.  
    9. .PHONY:clean
    10. clean:
    11. 	rm -f udp_client udp_server

  • 相关阅读:
    springweb层控制类的使用
    Couchdb-权限绕过--命令执行--(CVE-2017-12635)&&(CVE-2017-12636)--H2database命令执行--(CVE-2022-23221)
    Android 基础知识4-2.1常用控件列表（ListView）
    达梦数据库DM8开启归档日志
    Python-爬虫（正则表达式基础、修饰符、元字符、数量修饰符，练习判断身份证是否正确）
    CentOS7---Nginx安装并配置虚拟主机
    高纯度高活性艾美捷人重组MEGACD40L蛋白（可溶性）
    java 多线程（）—— 死锁
    Python 全栈安全（一）
    2.4 图解CIO工作指南(IT 架构模型) --- 构成IT 架构的技术要素
  • 原文地址：https://blog.csdn.net/zzxz8/article/details/132743358