• TCP/IP网络编程(9) 进程间通信

    1. IPC基本概念

    进程间通信(Inter process communication, IPC)以为这两个不同进程间可以交换数据，为了完成这一点，操作系统中应该提供两个进程可以同时访问的内存空间。因此，只要有两个进程可以同时访问的内存空间，就可以通过此空间来交换数据。但是，进程具有完全独立的内存结构，即使通过fork()函数创建的子进程，也不会与父进程共享内存，因此进程间通信需要通过其他的特殊方法完成。

    1.1 利用管道实现进程间通信

    为完成进程之间的通信，需要创建管道，管道不属于进程的资源(即不是fork复制的对象)，而是和套接字一样，属于操作系统资源，因此通过管道首实现IPC的原理是，两个进程通过操作系统提供的内存进行通信。
     

    创建管道的方式：

    1. #include 
    2.  
    3. /*
    4.    成功时返回0
    5.    失败时返回-1
    7.    参数：
    8.    fields[0]  通过管道接收数据时使用的文件描述符
    9.    fields[1]  通过管道传输数据时使用的文件描述符
    10. */
    11. int pipe(int fields[2]);

    父进程调用pipe函数时将创建管道，同时获取对应于管道出入口的文件描述符，此时父进程可以读写同一管道，在调用fork创建子进程之后，子进程也将同时拥有管道出入口的文件描述符：

    1. #include 
    2. #include 
    3.  
    4. #define BUFF_SIZE 30
    5.  
    6. int main(int argc, char** argv)
    7. {
    8.     int fds[2];     // 用于存储管道出入口的文件描述符
    9.  
    10.     char message[] = "Hello Process\n";
    11.  
    12.     char buffer[BUFF_SIZE];
    13.  
    14.     if (pipe(fds) == -1)
    15.     {
    16.         printf("Falied to create pipe.\n");
    17.         return -1;
    18.     }
    19.  
    20.     pid_t pid = fork();      // 创建进行
    21.  
    22.     if (pid == 0)
    23.     { 
    24.         // 子进程   写入管道
    25.         write(fds[1], message, sizeof(message));
    26.     }
    27.     else
    28.     { 
    29.         // 父进程   读取管道
    30.         read(fds[0], buffer, BUFF_SIZE);
    31.         fputs(buffer, stdout);
    32.  
    33.     }
    34.     return 0;
    35. }

    进程间通信的示意图图下所示：

    1.2 通过管道实现进程间双向通信

    方案1：通过一个管道实现两个进程间的双向通信

    代码实例：

    1. #include 
    2. #include 
    3. #include 
    4.  
    5. #define BUFF_SIZE 30
    6.  
    7. int main(int argc, char** argv)
    8. {
    9.     int fds[2];     // 用于存储管道出入口的文件描述符
    10.  
    11.     char message1[] = "Hello Process\n";
    12.     char message2[] = "I am vac!\n";
    13.  
    14.     char buffer[BUFF_SIZE];
    15.     memset(buffer, 0, BUFF_SIZE);
    16.  
    17.     if (pipe(fds) == -1)
    18.     {
    19.         printf("Falied to create pipe.\n");
    20.         return -1;
    21.     }
    22.  
    23.     pid_t pid = fork();      // 创建进行
    24.  
    25.     if (pid == 0)
    26.     { 
    27.         // 子进程   写入管道
    28.         write(fds[1], message1, sizeof(message1));
    29.         sleep(2);
    30.         read(fds[0], buffer, BUFF_SIZE);
    31.         printf("Child proc received data: %s\n", buffer);
    32.         memset(buffer, 0, BUFF_SIZE);
    33.     }
    34.     else
    35.     { 
    36.         // 父进程   读取管道
    37.         read(fds[0], buffer, BUFF_SIZE);
    38.         printf("Parent proc received data: %s\n", buffer);
    39.         write(fds[1], message2, sizeof(message2));
    40.         sleep(3);
    41.         memset(buffer, 0, BUFF_SIZE);
    42.     }
    43.     return 0;
    44. }

    运行结果：

     如果将子进程中的sleep方法注释掉，再运行代码：

    1.     if (pid == 0)
    2.     { 
    3.         // 子进程   写入管道
    4.         write(fds[1], message1, sizeof(message1));
    5.         //sleep(2);
    6.         read(fds[0], buffer, BUFF_SIZE);
    7.         printf("Child proc received data: %s\n", buffer);
    8.         memset(buffer, 0, BUFF_SIZE);
    9.     }

    结果如下所示：                            

     此时父进程无法再收到数据，因为子进程在写入数据后，未经过延时，立即又读取数据，因此管道中写入的数据被子进程提前取走。按照管道数据的读取机制，数据在写入管道之后，称为无主数据，哪个进程先通过read函数读取数据，哪个进程就先得到数据。结果导致父进程通过read函数再也读取不到数据，而将无限期等待管道数据。

    注：利用一个管道进行进程间数据双向通信，程序设计者需要提前预测并控制运行流程，程序实现难度很大。

    方案2：通过两个管道实现进程间相互通信

    通过创建两个管道，各自在进程间负责不同的数据流，即可方便的实现进程间双向通信，采用两个管道即可避免程序运行流程的预测或控制。

     

     代码实例：

    1. #include 
    2. #include 
    3. #include 
    4.  
    5. #define BUFF_SIZE 30
    6.  
    7. int main(int argc, char** argv)
    8. {
    9.     int fds1[2];     // 子进程向父进程写数据
    10.     int fds2[2];     // 父进程向子进程写数据
    11.  
    12.     char message1[] = "Hello Process\n";
    13.     char message2[] = "I am vac!\n";
    14.  
    15.     char buffer[BUFF_SIZE];
    16.     memset(buffer, 0, BUFF_SIZE);
    17.  
    18.     if (pipe(fds1) == -1)
    19.     {
    20.         printf("Falied to create pipe.\n");
    21.         return -1;
    22.     }
    23.  
    24.     if (pipe(fds2) == -1)
    25.     {
    26.         printf("Falied to create pipe.\n");
    27.         return -1;
    28.     }
    29.  
    30.     pid_t pid = fork();      // 创建进行
    31.  
    32.     if (pid == 0)
    33.     { 
    34.         // 子进程   写入管道
    35.         write(fds1[1], message1, sizeof(message1));
    36.         read(fds2[0], buffer, BUFF_SIZE);
    37.         printf("Child proc received data: %s\n", buffer);
    38.         memset(buffer, 0, BUFF_SIZE);
    39.     }
    40.     else
    41.     { 
    42.         // 父进程   读取管道
    43.         sleep(1);     // 可以延时更短，稍微等待一下
    44.         read(fds1[0], buffer, BUFF_SIZE);
    45.         printf("Parent proc received data: %s\n", buffer);
    46.         write(fds2[1], message2, sizeof(message2));
    47.         memset(buffer, 0, BUFF_SIZE);
    48.     }
    49.     return 0;
    50. }

    运行结果：

     2. 进程间通信的应用

          保存消息的回声服务器: 在向客户端提供服务的同时，服务端能够将接收到的数据同步保存到文件中。

    客户端代码可复用《TCP/IP网络编程(8) 基于Linux的多进程服务器》中的客户端代码

    1. /* 
    2.     客户端
    3.     create_date: 2022-7-29
    4. */
    5.  
    6. #include 
    7. #include 
    8. #include 
    9. #include 
    10. #include 
    11. #include 
    12.  
    13. #define BUFF_SIZE  30
    14. #define ADDRESS    "127.0.0.1"
    15. #define PORT       13100
    16.  
    17. void readRoutine(int sock, char* buf);
    18. void writeRoutine(int sock, char* buf);
    19.  
    20. int main(int argc, char** argv)
    21. {
    22.     char buffer[BUFF_SIZE];
    23.  
    24.     struct sockaddr_in serverAddr;               // 服务端地址
    25.     memset(&serverAddr, 0, sizeof(serverAddr));   
    26.     serverAddr.sin_family = AF_INET;
    27.     serverAddr.sin_addr.s_addr = inet_addr(ADDRESS);
    28.     serverAddr.sin_port = htons(PORT);
    29.  
    30.     memset(buffer, 0, BUFF_SIZE);
    31.  
    32.     int socket = ::socket(PF_INET, SOCK_STREAM, 0);
    33.  
    34.     if (socket == -1)
    35.     {
    36.         printf("Failed to init socket.\n");
    37.         return -1;
    38.     }
    39.  
    40.     if (connect(socket, (struct sockaddr*)&serverAddr, sizeof(serverAddr)) == -1)
    41.     {
    42.         printf("Failed to connect to server.\n");
    43.         return -2;
    44.     }
    45.  
    46.     printf("Successfully connect to the server.\n");
    47.  
    48.     pid_t pid = fork();
    49.  
    50.     if (pid == 0)
    51.     {
    52.         // 子进程负责发送
    53.         writeRoutine(socket, buffer);
    54.     }
    55.     else
    56.     {
    57.         // 父进程负责接收
    58.         readRoutine(socket, buffer);
    59.     }
    60.  
    61.     close(socket);
    62.  
    63.     return 0;
    64.     
    65. }
    66.  
    67.  
    68. void readRoutine(int sock, char* buf)
    69. {
    70.     while (true)
    71.     {
    72.         memset(buf, 0, BUFF_SIZE);
    73.  
    74.         int str_len = read(sock, buf, BUFF_SIZE);
    75.  
    76.         if (str_len == 0)    // EOF
    77.         {
    78.             printf("Receive EOF\n");
    79.             return;
    80.         }
    81.  
    82.         printf("\nReceive data from server: %s\n", buf);
    83.     }
    84.     
    85. }
    86.  
    87. void writeRoutine(int sock, char* buf)
    88. {
    89.     while (true)
    90.     {
    91.         fputs("Input message(Type Q(q) to quit): ", stdout);
    92.  
    93.         fgets(buf, BUFF_SIZE, stdin);
    94.  
    95.         if (strncmp(buf, "Q\n", 2) == 0 || strncmp(buf, "q\n", 2) == 0)
    96.         {
    97.             shutdown(sock, SHUT_WR);
    98.             return;
    99.         }
    100.             
    101.         write(sock, buf, strlen(buf));
    102.     }
    103.     
    104. }

    服务端示例代码代码示例：

    1. #include 
    2. #include 
    3. #include 
    4. #include 
    5. #include 
    6. #include 
    7. #include 
    8. #include 
    9.  
    10. void readChildProcess(int signo)
    11. {
    12.     int status;
    13.     pid_t pid = waitpid(-1, &status, WNOHANG);   // 非阻塞
    14.  
    15.     if (WIFEXITED(status))
    16.     {
    17.         printf("Removed child process %d\n", WEXITSTATUS(status));
    18.     }
    19. }
    20.  
    21. #define BUFF_SIZE 30
    22. #define PORT  13100
    23.  
    24. int main(int argc, char** argv)
    25. {
    26.     int serverSocket;
    27.     int clientSocket;
    28.  
    29.     struct sockaddr_in servAddr;
    30.     struct sockaddr_in clientAddr;
    31.  
    32.     char buffer[BUFF_SIZE];
    33.     memset(buffer, 0, BUFF_SIZE);
    34.  
    35.     socklen_t addrSize; 
    36.  
    37.     int fds[2];     // 管道的文件描述符
    38.  
    39.     struct sigaction sigact;
    40.     sigact.sa_handler = readChildProcess;
    41.     sigemptyset(&sigact.sa_mask);
    42.     sigact.sa_flags = 0;
    43.     sigaction(SIGCHLD, &sigact, 0);      // 注册信号
    44.  
    45.     serverSocket = socket(PF_INET, SOCK_STREAM, 0);
    46.     if (serverSocket == -1)
    47.     {
    48.         printf("Failed to init server socket.\n");
    49.         return -1;
    50.     }
    51.  
    52.     memset(&servAddr, 0, sizeof(servAddr));
    53.     servAddr.sin_family = AF_INET;
    54.     servAddr.sin_addr.s_addr = htonl(INADDR_ANY);
    55.     servAddr.sin_port = htons(PORT);
    56.  
    57.     if (bind(serverSocket, (struct sockaddr*)&servAddr, sizeof(servAddr)) == -1)
    58.     {
    59.         printf("Failed to bind the server socket.\n");
    60.         return -1;
    61.     }
    62.  
    63.     if (listen(serverSocket, 5) == -1)
    64.     {
    65.         printf("Failed to listen client.\n");
    66.         return -1;
    67.     }
    68.  
    69.     // 创建管道
    70.     pipe(fds);
    71.  
    72.     // 创建写文件进程
    73.     pid_t pid = fork();
    74.  
    75.     if (pid == 0)
    76.     {
    77.         // 写文件子进程
    78.         FILE* fp = fopen("data.txt", "a");
    79.  
    80.         char msg[100];
    81.  
    82.         while (true)
    83.         {
    84.             /* code */
    85.             usleep(1000*40);
    86.  
    87.             memset(msg, 0, 100);
    88.  
    89.             int len = read(fds[0], msg, 100);
    90.  
    91.             // 解析读取到的内容
    92.             if (len <= 5)
    93.             {
    94.                 break;
    95.             }
    96.  
    97.             char magicChar1 = msg[0];
    98.             char magicChar2 = msg[1];
    99.  
    100.             if (magicChar1 != 'M' || magicChar2 != 'F')
    101.             {
    102.                 break;
    103.             }
    104.  
    105.             if (msg[2] == 1)
    106.             {
    107.                 break;
    108.             }
    109.  
    110.             ushort dataLen = msg[3] | (msg[4] << 8);
    111.  
    112.             if (dataLen <= 0)
    113.             {
    114.                 continue;
    115.             }
    116.  
    117.             // 将解析到的内容写入文件进行保存
    118.             fwrite(&msg[5], dataLen, 1, fp);
    119.         }
    120.  
    121.         fclose(fp);
    122.         return 0;
    123.     }
    124.     else
    125.     {
    126.         // 父进程
    127.         while (true)
    128.         {
    129.             printf("Waiting connect from client...\n");
    130.  
    131.             addrSize = sizeof(clientAddr);
    132.             
    133.             clientSocket = accept(serverSocket, (sockaddr*)&clientAddr, &addrSize);
    134.  
    135.             if (clientSocket == -1)
    136.                 continue;
    137.  
    138.             printf("Accept new connection from %s:%d.\n", inet_ntoa(clientAddr.sin_addr), ntohs(clientAddr.sin_port));
    139.  
    140.             pid_t processId = fork();
    141.  
    142.             if (processId == 0)
    143.             {
    144.                 // 子进程
    145.                 close(serverSocket);
    146.                 // 接受消息
    147.  
    148.                 int recvLen = 0;
    149.  
    150.                 // 构造报文
    151.                 char message[100];
    152.  
    153.                 usleep(1000*100);
    154.             
    155.                 while ((recvLen = read(clientSocket, buffer, BUFF_SIZE)) != 0)
    156.                 {
    157.                     /* code */
    158.                     write(clientSocket, buffer, recvLen);
    159.  
    160.                     // 向管道写数据
    161.                     memset(message, 0, 100);
    162.  
    163.                     message[0] = 'M';
    164.                     message[1] = 'F';
    165.                     message[2] = 0;
    166.                     message[3] = recvLen & 0xFF;
    167.                     message[4] = (recvLen >> 8) & 0xFF;
    168.  
    169.                     memcpy(&message[5], buffer, recvLen);       // 注意：memcpy里面的指针不要转换成void*
    170.                     
    171.                     
    172.                     write(fds[1], message, recvLen + 5);
    173.                      
    174.  
    175.                     memset(buffer, 0, BUFF_SIZE);
    176.                 } 
    177.  
    178.                 close(clientSocket);
    179.  
    180.                 
    181.                 memset(message, 0, 100);
    182.  
    183.                 // 发送结束标志
    184.                 message[0] = 'M';
    185.                 message[1] = 'F';
    186.                 message[2] = 1;
    187.                 message[3] = 0;
    188.                 message[4] = 0;
    189.  
    190.                 write(fds[1], (void*)message, 5);
    191.                 
    192.  
    193.                 printf("Disconnecting from server...\n");
    194.  
    195.                 sleep(2);
    196.  
    197.                 return 0;
    198.                 
    199.             }            
    200.             else if (processId == -1)
    201.             {
    202.                 close(clientSocket);
    203.                 continue;
    204.             }
    205.             else
    206.             {
    207.                 // 主进程
    208.                 printf("Create new process %d for client.\n", processId);
    209.  
    210.                 close(clientSocket);
    211.  
    212.                 memset(&clientAddr, 0, sizeof(clientAddr));
    213.  
    214.                 continue;
    215.             }
    216.         }
    217.         
    218.     }
    219.  
    220.     close(serverSocket);
    221.  
    222.     return 0;
    223. }
    224.  
    225.  
    226. /*
    228. 服务器进程和文件进程之间设计通信格式
    230. 0    M          // 报文头
    231. 1    F          // 报文头
    232. 2    0/1        // 报文类型  0数据  / 1 结束
    233. 3    len        // 数据长度  低位
    234. 4    len        // 数据长度  高位
    235. 5    x          // 数据           
    236. 6    x          // 数据
    237. 7    x          // 数据
    238. .               // .
    239. .
    240. .
    242. */

    客户端运行结果：

     服务端运行结果：

     通过管道进行进程间通信，将服务端收到的消息发送给写文件进程，将收到的消息保存至文件：

     -----------------------------------------//end//---------------------------------------------

  • 相关阅读:
    得物Flink内核探索实践
    一文了解io包中的discard类型
    MCE虚拟筛选化合物库
    （附源码）ssm考试题库管理系统 毕业设计 069043
    ES6 入门教程 22 Class 的基本语法 22.9 静态属性 & 22.10 私有方法和私有属性
    Xilinx IP 10 Gigabit Ethernet Subsystem IP
    物联网的挑战
    nginx的优先级和匹配方式
    Linux Vim编辑器的基本使用
    deployment html---＞JDBC---＞mysql
  • 原文地址：https://blog.csdn.net/zj1131190425/article/details/126082415