【Linux】进程间通信

用户11290673

发布于 2025-08-19 11:26:56

2170

文章被收录于专栏：学习学习

1.进程间通信介绍

1.1进程间通信⽬的

数据传输：⼀个进程需要将它的数据发送给另⼀个进程

资源共享：多个进程之间共享同样的资源。

通知事件：⼀个进程需要向另⼀个或⼀组进程发送消息，通知它（它们）发⽣了某种事件（如进

程终⽌时要通知⽗进程）。

进程控制：有些进程希望完全控制另⼀个进程的执⾏（如Debug进程），此时控制进程希望能够

拦截另⼀个进程的所有陷⼊和异常，并能够及时知道它的状态改变。

1.2 进程间通信发展

管道

System V进程间通信

POSIX进程间通信

1.3 进程间通信分类

管道匿名管道pipe 命名管道
System V IPC System V 消息队列 System V 共享内存 System V 信号量
POSIX IPC 消息队列共享内存信号量互斥量条件变量读写锁

2. 管道

什么是管道

管道是Unix中最古⽼的进程间通信的形式。

我们把从⼀个进程连接到另⼀个进程的⼀个数据流称为⼀个“管道”

3.匿名管道

# include <unistd.h> 功能 : 创建⼀⽆名管道原型 int pipe ( int fd[ 2 ]); 参数 fd ：⽂件描述符数组 , 其中 fd[ 0 ] 表⽰读端 , fd[ 1 ] 表⽰写端返回值 : 成功返回 0 ，失败返回错误代码

3.1实例代码

# include <stdio.h> # include <stdlib.h> # include <string.h> # include <unistd.h> int main ( void ) { int fds[ 2 ]; char buf[ 100 ]; int len; if ( pipe(fds) == -1 ) perror( "make pipe" ), exit ( 1 ); // read from stdin while ( fgets(buf, 100 , stdin ) ) { len = strlen (buf); // write into pipe if ( write(fds[ 1 ], buf, len) != len ) { perror( "write to pipe" ); break ; } memset (buf, 0x00 , sizeof (buf)); // read from pipe if ( (len=read(fds[ 0 ], buf, 100 )) == -1 ) { perror( "read from pipe" ); break ; } // write to stdout if ( write( 1 , buf, len) != len ) { perror( "write to stdout" ); break ; } } }

3.2⽤ fork 来共享管道原理

3.3 站在⽂件描述符⻆度-深度理解管道

3.4站在内核⻆度-管道本质

所以，看待管道，就如同看待⽂件⼀样！管道的使⽤和⽂件⼀致，迎合了“Linux⼀切皆⽂件思

想”。

3.5管道样例

测试管道读写

# include <unistd.h> # include <stdlib.h> # include <stdio.h> # include <errno.h> # include <string.h> # define ERR_EXIT(m) \ do \ { \ perror(m); \ exit(EXIT_FAILURE); \ } while(0) int main ( int argc, char *argv[]) { int pipefd[ 2 ]; if (pipe(pipefd) == -1 ) ERR_EXIT( "pipe error" ); pid_t pid; pid = fork(); if (pid == -1 ) ERR_EXIT( "fork error" ); if (pid == 0 ) { close(pipefd[ 0 ]); write(pipefd[ 1 ], "hello" , 5 ); close(pipefd[ 1 ]); exit (EXIT_SUCCESS); } close(pipefd[ 1 ]); char buf[ 10 ] = { 0 }; read(pipefd[ 0 ], buf, 10 ); printf ( "buf=%s\n" , buf); return 0 ; }

创建进程池处理任务

Channel.hpp

# ifndef __CHANNEL_HPP__ # define __CHANNEL_HPP__ # include <iostream> # include <string> # include <unistd.h> // 先描述 class Channel { public : Channel ( int wfd, pid_t who) : _wfd(wfd), _who(who) { // Channel-3-1234 _name = "Channel-" + std:: to_string (wfd) + "-" + std:: to_string (who); } std::string Name () { return _name; } void Send ( int cmd) { :: write (_wfd, &cmd, sizeof (cmd)); } void Close () { :: close (_wfd); } pid_t Id () { return _who; } int wFd () { return _wfd; } ~ Channel () { } private : int _wfd; std::string _name; pid_t _who; }; # endif

ProcessPool.hpp

# ifndef __PROCESS_POOL_HPP__ # define __PROCESS_POOL_HPP__ # include <iostream> # include <string> # include <vector> # include <cstdlib> # include <unistd.h> # include <sys/types.h> # include <sys/wait.h> # include <functional> # include "Task.hpp" # include "Channel.hpp" // typedef std::function<void()> work_t; using work_t = std::function< void ()>; enum { OK = 0 , UsageError, PipeError, ForkError }; class ProcessPool { public : ProcessPool ( int n, work_t w) : processnum (n), work (w) { } // channels : 输出型参数 // work_t work: 回调 int InitProcessPool () { // 2. 创建指定个数个进程 for ( int i = 0 ; i < processnum; i++) { // 1. 先有管道 int pipefd[ 2 ] = { 0 }; int n = pipe (pipefd); if (n < 0 ) return PipeError; // 2. 创建进程 pid_t id = fork(); if (id < 0 ) return ForkError; // 3. 建⽴通信信道 if (id == 0 ) { // 关闭历史 wfd std::cout << getpid () << ", child close history fd: " ; for ( auto &c : channels) { std::cout << c. wFd () << " " ; c. Close (); } std::cout << " over" << std::endl; :: close (pipefd[ 1 ]); // read // child std::cout << "debug: " << pipefd[ 0 ] << std::endl; dup2 (pipefd[ 0 ], 0 ); work (); :: exit ( 0 ); } // ⽗进程执⾏ :: close (pipefd[ 0 ]); // write channels. emplace_back (pipefd[ 1 ], id); // Channel ch(pipefd[1], id); // channels.push_back(ch); } return OK; } void DispatchTask () { int who = 0 ; // 2. 派发任务 int num = 20 ; while (num--) { // a. 选择⼀个任务，整数 int task = tm. SelectTask (); // b. 选择⼀个⼦进程 channel Channel &curr = channels[who++]; who %= channels. size (); std::cout << "######################" << std::endl; std::cout << "send " << task << " to " << curr. Name () << ", 任务还剩 : " << num << std::endl; std::cout << "######################" << std::endl; // c. 派发任务 curr. Send (task); sleep ( 1 ); } } void CleanProcessPool () { // version 3 for ( auto &c : channels) { c. Close (); pid_t rid = :: waitpid (c. Id (), nullptr , 0 ); if (rid > 0 ) { std::cout << "child " << rid << " wait ... success" << std::endl; } } // version 2 // for (auto &c : channels) // for(int i = channels.size()-1; i >= 0; i--) // { // channels[i].Close(); // pid_t rid = ::waitpid(channels[i].Id(), nullptr, 0); // 阻塞了！ // if (rid > 0) // { // std::cout << "child " << rid << " wait ... success" << std::endl; // } // } // version 1 // for (auto &c : channels) // { // c.Close(); // } //? // for (auto &c : channels) // { // pid_t rid = ::waitpid(c.Id(), nullptr, 0); // if (rid > 0) // { // std::cout << "child " << rid << " wait ... success" << std::endl; // } // } } void DebugPrint () { for ( auto &c : channels) { std::cout << c. Name () << std::endl; } } private : std::vector<Channel> channels; int processnum; work_t work; }; # endif

Task.hpp

# pragma once # include <iostream> # include <unordered_map> # include <functional> # include <ctime> # include <sys/types.h> # include <unistd.h> using task_t = std::function< void ()>; class TaskManger { public : TaskManger () { srand ( time ( nullptr )); tasks. push_back ([]() { std::cout << "sub process[" << getpid () << " ] 执⾏访问数据库的任务 \n" << std::endl; }); tasks. push_back ([]() { std::cout << "sub process[" << getpid () << " ] 执⾏ url 解析 \n" << std::endl; }); tasks. push_back ([]() { std::cout << "sub process[" << getpid () << " ] 执⾏加密任务 \n" << std::endl; }); tasks. push_back ([]() { std::cout << "sub process[" << getpid () << " ] 执⾏数据持久化任务 \n" << std::endl; }); } int SelectTask () { return rand () % tasks. size (); } void Excute ( unsigned long number) { if (number > tasks. size () || number < 0 ) return ; tasks[number](); } ~ TaskManger () { } private : std::vector< task_t > tasks; }; TaskManger tm; void Worker () { while ( true ) { int cmd = 0 ; int n = :: read ( 0 , &cmd, sizeof (cmd)); if (n == sizeof (cmd)) { tm. Excute (cmd); } else if (n == 0 ) { std::cout << "pid: " << getpid () << " quit..." << std::endl; break ; } else { } } }

Main.cc

# include "ProcessPool.hpp" # include "Task.hpp" void Usage (std::string proc) { std::cout << "Usage: " << proc << " process-num" << std::endl; } // 我们⾃⼰就是 master int main ( int argc, char *argv[]) { if (argc != 2 ) { Usage (argv[ 0 ]); return UsageError; } int num = std:: stoi (argv[ 1 ]); ProcessPool *pp = new ProcessPool (num, Worker); // 1. 初始化进程池 pp-> InitProcessPool (); // 2. 派发任务 pp-> DispatchTask (); // 3. 退出进程池 pp-> CleanProcessPool (); // std::vector<Channel> channels; // // 1. 初始化进程池 // InitProcessPool(num, channels, Worker); // // 2. 派发任务 // DispatchTask(channels); // // 3. 退出进程池 // CleanProcessPool(channels); delete pp; return 0 ; }

Makefile

BIN=processpool CC=g++ FLAGS=-c -Wall -std=c++11 LDFLAGS=-o # SRC=$(shell ls *.cc) SRC= $( wildcard *.cc) OBJ=$(SRC:.cc=.o) $(BIN) : $(OBJ) $(CC) $(LDFLAGS) $@ $^ %.o:%.cc $(CC) $(FLAGS) $< .PHONY :clean clean: rm -f $(BIN) $(OBJ) .PHONY :test test: @echo $(SRC) @echo $(OBJ)

3.6管道读写规则

当没有数据可读时

O_NONBLOCK disable：read调⽤阻塞，即进程暂停执⾏，⼀直等到有数据来到为⽌。

O_NONBLOCK enable：read调⽤返回-1，errno值为EAGAIN。

当管道满的时候

O_NONBLOCK disable： write调⽤阻塞，直到有进程读⾛数据

O_NONBLOCK enable：调⽤返回-1，errno值为EAGAIN

如果所有管道写端对应的⽂件描述符被关闭，则read返回0

如果所有管道读端对应的⽂件描述符被关闭，则write操作会产⽣信号SIGPIPE,进⽽可能导致

write进程退出

当要写⼊的数据量不⼤于PIPE_BUF时，linux将保证写⼊的原⼦性。

当要写⼊的数据量⼤于PIPE_BUF时，linux将不再保证写⼊的原⼦性。

3.7管道特点

只能⽤于具有共同祖先的进程（具有亲缘关系的进程）之间进⾏通信；通常，⼀个管道由⼀个进

程创建，然后该进程调⽤fork，此后⽗、⼦进程之间就可应⽤该管道。

管道提供流式服务

⼀般⽽⾔，进程退出，管道释放，所以管道的⽣命周期随进程

⼀般⽽⾔，内核会对管道操作进⾏同步与互斥

管道是半双⼯的，数据只能向⼀个⽅向流动；需要双⽅通信时，需要建⽴起两个管道

4.命名管道

4.1创建一个命名管道

命名管道可以从命令⾏上创建，命令⾏⽅法是使⽤下⾯这个命令：

mkfifo filename

命名管道也可以从程序⾥创建，相关函数有：

int mkfifo ( const char *filename, mode_t mode);

创建命名管道:

int main ( int argc, char *argv[]) { mkfifo( "p2" , 0644 ); return 0 ; }

4.2匿名管道与命名管道的区别

匿名管道由pipe函数创建并打开。

命名管道由mkfifo函数创建，打开⽤open

FIFO（命名管道）与pipe（匿名管道）之间唯⼀的区别在它们创建与打开的⽅式不同，⼀但这些

⼯作完成之后，它们具有相同的语义。

4.3 命名管道的打开规则

如果当前打开操作是为读⽽打开FIFO时

O_NONBLOCK disable：阻塞直到有相应进程为写⽽打开该FIFO

O_NONBLOCK enable：⽴刻返回成功

如果当前打开操作是为写⽽打开FIFO时

O_NONBLOCK disable：阻塞直到有相应进程为读⽽打开该FIFO

O_NONBLOCK enable：⽴刻返回失败，错误码为ENXIO

5.共享内存

共享内存区是最快的IPC形式。⼀旦这样的内存映射到共享它的进程的地址空间，这些进程间数据传递不再涉及到内核，换句话说是进程不再通过执⾏进⼊内核的系统调⽤来传递彼此的数据

共享内存函数

shmget函数

功能：⽤来创建共享内存原型 int shmget(key_t key, size_t size, int shmflg); 参数 key: 这个共享内存段名字 size: 共享内存⼤⼩ shmflg: 由九个权限标志构成，它们的⽤法和创建⽂件时使⽤的 mode 模式标志是⼀样的取值为 IPC_CREAT ：共享内存不存在，创建并返回；共享内存已存在，获取并返回。取值为 IPC_CREAT | IPC_EXCL ：共享内存不存在，创建并返回；共享内存已存在，出错返回。返回值：成功返回⼀个⾮负整数，即该共享内存段的标识码；失败返回 -1

shmat函数

功能：将共享内存段连接到进程地址空间原型 void *shmat(int shmid, const void *shmaddr, int shmflg); 参数 shmid: 共享内存标识 shmaddr: 指定连接的地址 shmflg: 它的两个可能取值是 SHM_RND 和 SHM_RDONLY 返回值：成功返回⼀个指针，指向共享内存第⼀个节；失败返回 -1
shmaddr 为 NULL ，核⼼⾃动选择⼀个地址 shmaddr 不为 NULL 且 shmflg ⽆ SHM_RND 标记，则以 shmaddr 为连接地址。 shmaddr 不为 NULL 且 shmflg 设置了 SHM_RND 标记，则连接的地址会⾃动向下调整为 SHMLBA 的整数倍。公式：shmaddr - (shmaddr % SHMLBA) shmflg=SHM_RDONLY ，表⽰连接操作⽤来只读共享内存

shmdt函数

功能：将共享内存段与当前进程脱离原型 int shmdt(const void *shmaddr); 参数 shmaddr: 由 shmat 所返回的指针返回值：成功返回 0 ；失败返回 -1 注意：将共享内存段与当前进程脱离不等于删除共享内存段

shmctl函数

功能：⽤于控制共享内存原型 int shmctl(int shmid, int cmd, struct shmid_ds *buf); 参数 shmid: 由 shmget 返回的共享内存标识码 cmd: 将要采取的动作（有三个可取值） buf: 指向⼀个保存着共享内存的模式状态和访问权限的数据结构返回值：成功返回 0 ；失败返回 -1