嵌入式 Linux 多进程 IPC 通信方式很多,如何选择?

星标公众号，让嵌入式知识 “投喂” 不停歇！

大家好，我是杂烩君。

在嵌入式Linux开发中，多进程架构是实现模块化、提升系统可靠性的重要手段。本文带大家梳理消息队列、共享内存、UNIX域套接字、管道、信号量、信号六种进程间通信（IPC）机制的区别及使用场景。

如何选型？

在实际嵌入式项目中，选择哪种IPC方式取决于具体需求：

• 需要传输大量数据且追求极致性能？ → 共享内存 + 信号量
• 需要结构化消息的异步传递？ → 消息队列
• 需要双向通信且接口要灵活？ → UNIX域套接字
• 父子进程间简单的数据流传递？ → 管道
• 仅需要通知/同步，不传输数据？ → 信号或信号量
• 需要跨主机通信？ → TCP/IP套接字

什么是进程

1、进程和线程的区别

进程是指正在运行的程序，它拥有独立的内存空间和系统资源，不同进程之间的数据不共享。进程是资源分配的基本单位。

线程是进程内的执行单元，它与同一进程内的其他线程共享进程的内存空间和系统资源。线程是调度的基本单位。

下面用图来对比两者的关系——进程之间相互隔离，而同一进程内的线程共享地址空间：

进程间通信方式

每个进程各自有不同的用户地址空间，任何一个进程的全局变量在另一个进程中都看不到，进程间通信是指在不同进程之间传播或交换信息的一种机制。

进程间通信的目的：

• 传输数据。比如进程 A 负责生成数据，进程 B 负责处理数据，数据需要从 A 进程传输至 B 进程。
• 共享资源。比如进程 A 与进程 B 共享某一块内存资源。
• 模块化。将系统功能划分为多个进程模块进行开发，方便开发维护。
• 加速计算。多核处理器环境，一个特定进程划分为几个进程并行运行。

Linux IPC（Inter-process Communication， 进程间通信）的方式：

下面先通过一张对比表快速了解各种IPC方式的特点，再逐一深入讲解：

1、消息队列

内核中的一个优先级队列，多个进程通过访问同一个队列，进行添加结点或者获取结点实现通信。

下图展示了消息队列的工作原理——发送方将消息按优先级插入队列，接收方从队列头部取出消息：

POSIX消息队列头文件：

#include<fcntl.h>           /* For O_* constants */#include<sys/stat.h>        /* For mode constants */#include<mqueue.h>

编译链接需要加上 -lrt 链接。

消息队列基本API接口使用例子：发送进程给接收进程发送测试数据。

本例子消息队列通信的完整流程如下：

send.c：

#include<stdio.h>#include<stdlib.h>#include<string.h>#include<unistd.h>#include<fcntl.h>           /* For O_* constants */#include<sys/stat.h>        /* For mode constants */#include<mqueue.h>#define MQ_MSG_MAX_SIZE    512   ///< 最大消息长度 #define MQ_MSG_MAX_ITEM    5     ///< 最大消息数目staticmqd_t s_mq;typedefstruct _msg_data{char buf[128];int cnt;}msg_data_t;voidsend_data(void){staticint cnt = 0;msg_data_t send_data = {0};    cnt++;strcpy(send_data.buf, "hello");    send_data.cnt = cnt;int ret = mq_send(s_mq, (char*)&send_data, sizeof(send_data), 0);if (ret < 0)    {        perror("mq_send error");return;    }printf("send msg = %s, cnt = %d\n", send_data.buf, send_data.cnt);}intmain(void){int ret = 0;structmq_attrattr;///< 创建消息队列memset(&attr, 0, sizeof(attr));    attr.mq_maxmsg = MQ_MSG_MAX_ITEM;    attr.mq_msgsize = MQ_MSG_MAX_SIZE;    attr.mq_flags = 0;    s_mq = mq_open("/mq", O_CREAT|O_RDWR, 0777, &attr);if(-1 == s_mq)    {        perror("mq_open error");return-1;    }for (size_t i = 0; i < 10; i++)    {        send_data();        sleep(1);    }    mq_close(s_mq);return0;}

recv.c：

#include<stdio.h>#include<stdlib.h>#include<string.h>#include<unistd.h>#include<fcntl.h>           /* For O_* constants */#include<sys/stat.h>        /* For mode constants */#include<mqueue.h>#define MQ_MSG_MAX_SIZE    512   ///< 最大消息长度 #define MQ_MSG_MAX_ITEM    5     ///< 最大消息数目staticmqd_t s_mq;typedefstruct _msg_data{char buf[128];int cnt;}msg_data_t;intmain(void){msg_data_t recv_data = {0};int prio = 0;ssize_t len = 0;    s_mq = mq_open("/mq", O_RDONLY);if(-1 == s_mq)    {        perror("mq_open error");return-1;    }while (1)    {if((len = mq_receive(s_mq, (char*)&recv_data, MQ_MSG_MAX_SIZE, &prio)) == -1)        {            perror("mq_receive error");return-1;        }printf("recv_msg = %s, cnt = %d\n", recv_data.buf, recv_data.cnt);        sleep(1);    }    mq_close(s_mq);    mq_unlink("/mq");return0;}

编译、运行：

gcc send.c -o send_process -lrtgcc recv.c -o recv_process -lrt

2、共享内存

消息队列的读取和写入的过程，会有发生用户态与内核态之间的消息拷贝过程。而共享内存的方式则没有这个拷贝过程，进程间通信速度较快。

在物理内存上开辟一块内存空间，多个进程可以将同一块物理内存空间映射到自己的虚拟地址空间，通过自己的虚拟地址直接访问这块空间，通过这种方式实现数据共享。

下图展示了共享内存的核心原理——多个进程的虚拟地址通过页表映射到同一块物理内存，实现零拷贝通信：

对比消息队列需要两次数据拷贝，共享内存的优势一目了然：

POSIX共享内存头文件：

#include<sys/mman.h>#include<sys/stat.h>#include<unistd.h>

共享内存基本API接口使用例子：发送进程给接收进程发送测试数据。

send.c：

#include<stdio.h>#include<stdlib.h>#include<string.h>#include<unistd.h>#include<fcntl.h>           /* For O_* constants */#include<sys/stat.h>        /* For mode constants */#include<sys/mman.h>#include<semaphore.h>#define SHM_NAME "/shm"#define SEM_NAME "/shm_sem"intmain(void){///< 创建信号量，初始值为0，表示数据尚未就绪sem_t *sem = sem_open(SEM_NAME, O_CREAT, 0666, 0);if (sem == SEM_FAILED)    {        perror("sem_open error");return-1;    }///< 创建和读端相同的文件标识int shm_fd = shm_open(SHM_NAME, O_RDWR | O_CREAT, 0666);if (shm_fd == -1)     {        perror("shm_open error");return-1;    }///< 设置共享内存文件为8KB    ftruncate(shm_fd , 8 * 1024);///< 获取共享内存文件相关属性信息structstatfilestat = {0};    fstat(shm_fd, &filestat);printf("st_size = %ld\n",filestat.st_size);///< 内存映射char *shm_ptr = (char*)mmap(NULL, filestat.st_size, PROT_READ|PROT_WRITE, MAP_SHARED, shm_fd, 0);    close(shm_fd);///< 向共享内存中写入数据char buf[] = "hello world";    memmove(shm_ptr, buf, sizeof(buf));printf("pid %d, %s\n", getpid(), shm_ptr);///< 写入完成，通知接收方数据已就绪    sem_post(sem);///< 解除映射，关闭信号量    munmap(shm_ptr, filestat.st_size);    sem_close(sem);return0;}

recv.c：

#include<stdio.h>#include<stdlib.h>#include<string.h>#include<unistd.h>#include<fcntl.h>           /* For O_* constants */#include<sys/stat.h>        /* For mode constants */#include<sys/mman.h>#include<semaphore.h>#define SHM_NAME "/shm"#define SEM_NAME "/shm_sem"intmain(void){///< 打开信号量sem_t *sem = sem_open(SEM_NAME, O_CREAT, 0666, 0);if (sem == SEM_FAILED)    {        perror("sem_open error");return-1;    }///< 创建共享内存文件标识符int shm_fd = shm_open(SHM_NAME, O_RDWR | O_CREAT, 0666);if (shm_fd == -1)     {        perror("shm_open failed");exit(EXIT_FAILURE);    }///< 设置共享内存文件为8KB    ftruncate(shm_fd , 8192);///< 获取共享内存文件相关属性信息structstatfilestat;    fstat(shm_fd, &filestat);printf("st_size = %ld\n",filestat.st_size);///< 映射共享内存，并获取共享内存的地址char *shm_ptr = (char*)mmap(NULL, filestat.st_size, PROT_READ|PROT_WRITE, MAP_SHARED, shm_fd, 0);    close(shm_fd);///< 等待发送方写入完成    sem_wait(sem);///< 获取共享内存地址中的内容并打印printf("pid = %d, %s\n", getpid(), shm_ptr);///< 解除映射，清理信号量和共享内存    munmap(shm_ptr, filestat.st_size);    shm_unlink(SHM_NAME);    sem_close(sem);    sem_unlink(SEM_NAME);return0;}

编译、运行：

gcc send.c -o send_process -lrtgcc recv.c -o recv_process -lrt

对具有多个处理核系统消息传递的性能要优于共享内存。共享内存会有高速缓存一致性问题，这是由共享数据在多个高速缓存之间迁移而引起的。随着系统的处理核的数量的日益增加，可能导致消息传递作为 IPC 的首选机制。

注意： 共享内存本身不提供任何同步机制。上面的示例使用有名信号量（初始值为0）实现了"写完再读"的同步：发送方写入后 sem_post 将信号量加1，接收方 sem_wait 阻塞直到信号量大于0才读取。生产代码中如果涉及多读多写的并发场景，还需要进一步使用互斥锁保护临界区。

3、socket

消息队列和共享内存适合数据传输，但如果需要像网络编程一样灵活的双向通信，UNIX域套接字是更好的选择。

UNIX域套接字与传统基于TCP/IP协议栈的socket不同，unix domain socket以文件系统作为地址空间，不需经过TCP/IP的头部封装、报文ack确认、路由选择、数据校验与重传过程，因此传输速率上也不会受网卡带宽的限制。

unix domain socket在进程间通信同样是基于“客户端—服务器”（C-S）模式。

UNIX 域套接字与 TCP/IP 套接字的对比：

UNIX域套接字基本API接口使用例子：基于UNIX域套接字客户端进程向服务端进程发送测试数据。

下图展示了 UNIX 域套接字的 C/S 通信建立过程：

server.c：

#include<stdio.h>#include<stdlib.h>#include<string.h>#include<unistd.h>#include<fcntl.h>           /* For O_* constants */#include<sys/stat.h>        /* For mode constants */#include<sys/socket.h>#include<netinet/in.h>#include<arpa/inet.h>#include<sys/un.h>#define SERVER_PATH "/tmp/server"intmain(void){///< 创建UNIX域字节流套接字int server_fd = socket(AF_LOCAL, SOCK_STREAM, 0);if(server_fd < 0)    {printf("socket error\n");exit(EXIT_FAILURE);    }///< 绑定服务端地址    unlink(SERVER_PATH);structsockaddr_unserver_addr;memset((char*)&server_addr, 0, sizeof(server_addr));    server_addr.sun_family = AF_LOCAL;strncpy(server_addr.sun_path, SERVER_PATH, sizeof(server_addr.sun_path)-1);if(bind(server_fd, (struct sockaddr*)&server_addr, sizeof(server_addr)) < 0)    {printf("bind error\n");        close(server_fd);exit(EXIT_FAILURE);    }///< 监听if(listen(server_fd, 10) < 0)       {printf("listen error\n");        close(server_fd);exit(EXIT_FAILURE);    }///< 等待客户端连接int addr_len = sizeof(struct sockaddr);structsockaddr_unclient_addr;int client_fd = accept(server_fd, (struct sockaddr*)&client_addr, (socklen_t *)&addr_len);if(client_fd < 0)    {printf("accept error\n");        close(server_fd);        unlink(SERVER_PATH);exit(1);     }else    {printf("connected client: %s\n", client_addr.sun_path);    }while(1)    {char buf[128] = {0};int recv_len = read(client_fd, buf, sizeof(buf)); if(recv_len <= 0)        {printf("recv error!\n");            close(client_fd);exit(EXIT_FAILURE);        }printf("recv : %s\n", buf);    }    unlink(SERVER_PATH);    close(server_fd);    close(client_fd);return0;}

client.c：

#include<stdio.h>#include<stdlib.h>#include<string.h>#include<unistd.h>#include<fcntl.h>           /* For O_* constants */#include<sys/stat.h>        /* For mode constants */#include<sys/socket.h>#include<netinet/in.h>#include<arpa/inet.h>#include<sys/un.h>#define SERVER_PATH "/tmp/server"#define CLIENT_PATH "/tmp/client"intmain(void){///< 创建UNIX域字节流套接字int client_fd = socket(AF_UNIX, SOCK_STREAM, 0);if(client_fd < 0)    {printf("socket error\n");exit(EXIT_FAILURE);    }///< 显式绑定客户端地址structsockaddr_unclient_addr;memset((char*)&client_addr, 0, sizeof(client_addr));    client_addr.sun_family = AF_UNIX;strncpy(client_addr.sun_path, CLIENT_PATH, sizeof(client_addr.sun_path)-1);    unlink(CLIENT_PATH);if(bind(client_fd, (struct sockaddr*)&client_addr, sizeof(client_addr)) < 0)    {printf("bind error\n");        close(client_fd);exit(1);     }///< 连接服务端structsockaddr_unserver_addr;    server_addr.sun_family  = AF_UNIX;strncpy(server_addr.sun_path, SERVER_PATH, sizeof(server_addr.sun_path)-1);int ret = connect(client_fd, (struct sockaddr*)&server_addr, sizeof(server_addr)); if(ret < 0)    {printf("connect error\n");        close(client_fd);        unlink(CLIENT_PATH);exit(1);      } printf("connect to server: %s\n", server_addr.sun_path);while(1)    {char buf[128] = {0};if (scanf("%s", buf))        {int send_len = write(client_fd, buf, strlen(buf));if (send_len <= 0)            {printf("write error!\n");                close(client_fd);exit(EXIT_FAILURE);              }else            {printf("send success! send: %s, send_len: %d\n", buf, send_len);            }        }     }    unlink(SERVER_PATH);    close(client_fd);return0;}

编译、运行：

gcc server.c -o server_processgcc client.c -o client_process

类socket的其它进程间通信方式：