首页 › 分类存档 › 进程间通信

(C语言)多进程实现消费者生产者pv操作(Windows和Linux版本)

第七根弦 2019 年 10 月 12 日 Visual Studio, 未分类, 进程间通信 0

多进程实现消费者生产者问题

一，实验目的
1，了解生产者消费者的互斥与同步问题
2，掌握Windows和Linux的进程通信方法

二，实验要求
完成Windows版本和Linux版本。
一个大小为3的缓冲区，初始为空。
2个生产者
随机等待一段时间，往缓冲区添加数据，
若缓冲区已满，等待消费者取走数据后再添加
重复6次

3个消费者
随机等待一段时间，从缓冲区读取数据
若缓冲区为空，等待生产者添加数据后再读取
重复4次

显示每次添加和读取数据的时间及缓冲区的状态

三，实验环境
Windows版本：Windows 10 64位系统，Dev-cpp编译器
Linux版本：Fedora29版本，gcc环境 vim文本编辑器

四，实验代码结构
1),pv操作伪代码：
array[3]:interger//缓冲区定义,大小为三
int empty=3,full=0;
int mutex=1;
i=0,j=0//缓冲区指针
x，y：item //产品变量
生产者： 消费者：
begin：
produce a product to x;
P(empty);
P(mutex);
array[i]=x;
ii=(i+1)%3;
V(full);
V(mutex);
,,,,,, ………,
End

消费者：

2）实验代码分析

Windows版本：

思路分析：Windows创建多进程使用creatprocess()函数调用自己，通过多次创建得到两个生产者进程三个消费者进程，在之中运行相应的生产者函数，消费者函数。在通过传入参数不同，来辨别是第一次主进程还是生产者进程，消费者进程。通过构建共享内存区进行进程间通信。

①多进程创建

② 构建共享内存区，再将文件映射到本进程，初始化

③在主创建进程间信号量full empty

分别在生产者消费者进程创建互斥访问量mutex

④同过argv量的不同判断进程归属

⑤运行结果：

全部代码见后

Linux版本：

思路分析：Linux使用fork进行多进程创建，分别在进程中运行消费者函数，生产者函数。建立共享主存区很信号量在进程建进行通信和缓存访问

②建立共享主存并进行映射

③创建进程间信号量full，empty和互斥量 mutex，并初始化

④实验结果：

五， 实验总结

本次实验获得圆满成功。

本次实验通过分别编写Windows和Linux版本的多进程实现消费者和生产者问题，了解生产者消费者的互斥与同步问题，掌握Windows和Linux的进程通信方法，也同时加强自己对多进程操作的理解。

代码：
Windows版本：

C++
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
//实验三生产者消费者
#include <windows.h>
#include <stdio.h>
#include <time.h>
 
 
HANDLE handleOfProcess[5];
 
struct buf
{
    char buffer[3];
    int write;
    int read;
};
 
int rand_1()
{
    return rand() % 100 + 1000;
}
 
char rand_char()
{
    return rand() % 26 + 'A';
}
 
void StartClone(int nCloneID)
{
    TCHAR szFilename[MAX_PATH];
    GetModuleFileName(NULL, szFilename, MAX_PATH);
 
    TCHAR szCmdLine[MAX_PATH];
    sprintf(szCmdLine, "\"%s\" %d", szFilename, nCloneID);
    //printf("%s\n",szCmdLine);
 
    STARTUPINFO si;
    ZeroMemory(reinterpret_cast<void*>(&si), sizeof(si));
 
    si.cb = sizeof(si);
    PROCESS_INFORMATION pi;
 
    BOOL bCreateOK = CreateProcess(
        szFilename,
        szCmdLine,
        NULL,
        NULL,
        FALSE,
        CREATE_DEFAULT_ERROR_MODE,
        NULL,
        NULL,
        &si,
        &pi);
    if (bCreateOK)
        handleOfProcess[nCloneID] = pi.hProcess;
    else
    {
        printf("Error in create process!\n");
        exit(0);
    }
}
 
void pro()//生产者
{
    HANDLE mutex = CreateMutex(NULL, FALSE, "MYMUTEX");
    HANDLE empty = OpenSemaphore(SEMAPHORE_ALL_ACCESS, FALSE, "MYEMPTY");
    HANDLE full = OpenSemaphore(SEMAPHORE_ALL_ACCESS, FALSE, "MYFULL");
 
    HANDLE hMap = OpenFileMapping(FILE_MAP_ALL_ACCESS, FALSE, "myfilemap");
    LPVOID Data = MapViewOfFile(//文件映射
        hMap,
        FILE_MAP_ALL_ACCESS,
        0,
        0,
        0);
    struct buf* pint = reinterpret_cast<struct buf*>(Data);
 
    for (int i = 0; i < 6; i++)
    {
        WaitForSingleObject(empty, INFINITE);
        //sleep
        srand((unsigned)time(0));
        int tim = rand_1();
        Sleep(tim);
 
        WaitForSingleObject(mutex, INFINITE);
 
        //code
        pint->buffer[pint->write] = rand_char();
        pint->write = (pint->write + 1) % 3;
 
 
        ReleaseMutex(mutex);
        ReleaseSemaphore(full, 1, NULL);
 
        SYSTEMTIME syst;
        time_t t = time(0);
        GetSystemTime(&syst);
        char tmpBuf[10];
        strftime(tmpBuf, 10, "%H:%M:%S", localtime(&t));
        printf("生产者%d向缓冲区写入数据:\t%c\t%c\t%c\t@%s.%d\n",
            (int)GetCurrentProcessId(), pint->buffer[0], pint->buffer[1],
            pint->buffer[2], tmpBuf, syst.wMilliseconds);
        fflush(stdout);
    }
    UnmapViewOfFile(Data);//解除映射
    Data = NULL;
    CloseHandle(mutex);
    CloseHandle(empty);
    CloseHandle(full);
}
 
void con()//消费者
{
    HANDLE mutex = CreateMutex(NULL, FALSE, "MYMUTEX");
    HANDLE empty = OpenSemaphore(SEMAPHORE_ALL_ACCESS, FALSE, "MYEMPTY");
    HANDLE full = OpenSemaphore(SEMAPHORE_ALL_ACCESS, FALSE, "MYFULL");
 
    HANDLE hMap = OpenFileMapping(FILE_MAP_ALL_ACCESS, FALSE, "myfilemap");
    LPVOID Data = MapViewOfFile(//文件映射
        hMap,
        FILE_MAP_ALL_ACCESS,
        0,
        0,
        0);
    struct buf* pint = reinterpret_cast<struct buf*>(Data);
 
    for (int i = 0; i < 4; i++)
    {
        WaitForSingleObject(full, INFINITE);
        //sleep
        srand((unsigned)time(0));
        int tim = rand_1();
        Sleep(tim);
 
        WaitForSingleObject(mutex, INFINITE);
 
        pint->buffer[pint->read] = ' ';
        pint->read = (pint->read + 1) % 3;
 
        ReleaseMutex(mutex);
        ReleaseSemaphore(empty, 1, NULL);
        //code
 
        time_t t = time(0);
        char tmpBuf[10];
        SYSTEMTIME syst;
        GetSystemTime(&syst);
        strftime(tmpBuf, 10, "%H:%M:%S", localtime(&t));
 
        printf("消费者%d从缓冲区读取数据:\t%c\t%c\t%c\t@%s.%d\n",
            (int)GetCurrentProcessId(), pint->buffer[0], pint->buffer[1],
            pint->buffer[2], tmpBuf, syst.wMilliseconds);
        fflush(stdout);
 
    }
    UnmapViewOfFile(Data);//解除映射
    Data = NULL;
    CloseHandle(mutex);
    CloseHandle(empty);
    CloseHandle(full);
}
 
int main(int argc, char* argv[])
{
    int nCloneID = 20;
    if (argc > 1)
    {
        sscanf(argv[1], "%d", &nCloneID);
    }
 
    if (nCloneID < 2)//生产者进程
    {
        pro();
    }
    else if (nCloneID < 5)//消费者进程
    {
        con();
    }
    else//主进程
    {
        HANDLE hMap = CreateFileMapping(
            NULL,
            NULL,
            PAGE_READWRITE,
            0,
            sizeof(struct buf),
            "myfilemap");
 
        if (hMap != INVALID_HANDLE_VALUE)
        {
            LPVOID Data = MapViewOfFile(//文件映射
                hMap,
                FILE_MAP_ALL_ACCESS,
                0,
                0,
                0);
 
            if (Data != NULL)
            {
                ZeroMemory(Data, sizeof(struct buf));
            }
            struct buf* pnData = reinterpret_cast<struct buf*>(Data);
            pnData->read = 0;
            pnData->write = 0;
            memset(pnData->buffer, 0, sizeof(pnData->buffer));
            UnmapViewOfFile(Data);//解除映射
            Data = NULL;
        }
 
        HANDLE empty = CreateSemaphore(NULL, 3, 3, "MYEMPTY");
        HANDLE full = CreateSemaphore(NULL, 0, 3, "MYFULL");
        for (int i = 0; i < 5; i++)//创建子进程
            StartClone(i);
        WaitForMultipleObjects(5, handleOfProcess, TRUE, INFINITE);
        CloseHandle(empty);
        CloseHandle(full);
    }
}

Linux版本:

C++
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
//实验三：生产者消费者                                                                                                                                                                                            
#include<string.h>
#include <sys/time.h>
#include<stdio.h>
#include<sys/types.h>
#include<unistd.h>
#include<stdlib.h>
#include<sys/sem.h>
#include<sys/select.h>
#include<sys/wait.h>
#include<sys/ipc.h>
#include<sys/shm.h>
#include<time.h>
#define SEM_ID1 225
#define SEM_ID2 97
#define SEM_ID3 234
#define SHMKEY 75
 
struct buf {
    char buffer[3];
    int read;
    int write;
};
 
int rand_1()
{
    return rand() % 300;
}
 
void sleep_ms(int s)
{
    usleep(s * 10000);
}
 
char* cur_time()
{
    time_t timep;
    time(&timep);
    return ctime(&timep);
}
char rand_char()
{
    return rand() % 26 + 'A';
}
void P(int s)//p操作
{
    struct sembuf sem_op;
    sem_op.sem_num = 0;
    sem_op.sem_op = -1;
    sem_op.sem_flg = 0;
    semop(s, &sem_op, 1);
}
void V(int s)//v操作
{
    struct sembuf sem_op;
    sem_op.sem_num = 0;
    sem_op.sem_op = 1;
    sem_op.sem_flg = 0;
    semop(s, &sem_op, 1);
 
}
void pro()//生产者
{
    int tim, shmid, i = 6;
    int sem_mutex, sem_empty, sem_full;
    void* addr;
    struct buf* pint;
    struct sembuf sem_op;
    struct timeval tv;
    sem_mutex = semget(SEM_ID1, 1, 0600);
    sem_empty = semget(SEM_ID2, 1, 0600);
    sem_full = semget(SEM_ID3, 1, 0600);
    shmid = shmget(SHMKEY, sizeof(struct buf), 0777);
    addr = shmat(shmid, 0, 0);
    while (i--)
    {
        gettimeofday(&tv, NULL);
        srand((unsigned)tv.tv_usec);
        tim = rand_1();
        sleep_ms(tim);
        //P(empty)
        P(sem_empty);
        //P(mutex)
        P(sem_mutex);
 
        pint = (struct buf*)addr;
 
        // pint[semctl(sem_full,0,GETVAL)]=time;
        pint->buffer[pint->write] = rand_char();
        pint->write = (pint->write + 1) % 3;
 
        printf("当前生产者进程:%d 写入数据:\t%c\t%c\t%c\t@%lds%ldus\n",
            getpid(), pint->buffer[0], pint->buffer[1],
            pint->buffer[2], tv.tv_sec, tv.tv_usec);
 
 
        //V(mutex)
        V(sem_mutex);
        //V(full)
        V(sem_full);
    }
    shmdt(addr);
}
void con()//消费者
{
    int tim, shmid, i = 4;
    int sem_mutex, sem_empty, sem_full;
    void* addr;
    struct buf* pint;
    struct sembuf sem_op;
    struct timeval tv;
    sem_mutex = semget(SEM_ID1, 1, 0600);
    sem_empty = semget(SEM_ID2, 1, 0600);
    sem_full = semget(SEM_ID3, 1, 0600);
    shmid = shmget(SHMKEY, sizeof(struct buf), 0777);
    addr = shmat(shmid, 0, 0);
    while (i--)
    {
        gettimeofday(&tv, NULL);
        srand((unsigned)tv.tv_usec);
        tim = rand_1();
        sleep_ms(tim);
        //P(full)
        P(sem_full);
        //P(mutex)
        P(sem_mutex);
 
        pint = (struct buf*)addr;
 
        pint->buffer[pint->read] = ' ';
        pint->read = (pint->read + 1) % 3;
 
        printf("当前消费者进程:%d 读取数据:\t%c\t%c\t%c\t@%ldst%ldms\n",
            getpid(), pint->buffer[0], pint->buffer[1],
            pint->buffer[2], tv.tv_sec, tv.tv_usec);
 
        //V(mutex)
        V(sem_mutex);
        //V(empty)
        V(sem_empty);
 
    }
    shmdt(addr);
}
 
int main()
{
    int sem_mutex, sem_empty, sem_full, shmid;
    void* addr;
    union semun {
        int val;
    }empty, full, mutex;
 
    //建立信号量
    sem_mutex = semget(SEM_ID1, 1, IPC_CREAT | 0600);
    sem_empty = semget(SEM_ID2, 1, IPC_CREAT | 0600);
    sem_full = semget(SEM_ID3, 1, IPC_CREAT | 0600);
 
    full.val = 0;
    empty.val = 3;
    mutex.val = 1;
 
    semctl(sem_mutex, 0, SETVAL, mutex);
    semctl(sem_empty, 0, SETVAL, empty);
    semctl(sem_full, 0, SETVAL, full);
 
    //建立共享内存并进行映射
    shmid = shmget(SHMKEY, sizeof(struct buf), 0777 | IPC_CREAT);
    if (-1 == shmid)
    {
        printf("建立共享内存失败\n");
        exit(0);
    }
    addr = shmat(shmid, 0, 0);
    memset(addr, 0, sizeof(struct buf));
 
    //执行生产者进程
    for (int i = 0; i < 2; i++)
        if (fork() == 0)
        {
            pro();
            exit(0);
        }
    //执行消费者进程
    for (int i = 0; i < 3; i++)
        if (fork() == 0)
        {
            con();
            exit(0);
        }
 
    while (-1 != wait(0));
    semctl(sem_mutex, 0, IPC_RMID);
    semctl(sem_empty, 0, IPC_RMID);
    semctl(sem_full, 0, IPC_RMID);
    shmdt(addr);
}

编程思想之多线程与多进程——以操作系统的角度述说线程与进程

第七根弦 2015 年 12 月 31 日 线程间通信, 进程间通信 0

什么是线程

什么是线程？线程与进程与有什么关系？这是一个非常抽象的问题，也是一个特别广的话题，涉及到非常多的知识。我不能确保能把它讲的话，也不能确保讲的内容全部都正确。即使这样，我也希望尽可能地把他讲通俗一点，讲的明白一点，因为这是个一直困扰我很久的，扑朔迷离的知识领域，希望通过我的理解揭开它一层一层神秘的面纱。
继续阅读 »

OS: 生产者消费者问题(多进程+共享内存+信号量)

第七根弦 2015 年 10 月 27 日 UC, 进程间通信 0

一. 引子

时隔一年再次用到 cout 的时候，哥潸然泪下，这是一种久别重逢的感动，虽然基本忘光了。趁着有大把时间，再把生产者消费者问题巩固一下，用纯C吧。珍惜能写代码的幸福时光。

二. 分析

生产者和消费者问题是多个相互合作的进程之间的一种抽象。生产者和消费者之间的关系:
1.  对缓冲区的访问是互斥的。由于两者都会修改缓冲区，因此，一方修改缓冲区时，另一方不能修改，这就是互斥。
2.  一方的行为影响另一方。缓冲区不空，才能消费，何时不空？生产了就不空；缓冲区满，就不能生产，何时不满？消费了就不满。这是同步关系。
为了描述这种关系，一方面，使用共享内存代表缓冲区；另一方面，使用 互斥信号量 控制对缓冲区的访问，使用同步信号量描述两者的依赖关系。
继续阅读 »

linux 本地通信实例 AF_UNIX

第七根弦 2015 年 3 月 26 日 UC, 进程间通信 0

程序说明：
程序里包含服务端和客户端两个程序，它们之间使用 AF_UNIX 实现本机数据流通信。使用 AF_UNIX 域实际上是使用本地 socket 文件来通信。
继续阅读 »

网络服务端开发小结(短连接、长连接、进程池、线程池)

第七根弦 2014 年 12 月 17 日 线程间通信, 进程间通信 0

平时对网络编程方面比较感兴趣，看了一些相关的资料，自己也尝试写过一些不同网络模型的服务程序。这次刚好有一个新的需求，需要开发一个转发服务器。之前开发的项目，网络通讯都是处理联机交易的，网络连接都是采用短连接，这次的服务端，采用长连接的方式。
继续阅读 »

一个进程池的服务器程序

第七根弦 2014 年 12 月 16 日 进程间通信 0

下面做了非常简单的http服务器，该服务器只能接收Get请求。
流程大概如下：
1，父进程listen，创建pipe（下面所有父子进程之间的通信都用该pipe）
2，父进程预fork n个子进程
3，各个子进程accept(listenfd)，即所有子进程竞争accept请求。由于listenfd是在fork之前就有的，所以所有子进程都可以访问到，不需用到“进程间文件描述符传递”问题；
4，子进程每accept到一个请求都告诉父进程，父进程把请求数加1；子进程没完成一个请求，父进程把请求数减1；当父进程发现请求数 >= 子进程数时，父进程创建新的子进程，并把子进程数加1（当然子进程数有个预先上限）；当父进程发现子进程数大于请求数加1时，父进程杀死多余的子进程。
总的来说，思想是让子进程accept并处理请求，父进程通过子进程发来的信息控制请求数与子进程数之间的关系。
继续阅读 »

linux下C 线程池的原理讲解和代码实现（能自行伸缩扩展线程数）

第七根弦 2014 年 11 月 24 日 UC, 进程间通信 0

什么线程池，为什么要使用线程池？下面是一个比喻。

阶段一、一个医院，每天面对成千上万的病人，处理方式是：来一个病人找来一个医生处理，处理完了医生也走了。当看病时间较短的时候，医生来去的时间，显得尤为费时了。

阶段二、医院引进了线程池的概念。设置门诊，把医生全派出去坐诊，病人来看病先挂号排队，医生根据病人队列顺序依次处理各个病人，这样就省去医生来来去去的时间了。但是，很多时候病人不多，医生却很多导致很多医生空闲浪费水电资源撒。

阶段三、医院引进了可伸缩性线程池的概念，如阶段二，但是门诊一开始只派出了部分医生，但是增加了一个领导，病人依旧是排队看病，领导负责协调整个医院的医生。当病人很多医生忙不过来的时候，领导就去多叫几个医生来帮忙；当病人不多医生太多的时候，领导就叫一些医生回家休息去免得浪费医院资源。
继续阅读 »

简单Linux C线程池

第七根弦 2014 年 11 月 24 日 UC, 进程间通信 0

大多数的网络服务器，包括Web服务器都具有一个特点，就是单位时间内必须处理数目巨大的连接请求，但是处理时间却是比较短的。在传统的多线程服务器模型中是这样实现的：一旦有个请求到达，就创建一个新的线程，由该线程执行任务，任务执行完毕之后，线程就退出。这就是”即时创建，即时销毁”的策略。尽管与创建进程相比，创建线程的时间已经大大的缩短，但是如果提交给线程的任务是执行时间较短，而且执行次数非常频繁，那么服务器就将处于一个不停的创建线程和销毁线程的状态。这笔开销是不可忽略的，尤其是线程执行的时间非常非常短的情况。
继续阅读 »

Linux下设计一个简单的线程池

第七根弦 2014 年 11 月 24 日 UC, 进程间通信 0

定义

什么是线程池？简单点说，线程池就是有一堆已经创建好了的线程，初始它们都处于空闲等待状态，当有新的任务需要处理的时候，就从这个池子里面取一个空闲等待的线程来处理该任务，当处理完成了就再次把该线程放回池中，以供后面的任务使用。当池子里的线程全都处理忙碌状态时，线程池中没有可用的空闲等待线程，此时，根据需要选择创建一个新的线程并置入池中，或者通知任务线程池忙，稍后再试。
继续阅读 »

Linux进程间通信——使用数据报套接字

第七根弦 2014 年 10 月 9 日 UC, 进程间通信 0
前一篇文章，Linux进程间通信——使用流套接字介绍了一些有关socket（套接字）的一些基本内容，并讲解了流套接字的使用，这篇文章将会给大家讲讲，数据报套接字的使用。

继续阅读 »