Linux 進程間通信(包含一個經典的生產者消費者實例代碼）

前言：編寫多進程程序時，有時不可避免的需要在多個進程之間傳遞數據，我們知道，進程的用戶的地址空間是獨立，父進程中對數據的修改並不會反映到子進程中，但內核是共享的，大多數進程間通信方式都是在內核中建立一塊存儲區域，用來實現進程間的通信(也可以將數據寫進文件，通過文件操作，但文件操作的開銷會比較大）。

一.管道通信方式：管道通信具有單向，無結構，先進先出的字節流特點;管道有2個端點，一個端點寫入數據，一個端點讀取數據，當數據從管道中讀出時，這些數據將被移走。當進程從空管道中讀取數據或向已滿的管道寫入數據時，進程將被掛起，直到有進程向管道中寫入數據或從管道中讀取數據，此時，進程將由等待狀態變為就緒狀態。對管道的操作和對文件的操作差不多。根據管道提供的應用接口的不同，管道可分為命名管道和無名管道。

1.無名管道

#include <unistd.h>

int pipe(int fd[2])        

 // 創建一個無名管道fd包含2個文件描述符的數組，fd[0]用於讀，fd[1]用於寫若成功返回0,否則反回-1

一般某個進程用於讀，另一個進程用於寫，用於讀的進程需要close(fd[1]),用於寫的進程需要close(fd[0]);

 

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
int main(void)
{
    pid_t pid;
    int fd[2],i,n;
    char chr;
    pipe(fd);
    pid=fork();
    if(pid==0)          //子進程
    {
        close(fd[1]);
        for(i=0;i<10;i++)
        {
            read(fd[0],&chr,1);
            printf("%c\n",chr);
        }
        close(fd[0]);
        exit(0);
    }
    close(fd[0]);           //父進程
    for(i=0;i<10;i++)
    {
        chr='a'+i;
        write(fd[1],&chr,1);
        sleep(1);
    }
    close(fd[1]);
    return 0;
}

若只想讀取某個進程的結果，或寫入某個進程的輸入可以使用popen函數，popen函數首先調用pipe創建管道，然后調用fork函數創建子進程，在子進程中調用execve函數

執行相關命令。

#include <stdio.h>

FILE *popen(const char *command,const char *type)

 //  command執行的shell命令，type命令的輸入輸出類型（r：打開命令執行的標准輸出w：打開命令執行的標准輸入)，成功返回文件I/O流否則返回NULL

int pclose(FILE* stream)       //返回命令執行的返回狀態

#include <stdio.h>
int main()
{
    FILE *fp;
    fp=popen("/bin/ls -a","r");
    char buf[256];
    int line=1;
    while(fgets(buf,256,fp)!=NULL)
    {
        printf("%d: %s",line++,buf);
    }
    pclose(fp);
    return 0;
}

命名管道

命名管道作為一個特殊的文件保存在文件系統中，任意一個進程都可以對命名管道進行讀寫，只要進程具有相應的讀寫權限

#include <sys/types.h>

#include <sys/stat.h>

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

//創建命名管道 pathname文件路徑名，mode存取權限，若成功返回0,否則返回-1

#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
int main()
{
    mkfifo("fifo",0660);   //創建一個命令管道，屬主和用戶組具有讀寫權限
    pid_t pid;
    pid=fork();
    if(pid==0)
    {
        char buf[256];
        int fd=open("fifo",O_RDONLY);   //子進程讀管道中的數據
        read(fd,buf,10);
        buf[10]=0;
        printf("%s",buf);
        close(fd);
        exit(0);
    }
    int fd=open("fifo",O_WRONLY);    //父進程向管道寫入數據
    write(fd,"fifo test\n",10);
    close(fd);
    return 0;
}

 IPC（進程間通信）:IPC的內容包括信號量，消息隊列和共享內存，即一個IPC對象可以是一個信號量也可以是一個消息隊列也可以是一個共享內存。每一個IPC對象都有一個正整數標識，與文件描述符不同的是ipc對象的標識是全局的，用於識別整個系統中不同的ipc對象。ipc對象的標識由進程運行時決定，不是每次都相同。因此進程為了通信，不僅要知道使用的IPC對象類型，而且也需要知道ipc對象標識。但是進程在執行前並不知道IPC對象標識，下面給出2種辦法

（1）創建IPC對象時，key使用IPC_PRIVATE,這樣就保證返回一個新的IPC對象，然后將這一標識存放於某個文件中，其他進程便可打開文件獲得該IPC對象標識

（2）創建某一個IPC對象時，key值隨機使用一個數字，在不同的進程中，對於創建IPC對象的進程，根據key值得到一個IPC對象標識，對於獲取IPC對象的進程，使用相同的key值就可以得到與該key值相對應的IPC對象的標識

下面一段代碼說明了key值的作用

#include <stdio.h>  
#include <sys/ipc.h>  
#include <sys/sem.h> 
#include <sys/types.h>  
#define MY_SEM_ID 120
int main ( int argc, char *argv[] )
{      

    int semid;
    semid=semget(MY_SEM_ID,1,0666|IPC_CREAT);//如果把MY_SEM_ID換成IPC_PRIVATE那么每次程序的執行
    printf("semid=%d",semid);                //結果都不相同，若是MY_SEM_ID，則結果相同即使在不同的
    return 0;                                //進程中，即只要key相同，則返回的標識就相同
    
} 

IPC相關API

信號量(用與多進程程序在存取共享資源時的同步控制）

#include <sys/types.h>

#include <sys/ipc.h>

#include <sys/sem.h>

int semget(key_t key,int nsems,int semflg)

 函數說明：功能：獲得或創建信號量，key:根據key生成信號量標識，nsems：創建的信號量集中的信號量的個數，該參數只在創建信號量集時有效。，semflg：存取權限或創建條件若為0則用於獲得已存在的信號量若為IPC_CREAT|perm perm為存取權限，則用於創建信號量，成功返回信號量標識，出錯返回-1

 

int semop(int semid,struct sembuf* sops,unsigned nsops)

函數說明：功能：獲得或者釋放信號量，semid:信號量標識，sops指向由sembuf組成的數組，nsops信號量的個數，成功返回0,否則返回-1

struct sembuf{

ushort sem_num;   //在信號量數組中的索引

short sem_op;       //要執行的操作,若sem_op大於0那么操作為將sem_op加入到信號量的值中，並喚醒等待信號增加的進程;若sem_op等於0,當信號量的值也是0時， 函數返回，否則阻塞直到信號量的值為0；若sem_op小於0,則判斷信號量的值加上sem_op的值，如果結果為0,喚醒等待信號量為0的進程，如果小於0，調用該函數的進程阻塞，如果大於0,那么從信號量里減去這個值並返回。

short sem_flg;         //操作標致SEM_UNDO會阻塞，IPC_NOWAIT不會阻塞

};

 

int semctl(int semid,int semnum,int cmd,union semun arg);

函數說明：功能：在信號量集上的控制操作，semid信號量集的標識，semnum信號量集的第幾個信號量，撤銷信號量集時，次參數可缺省，cmd用於指定操作類別，值為GETVAL獲得信號量的值，SETVAL設置信號量的值，GETPID獲得最后一次操作信號量的進程，GETNCNT獲得正在等待信號量的進程數，GETZCNT獲得等待信號量值變為0的進程數，IPC_RMID 刪除信號量或信號量數組

 

共享內存

共享內存是內核中的一塊存儲空間，這塊內存被映射至多個進程的虛擬地址空間。共享內存在不同進程虛擬地址空間中映射地址未必相同。

相關API

#include <sys/ipc.h>

 #include <sys/shm.h>

int shmget(key_t key,int size,int shmflg)

函數說明：功能：創建或獲得共享內存，key：作用同上，size：共享內存的大小，shmflg:存取權限或創建條件，若為IPC_CREAT|perm perm為存取權限，則表示創建共享內存，為0表示獲得共享內存

 

void * shmat(int shmid,const void *shmaddr,int shmflg)

函數說明:功能：將創建的共享內存映射到進程虛擬地址空間的某個位置，shmid：共享內存標識，shmaddr要映射到的進程虛擬空間地址，若為NULL，則由系統決定對應的地址，shmflg：指定如何使用共享內存，若指定了SHM_RDONLY位則表示以只讀的方式使用此段，否則以讀寫的方式使用此段.成功返回映射的地址失敗返回-1

 

int shmdt(const void* shmaddr);

函數說明：解除共享內存的映射，shmaddr:共享內存的映射地址，成功返回0,否則返回-1

 

int shmctl(int shmid,int cmd,struct shmid_ds *buf)

函數說明：對以存在的共享內存進行操作，shmid：共享內存標識，cmd：操作類型：cmd 為IPC_STAT 獲取共享內存的狀態，IPC_/SET設置共享內存的權限,IPC_RMID刪除共享內存，IPC_LOCK 鎖定共享內存，使共享內存不被置換出去，IPC_UNLOCK解鎖

 

　　　　struct shmid_ds{

　　　　struct ipc_perm 　　shm_perm;   //存取權限

　　　　int　　　　　　　　shm_segsz; //共享內存大小

　　　　__kernel_time_t       shm_atime;  //最后映射時間

 　　　  __kernel_time_t       shm_dtime;  //最后解除映射時間

　　　　__kernel_time_t       shm_ctime;  //最后修改時間

　　　　__kernel_ipc_pid_t  shm_cpid;    //創建進程ID

　　　　__kernel_ipc_pid_t  shm_lpid;    //最近操作進程ID

　　　　unsigned short         shm_nattch; //建立映射的進程數

　　　　}

 

下面是一個經典的生產者消費者代碼實例　　　

//semshm.h 封裝了創建信號量刪除信號量，P操作，V操作
#ifndef SEMSHM_H
#define SEMSHM_H

#define SHM_KEY 9494
#define SEM_KEY_1 9399
#define SEM_KEY_2 9595
#define BUF_SIZE 1024
#include <string.h>
union semnum
{
    int     val;
    struct semid_ds *buf;
    unsigned short  *array;
};

int sem_create(key_t key,int val)   //創建一個信號量並置處值為val
{
    int semid;
    semid=semget(key,1,0666|IPC_CREAT);
    if(semid==-1)
    {
        perror("semget");
        exit(-1);
    }
    union semnum arg;  //聯合類型的變量初始化必須用{}，賦值使用 arg.val=0
    arg.val=val;         //設信號量的初始值
    if(semctl(semid,0,SETVAL,arg)==-1)
    {
        perror("semctl");
        exit(-1);
    }
    return semid;
} 
void sem_del(int semid)   //刪除信號量
{
    union semnum arg;
    arg.val=0;
    if(semctl(semid,0,IPC_RMID,arg)==-1)
    {
        perror("semctl");
        exit(-1);
    }
}

int P(int semid)  //P操作，使信號量的值減1
{
    struct sembuf sops={0,-1,SEM_UNDO};//第三個參數設置為SEM_UNDO時當信號量小於0時會阻塞，設置為IPC_NOWAIT則不會阻塞
    return (semop(semid,&sops,1));
    
}

int V(int semid)  //V操作，使信號量的值加一
{
    struct sembuf sops={0,+1,SEM_UNDO};
    return (semop(semid,&sops,1));
}

struct msg_data         //定義一個共享內存存儲的消息結構體
{
    char data[BUF_SIZE];
};

void Productor(struct msg_data *msg,int i)  //生產者
{
    const char * str="productid:";
    char *array[]={"1","2","3","4","5"};
    strcpy(msg->data,str);
    strcat(msg->data,array[i]);
    printf("Productor:%s\n",msg->data);
}

void Customer(struct msg_data *msg)     //消費者
{
    printf("Customer:%s\n",msg->data);
}
#endif

//allsemshm.c   創建了2個信號量用於實現生產者和消費者之間的同步問題 ，並創建了一個共享內存作為共享資源
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#include <sys/shm.h>
#include "semshm.h"
int main()
{
    int empty,full,shmid;
    empty=sem_create(SEM_KEY_1,1);//設置一個信號量置初值為1
    full=sem_create(SEM_KEY_2,0); //設置一個信號量置初值為0
    shmid=shmget(SHM_KEY,sizeof(struct msg_data),0666|IPC_CREAT);
    if(shmid==-1)
    {
        perror("shmget");
        exit(-1);
    }

    printf("empty=%d\tfull=%d\tshmid=%d\n",empty,full,shmid);
    return 0;
}

//semshm_s.c  生產者代碼
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#include <sys/shm.h>
#include "semshm.h"

int main()
{
    int empty,full,shmid;
    empty=semget(SEM_KEY_1,1,0);    //獲得信號量
    full=semget(SEM_KEY_2,1,0);
    shmid=shmget(SHM_KEY,0,0);      //獲得共享內存
    if(empty==-1||full==-1||shmid==-1)
    {
        perror("get");
        exit(-1);
    }

    struct msg_data *buf;
    void * tmp=shmat(shmid,NULL,0);  //映射共享內存
    buf=(struct msg_data*)tmp;
    int i=0;
    for(i=0;i<5;i++)
    {
        sleep(15);
        P(empty);      
        Productor(buf,i);　　　　//生產者向共享內存寫入數據
        V(full);
    }
    return 0;
}

//semshm_c.c  消費者代碼 
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#include <sys/shm.h>
#include "semshm.h"


int main()
{
    int empty,full,shmid;
    empty=semget(SEM_KEY_1,1,0);   //獲取信號量
    full=semget(SEM_KEY_2,1,0);
    shmid=shmget(SHM_KEY,0,0);     //獲取共享內粗
    if(empty==-1||full==-1||shmid==-1)
    {
        perror("get");
        exit(-1);
    }

    struct msg_data *buf;
    buf=(struct msg_data*)shmat(shmid,NULL,0);   //映射共享內存地址
    int i=0;
    for(i=0;i<5;i++)
    {
        P(full);
        Customer(buf);           //消費者從共享內存取數據
        V(empty);
    }

    
    return 0;
}

 

//delsemshm.c  //刪除所建的信號量和共享內存    
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#include <sys/shm.h>
#include "semshm.h"
int main()
{
    int empty,full,shmid;
    empty=semget(SEM_KEY_1,1,0);
    full=semget(SEM_KEY_2,1,0);
    shmid=shmget(SHM_KEY,0,0);
    if(empty==-1||full==-1||shmid==-1)
    {
        perror("del get");
        exit(-1);
    }
    sem_del(empty);
    sem_del(full);
    struct shmid_ds *ds=(struct shmid_ds*)malloc(sizeof(struct shmid_ds));
    shmctl(shmid,IPC_RMID,ds);
    printf("del success !\n");
    return 0;
}

運行：我們先運行allsemshm這個程序，建立信號量和共享內存，再運行semshm_s這個生產者程序（會sleep15秒），同時，運行2個semshm_c消費者程序，觀察程序的輸出，再調用delsemshm程序刪除信號量和共享內存

運行截圖

分析這段代碼，首先創建了empty和full2個信號量用來實現進程同步問題（這個生產者消費者只有一個緩沖去可以不考慮互斥問題，對於多個緩沖區的生產者消費者問題需要使用3個信號量完成同步和互斥操作），並創建了一個共享內存做為共享資源，接着生產者會生產5次，只有緩沖區內容被取走時才能進行生產操作，同時開啟了2個消費者程序。只有緩沖區有內容時，消費者才能執行他的動作。對於生產者消費者問題的信號量操作可以參考操作系統教程中相關內容。

消息隊列

消息隊列是存在於內核中的消息列表，一個進程可以將消息發送到消息列表。另一個進程可以從消息列表接受消息。消息隊列的操作方式為先進先出

相關API

#include <sys/types.h>

#include <sys/ipc.h>

#include <sys/msg.h>

int msgget(key_t key,int msgflg)

函數說明:功能:獲取或者創建一個消息隊列，key值同上，msgflg：存取或者創建條件值同上。成功返回消息隊列標識，失敗返回-1.

int msgsnd(int msgid,const void* msgp,size_t msgsz,int msgflg)

函數說明:功能:向消息隊列中發送消息，msgid:消息隊列標識，msgp消息結構體的地址，msgsz:消息結構體的字節，msgflg:操作標志，成功返回0,否則返回-1。在消息隊列沒有足夠的空間容納發送的消息時，該函數會阻塞，如果msgflg為IPC_NOWAIT ，則不管發送消息是否成功，該函數都不會阻塞。其中msgp必須指向這樣一個結構體

struct msgbuf{

long mtype;   //必須有且大於0

char mtext[1];  //這個可以自己定以，也可以定義其他成員

}

size_t msgrcv(int msgid,void *msgp,size_t msgsz,long msgtyp,int msgflg)

函數說明:獲取指定消息隊列中，msgtyp類型的消息，該值要根發送消息的結構體中msgp->mtype值一樣，msgsz，消息結構體的大小，msgflg操作標志，值同上

成功返回收到的字節個數，失敗返回-1

int msgctl(int msqid,int cmd,struct msqid_ds* buf)

函數說明:cmd操作類型，IPC_RMID刪除消息隊列，IPC_STAT獲取消息隊列的狀態，IPC_SET改變消息隊列的狀態，buf用來存放消息隊列的屬性信息，其結構體如下

struct msqid_ms{

struct ipc_perm msg_perm;   //權限

struct msg *msg_first;　　　//消息隊列的首

struct msg *msg_last;  　　 //消息隊列的尾

__kernel_time_t msg_stime;   //最后發送時間

__kernel_time_t msg_rtime;  //最后接受時間

__kernel_time_t msg_ctime; //最后修改時間

unsigned short msg_cbytes;    //當前消息隊列的字節數

unsigned short msg_qnum;  //消息隊列中的消息數

unsigned short msg_qbytes; //消息隊列的最大字節數

__kernel_ipc_pid_t  msg_lspid; //最后發送消息的進程ID

__kernel_ipc_pid_t  msg_lrpid; //最后接受消息的進程ID

};

演示代碼

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <unistd.h>
#define MSG_KEY 111
#define BUFSIZE 4096
struct msgbuf
{
    long mtype;
    char mtext[BUFSIZE];
};
int main()
{
    int mspid;
    pid_t pid;

    mspid=msgget(MSG_KEY,IPC_CREAT|0666);
    if(mspid==-1)
    {
        perror("msgget");
        exit(-1);
    }

    pid=fork();
    if(pid<0)
    {
        perror("fork");
        exit(-1);
    }
    else if(pid==0)
    {
        sleep(10);
        int msqid=msgget(MSG_KEY,0);
        if(msqid==-1)
        {
            perror("msgget");
            exit(-1);
        }
        struct msgbuf buf;
        if(msgrcv(msqid,(void *)&buf,sizeof(struct msgbuf),1,0)==-1)
        {
            perror("msgrcv");
            exit(-1);
        }
        printf("child:rcv a msg is %s\n",buf.mtext);
        exit(0);
    }
    else
    {
        struct msgbuf buf;
        buf.mtype=1;
        strcpy(buf.mtext,"Hello World!");
        if(msgsnd(mspid,(const void *)&buf,sizeof(struct msgbuf),0)==-1)
        {
            perror("msgsnd");
            exit(-1);
        }
        printf("parent:snd a msg is %s\n",buf.mtext);
        sleep(10);
    }

   // struct msqid_ds cbuf;
    msgctl(mspid,IPC_RMID,NULL);
    return 0;
}

總結：以上就是一些常用的進程通信的方法了，關於一些具體的API和一些結構體，以及宏所代表的意義，如果不是很清楚可以在Linux下man相關函數，Linux多進程編程就到這里了，接下來就是多線程編程相關的知識了。