进程间通信介绍
进程间通信目的
- 数据传输：一个进程需要将它的数据发送给另一个进程
- 资源共享：多个进程之间共享同样的资源。
- 通知事件：一个进程需要向另一个或一组进程发送消息，通知它（它们）发生了某种事件（如进程终止时要通知父进程）。
- 进程控制：有些进程希望完全控制另一个进程的执行（如Debug进程），此时控制进程希望能够拦截另一个进程的所有陷入和异常，并能够及时知道它的状态改变。
管道
什么是管道
管道是Unix中最古老的进程间通信的形式。我们把从一个进程连接到另一个进程的一个数据流称为一个“管道”如：
用fork来共享管道原理
站在文件描述符角度-深度理解管道 （内存级）
左边是进程管理，右边是文件管理，进程通过全局变量能找到文件描述符标，也就找到了对应的file文件也能打开磁盘上的文件拷贝到缓冲区里进行读写，父进程创建子进程，发生写实拷贝，类似于浅拷贝，此时不做文件操作，文件描述符里面的对于关系和父进程一样，struc_file没有关闭，因为父进程还在， struc_file的对应的引用计数不为0就一定不会关闭，既然访问的同一个struc_file，也可以同时访问其中的file文件，通过缓冲区可以进行文件的读写
- 父进程通过读写两种方式打开内存级的文件返回给上层完成管道的创建
- 子进程继承父进程的文件描述符表，发生浅拷贝，也能拿到父进程以读写打开的管道文件
- 父子都看得到，让父子单向通信，父进程写，子进程读，各自关闭掉自己不需要的文件描述符
这个是管道是OS单独设计的，得配上单独的系统调用：pipe 内存级的，不需要文件路径，没有文件名，所以叫匿名管道，那我们怎么保证，两个进程打开的是同一个管道的？
子进程继承了父进程的文件描述符表
站在内核角度-管道本质
匿名管道
#include <unistd.h>
功能:创建⼀⽆名管道
原型
int pipe(int fd[2]);
参数
fd：⽂件描述符数组,其中fd[0]表⽰读端, fd[1]表⽰写端
返回值:成功返回0，失败返回错误代码#include<iostream>
#include<unistd.h>
using namespace std;
int main()
{
int fds[2]={0};
int n=pipe(fds);
if(n<0)
{
cerr<<"Pipe error"<<endl;
return 1;
}
cout<<"fds[0]"<<fds[0]<<endl;
cout<<"fds[1]"<<fds[1]<<endl;
return 0;
}
0，1，2被三个标准占用了，从3，4开始
父写子读，实现通信
#include<iostream>
#include<unistd.h>
#include<sys/wait.h>
#include<sys/types.h>
#include<cstring>
using namespace std;
void ChildWrite(int fd)
{
char buffer[1024];
int cnt=0;
while(true)
{
snprintf(buffer,sizeof(buffer),"I am child,pid:%d cnt:%d\n",getpid(),cnt++);
write(fd,buffer,strlen(buffer));
sleep(1);
}
}
void FatherRead(int fd)
{
char buffer[1024];
while(true)
{
buffer[0]=0;
ssize_t n=read(fd,buffer,sizeof(buffer)-1);
if(n>0)
{
buffer[n]=0;
cout<<"child says: "<<buffer<<endl;
}
}
}
int main()
{ //1.创建管道
int fds[2]={0}; //fds[0]读端，fds[1]写端
int n=pipe(fds);
if(n<0)
{
cerr<<"Pipe error"<<endl;
return 1;
}
cout<<"fds[0]"<<fds[0]<<endl;
cout<<"fds[1]"<<fds[1]<<endl;
//3.创建子进程 f->r c->w
pid_t id=fork();
if(id==0)
{
//child
close(fds[0]); //关闭读端
ChildWrite(fds[1]);
close(fds[1]);
exit(0);
}
close(fds[1]); //关闭写端
FatherRead(fds[0]);
waitpid(id,nullptr,0);
close(fds[0]);
return 0;
}
父进程写的cnt在不断变化，子进程能读到，说明实现了管道通信
五种特性
1.匿名管道，只能用来进行具有血缘关系的进程进行进程间通信(常用父与子，如上述代码)
2.管道文件，自带同步机制（父子进程进行IO同时进行，一个读一个写，不管是父快还是子快，父不断写，子read读不到会阻塞住直到读到为止）
3.管道是面向字节流的
4.管道是单向通信的（要么父写子读，要么子写父读）
5. （管道）文件的生命周期随进程
4种通信情况
1.写慢，读快------>读端阻塞等待写端(进程)
2.写快，读慢------>缓冲区写满了，写要阻塞等待读端
3.写关，读开------>read会读到返回值0，表示文件结尾
//写一条就关
void ChildWrite(int fd)
{
char buffer[1024];
int cnt=0;
while(true)
{
snprintf(buffer,sizeof(buffer),"I am child,pid:%d cnt:%d\n",getpid(),cnt++);
write(fd,buffer,strlen(buffer));
sleep(1);
break;
}
}
//观察n的返回值
void FatherRead(int fd)
{
char buffer[1024];
while(true)
{
buffer[0]=0;
ssize_t n=read(fd,buffer,sizeof(buffer)-1);
if(n>0)
{
buffer[n]=0;
cout<<"child says: "<<buffer<<endl;
}
else
{
cout<< "n:"<<n<<endl;
}
}
}
4.读关，写开------->写端再写没有意义，OS会杀掉写端进程
#include<iostream>
#include<unistd.h>
#include<sys/wait.h>
#include<sys/types.h>
#include<cstring>
using namespace std;
void ChildWrite(int fd)
{
char buffer[1024];
int cnt=0;
while(true)
{
snprintf(buffer,sizeof(buffer),"I am child,pid:%d cnt:%d\n",getpid(),cnt++);
write(fd,buffer,strlen(buffer));
sleep(1);
}
}
void FatherRead(int fd)
{
char buffer[1024];
while(true)
{
buffer[0]=0;
ssize_t n=read(fd,buffer,sizeof(buffer)-1);
if(n>0)
{
buffer[n]=0;
cout<<"child says: "<<buffer<<endl;
}
else
{
cout<< "n:"<<n<<endl;
sleep(4);
}
break;
}
}
int main()
{ //1.创建管道
int fds[2]={0}; //fds[0]读端，fds[1]写端
int n=pipe(fds);
if(n<0)
{
cerr<<"Pipe error"<<endl;
return 1;
}
cout<<"fds[0]"<<fds[0]<<endl;
cout<<"fds[1]"<<fds[1]<<endl;
//3.创建子进程 f->r c->w
pid_t id=fork();
if(id==0)
{
//child
close(fds[0]); //关闭读端
ChildWrite(fds[1]);
close(fds[1]);
exit(0);
}
close(fds[1]); //关闭写端
FatherRead(fds[0]);
close(fds[0]);
int status;
int ret=waitpid(id,&status,0);
//获取到子进程的退出状态
if(ret>0)
{
printf("child code: %d exited status: %d\n",(status)>>8&0xff,(status)&0x7f);
}
return 0;
}
发送异常信号13 SIGPIPE
基于匿名管道----进程池
父进程创建多个子进程，用匿名管道分派任务
1.构建进程链，为进程池做准备
#ifndef __PROCESS__POOL_HPP__
#define __PROCESS__POOL_HPP__
#include <iostream>
#include <vector>
#include <unistd.h>
#include<cstdlib>
using namespace std;
const int gdefaultnum = 5; // 要创建几个进程
// 先描述 单个进程
class Channel
{
public:
Channel() {}
~Channel() {}
private:
int _wfd;
};
// 在组织 进程链
class ChannelManager
{
public:
ChannelManager() {}
~ChannelManager() {}
private:
vector<Channel> _channels;
};
2.创建进程池，提供管道条件
// 进程池
class ProcessPool
{
public:
ProcessPool(int num) : _process_num(num) {}
~ProcessPool() {}
bool Create()
{
int pipefd[2] = {0};
for (int i = 0; i < _process_num; i++)
{
// 1.创建管道
int n = pipe(pipefd);
if (n < 0)
return false;
}
// 2.创建子进程 各自关闭不需要的文件描述符
pid_t id = fork();
if (id < 0)
return false;
else if (id == 0)
{
// 子进程 --->读
// 3.关闭不需要的文件描述符
close(pipefd[1]);
exit(0);
}
else
{
// 父进程 --->写
// 3.关闭不需要的文件描述符
close(pipefd[0]);
}
return true;
}
private:
ChannelManager _cm; // 进程链
int _process_num; // 进程个数
};3.父子各自打印验证是否通信
void BuildChannel(int wfd, pid_t subid)
{
_channels.emplace_back(wfd, subid);
// Channel c(wfd,subid);
// _channels.push_back(c);
}
void Print()
{
for(auto &chnnel : _channels)
{
cout<<chnnel.Name()<<endl;
}
}
void Work(int rfd)
{
while (true)
{
cout << "我是子进程，我的rfd是：" << rfd << endl;
sleep(2);
}
}//ProcessPool.hpp
#ifndef __PROCESS__POOL_HPP__
#define __PROCESS__POOL_HPP__
#include <iostream>
#include <vector>
#include <unistd.h>
#include <cstdlib>
using namespace std;
const int gdefaultnum = 5; // 要创建几个进程
// 先描述 单个进程
class Channel
{
public:
Channel(int fd, pid_t id) : _wfd(fd), _subid(id) { _name = "chnnel-" + std::to_string(_wfd) + "-" + std::to_string(_subid); }
~Channel() {}
int Fd(){return _wfd;}
pid_t Subid(){return _subid;}
string Name(){return _name;}
private:
int _wfd;
pid_t _subid;
std::string _name;
};
// 在组织 进程链
class ChannelManager
{
public:
ChannelManager() {}
~ChannelManager() {}
void BuildChannel(int wfd, pid_t subid)
{
_channels.emplace_back(wfd, subid);
// Channel c(wfd,subid);
// _channels.push_back(c);
}
void Print()
{
for(auto &chnnel : _channels)
{
cout<<chnnel.Name()<<endl;
}
}
private:
vector<Channel> _channels;
};
// 进程池
class ProcessPool
{
public:
ProcessPool(int num) : _process_num(num) {}
~ProcessPool() {}
void Work(int rfd)
{
while (true)
{
cout << "我是子进程，我的rfd是：" << rfd << endl;
sleep(2);
}
}
bool Create()
{
for (int i = 0; i < _process_num; i++)
{
int pipefd[2] = {0};
// 1.创建管道
int n = pipe(pipefd);
if (n < 0)
return false;
// 2.创建子进程 父子各自关闭不需要的文件描述符
pid_t id = fork();
if (id < 0)
return false;
else if (id == 0)
{
// 子进程 --->读
// 3.关闭不需要的文件描述符
close(pipefd[1]);
Work(pipefd[0]);
exit(0);
}
else
{
// 父进程 --->写
// 3.关闭不需要的文件描述符
close(pipefd[0]);
_cm.BuildChannel(pipefd[1], id);
close(pipefd[1]);
}
}
return true;
}
void Debug()
{
_cm.Print();
}
private:
ChannelManager _cm; // 进程链
int _process_num; // 进程个数
};
#endif//Main.cc
#include"ProcessPool.hpp"
int main()
{
ProcessPool pp(gdefaultnum);
//创建进程池
pp.Create();
//打印进程池
pp.Debug();
sleep(1000);
return 0;
}实现通信
4.分配任务，子写父读
void Work(int rfd)
{
while (true)
{
int code = 0;
ssize_t n = read(rfd, &code, sizeof(code));
if (n > 0)
{
if (n == sizeof(code))
{
continue;
}
cout << "子进程[]"<<getpid()<<"]收到一个任务码:" << code << endl;
}
else if (n == 0)
{
cout << "子进程退出" << endl;
break;
}
else
{
// 读失败
cout << "读取错误" << endl;
break;
}
}
}
void PushTack(int taskcode)
{
// 1.选择一个子进程，采用轮询，防止负载均衡和负载不均衡
auto &c = _cm.Select();
cout << "选择一个子进程：" << c.Name() << endl;
// 2.发送任务
c.Send(taskcode);
cout << "发送了一个任务码：" << taskcode << endl;
}
完整代码
//ProcessPool.hpp
#ifndef __PROCESS__POOL_HPP__
#define __PROCESS__POOL_HPP__
#include <iostream>
#include <vector>
#include <unistd.h>
#include <cstdlib>
using namespace std;
const int gdefaultnum = 5; // 要创建几个进程
// 先描述 单个进程
class Channel
{
public:
Channel(int fd, pid_t id) : _wfd(fd), _subid(id) { _name = "chnnel-" + std::to_string(_wfd) + "-" + std::to_string(_subid); }
~Channel() {}
int Fd() { return _wfd; }
pid_t Subid() { return _subid; }
string Name() { return _name; }
void Send(int code)
{
int n = write(_wfd, &code, sizeof(code));
(void)n;
}
private:
int _wfd;
pid_t _subid;
std::string _name;
};
// 在组织 进程链
class ChannelManager
{
public:
ChannelManager() : _next(0) {}
~ChannelManager() {}
void BuildChannel(int wfd, pid_t subid)
{
_channels.emplace_back(wfd, subid);
// Channel c(wfd,subid);
// _channels.push_back(c);
}
// 轮询
Channel &Select()
{
auto &c = _channels[_next];
_next++;
_next %= _channels.size();
return c;
}
void Print()
{
for (auto &chnnel : _channels)
{
cout << chnnel.Name() << endl;
}
}
private:
vector<Channel> _channels;
int _next;
};
// 进程池
class ProcessPool
{
public:
ProcessPool(int num) : _process_num(num) {}
~ProcessPool() {}
void Work(int rfd)
{
while (true)
{
int code = 0;
ssize_t n = read(rfd, &code, sizeof(code));
if (n > 0)
{
if (n == sizeof(code))
{
continue;
}
cout << "子进程[]"<<getpid()<<"]收到一个任务码:" << code << endl;
}
else if (n == 0)
{
cout << "子进程退出" << endl;
break;
}
else
{
// 读失败
cout << "读取错误" << endl;
break;
}
}
}
bool Create()
{
for (int i = 0; i < _process_num; i++)
{
int pipefd[2] = {0};
// 1.创建管道
int n = pipe(pipefd);
if (n < 0)
return false;
// 2.创建子进程 父子各自关闭不需要的文件描述符
pid_t id = fork();
if (id < 0)
return false;
else if (id == 0)
{
// 子进程 --->读
// 3.关闭不需要的文件描述符
close(pipefd[1]);
Work(pipefd[0]);
exit(0);
}
else
{
// 父进程 --->写
// 3.关闭不需要的文件描述符
close(pipefd[0]);
_cm.BuildChannel(pipefd[1], id);
close(pipefd[1]);
}
}
return true;
}
void Debug()
{
_cm.Print();
}
void PushTack(int taskcode)
{
// 1.选择一个子进程，采用轮询，防止负载均衡和负载不均衡
auto &c = _cm.Select();
cout << "选择一个子进程：" << c.Name() << endl;
// 2.发送任务
c.Send(taskcode);
cout << "发送了一个任务码：" << taskcode << endl;
}
private:
ChannelManager _cm; // 进程链
int _process_num; // 进程个数
};
#endif//Main.cc
#include"ProcessPool.hpp"
int main()
{
ProcessPool pp(gdefaultnum);
//创建进程池
pp.Create();
//打印进程池
//pp.Debug();
int task_code = 1;
while(true)
{
pp.PushTack(task_code++);
sleep(1);
}
return 0;
}创建进程池后，OS关闭了没有意义的管道，每次选择一个管道接受消息
