在Linux操作系统下用C实现经典同步问题：生产者—消费者，具体要求如下：
(1) 一个大小为10的缓冲区，初始状态为空。
(2) 2个生产者，随机等待一段时间，往缓冲区中添加数据，若缓冲区已满，等待消费者取走数据之后再添加，重复10次。
(3) 2个消费者，随机等待一段时间，从缓冲区中读取数据，若缓冲区为空，等待生产者添加数据之后再读取，重复10次。
本实验的主要目的是模拟操作系统中进程同步和互斥。在系统进程并发执行异步推进的过程中，由于资源共享和进程间合作而造成进程间相互制约。进程间的相互制约有两种不同的方式。
（1）间接制约。这是由于多个进程共享同一资源（如CPU、共享输入/输出设备）而引起的，即共享资源的多个进程因系统协调使用资源而相互制约。
（2）直接制约。只是由于进程合作中各个进程为完成同一任务而造成的，即并发进程各自的执行结果互为对方的执行条件，从而限制各个进程的执行速度。
生产者和消费者是经典的进程同步问题，在这个问题中，生产者不断的向缓冲区中写入数据，而消费者则从缓冲区中读取数据。生产者进程和消费者对缓冲区的操作是互斥，即当前只能有一个进程对这个缓冲区进行操作，生产者进入操作缓冲区之前，先要看缓冲区是否已满，如果缓冲区已满，则它必须等待消费者进程将数据取出才能写入数据，同样的，消费者进程从缓冲区读取数据之前，也要判断缓冲区是否为空，如果为空，则必须等待生产者进程写入数据才能读取数据。
在本实验中，进程之间要进行通信来操作同一缓冲区。一般来说，进程间的通信根据通信内容可以划分为两种：即控制信息的传送与大批量数据传送。有时，也把进程间控制信息的交换称为低级通信，而把进程间大批量数据的交换称为高级通信。目前，计算机系统中用得比较普遍的高级通信机制可分为3大类：共享存储器系统、消息传递系统及管道通信系统。
在这个问题当中，我们采用信号量机制进行进程之间的通信，设置两个信号量，空的信号量和满的信号量。
在Linux系统中，一个或多个信号量构成一个信号量集合。使用信号量机制可以实现进程之间的同步和互斥，允许并发进程一次对一组信号量进行相同或不同的操作。每个P、V操作不限于减1或加1，而是可以加减任何整数。在进程终止时，系统可根据需要自动消除所有被进程操作过的信号量的影响。
shmget()系统调用实现共享主存段的创建，shmget()返回共享内存区的ID。对于已经申请到的共享段，进程需把它附加到自己的虚拟空间中才能对其进行读写。semget()函数，同时建立信号量的数量。在信号量建立后，调用semctl()对信号量进行初始化，例如本实验中，可以建立两个信号SEM_EMPTY、SEM_FULL，初始化时设置SEM_EMPTY为10，SEM_FULL为0。使用操作信号的函数semop()做排除式操作，使用这个函数防止对共享内存的同时操作。对共享内存操作完毕后采用shmctl()函数撤销共享内存段。fflush()刷新缓冲区。semctl()函数释放内存。主程序流程图：
生产者进程流程图
消费者进程流程图
P操作流程图
V操作流程图
自定义结构体buf：实现循环队列
struct buf //循环队列
{
int in; //指向缓冲中下一个空位
int out; //指向缓冲中第一个满位
char str[MAX_BUFFER_SIZE]; //存放字母
int num; //产品个数
int is_empty; //是否为空
};
struct buf *bf; //定义一个循环队列指针
库<sys/time.h>中的结构体timeval：实现时间
/*
struct timeval
{
time_t tv_sec; //seconds
suseconds_t tv_usec; //icroseconds
};
*/
struct timeval start; //开始时间
struct timeval now; //当前时间
库<sys/sem.h>中的结构体sembuf：对信号量操作
/*
struct sembuf
{
unsigned short int sem_num; //信号量编号
short int sem_op; //对信号量的操作
short int sem_flg; //通常设为SEM_UNDO
}
*/
struct sembuf sb;
整型
//缓冲区大小
#define MAX_BUFFER_SIZE 10
//共享内存和信号量：当前用户的进程可读写
#define SHM_MODE 0600
#define SEM_MODE 0600
//信号量的键值
#define SEM_FULL 1 //信号量 1 --- full
#define SEM_EMPTY 2 //信号量 2 --- empty
#define MUTEX 3 //信号量 3 --- 互斥量
//共享内存的键值
#define SHMKEY 75
const int N_Consumer = 2; //消费者数量
const int N_Producer = 2; //生产者数量
const int N_Buffer = 10; //缓冲区大小
const int N_Worktime = 10; //每个进程执行的次数
int shm_id; //shmget函数返回的共享内存标识符
int full, empty, mutex; //semget函数返回的信号量标识符
void指针
void *addr; //地址指针，连接共享内存段的地址空间
自定义函数（实现详见下方代码）：
int get_rand_num() //0~99 随机数
char get_rand_letter() //随机生成A~Z
void Wait(int sem_id) //P操作
void Signal(int sem_id) //V操作
void printTime() //输出时间
void Producer(struct buf *bf) //生产者操作
void Consumer(struct buf *bf) //消费者操作
各类库中的关键函数
<sys/shm.h>中：
shmget()函数：用来创建共享内存。
int shmget(key_t key, size_t size, int shmflg);
shmat()函数：启动对该共享内存的访问，并把共享内存连接到当前进程的地址空间。
void *shmat(int shm_id, const void *shm_addr, int shmflg);
shmdt()函数：用于将共享内存从当前进程中分离。
int shmdt(const void *shmaddr);
shmctl()函数：用来控制共享内存。
int shmctl(int shm_id, int command, struct shmid_ds *buf);
<sys/sem.h>中：
semget()函数：创建一个新的信号量集
int semget(key_t key,int nsems,int semflg);
semctl()函数：用来控制信号量
int semctl(int semid,int semnum,int cmd,union semun arg);
第一个参数：sem_id是semget()函数返回的信号量标识符。
第二个参数：semnum是信号量编号，当要用到信号量集时才用到。单个信号量则为0。
第三个参数：cmd是要采取的操作，它可以取下面的几个值:
IPC_STAT（或STAT）：获得该信号量（或信号量集）的semid_ds结构（描述信号量的结构），并存放在第四个参数arg结构变量的buf域指向的semid_ds结构中。
IPC_SETVAL（或SETVAL）：将信号量值设置为arg的值。
IPC_GETVAL （或GETVAL）：返回信号量的当前值。
IPC_RMID：删除信号量。
调用成功时根据cmd值的不同返回不同值，失败时返回-1。
semop()函数：对信号量进行操作
int semop(int semid,struct sembuf*sops,size_t nsops );
第一个参数：sem_id是semget()函数返回的信号量标识符。
第二个参数：sops指向信号量操作结构体。
struct sembuf
{
unsigned short int sem_num; //信号量编号
short int sem_op; //对信号量的操作(-1:P +1:V)
short int sem_flg; //通常设为SEM_UNDO
}
第三个参数：nsop是sops中的操作个数，通常为1，代表单个操作。
调用成功时返回信号量标识符，失败时返回-1。
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <time.h>
#include <sys/shm.h>
#include <sys/sem.h>
#include <sys/time.h>
#include <sys/wait.h>
#include <string.h>
//缓冲区大小
#define MAX_BUFFER_SIZE 10
//共享内存和信号量：当前用户的进程可读写
#define SHM_MODE 0600
#define SEM_MODE 0600
//信号量的键值
#define SEM_FULL 1 //信号量 1 --- full
#define SEM_EMPTY 2 //信号量 2 --- empty
#define MUTEX 3 //信号量 3 --- 互斥量
//共享内存的键值
#define SHMKEY 75
struct buf //循环队列
{
int in; //指向缓冲中下一个空位
int out; //指向缓冲中第一个满位
char str[MAX_BUFFER_SIZE]; //存放字母
int num; //产品个数
int is_empty; //是否为空
};
const int N_Consumer = 2; //消费者数量
const int N_Producer = 2; //生产者数量
const int N_Buffer = 10; //缓冲区大小
const int N_Worktime = 10; //每个进程执行的次数
int shm_id; //shmget函数返回的共享内存标识符
int full, empty, mutex; //semget函数返回的信号量标识符
struct timeval start; //开始时间
struct timeval now; //当前时间
int get_rand_num() //0~99 随机数
{
gettimeofday(&now, NULL);
int digit;
srand((unsigned)(now.tv_usec)); //当前时间（微秒）作为随机数种子
digit = rand() % 100;
return digit;
}
char get_rand_letter() //随机生成A~Z
{
gettimeofday(&now, NULL);
char letter;
srand((unsigned)(now.tv_usec));
letter = (char)((rand() % 26) + 'A');
return letter;
}
/*
//信号量操作结构体(在sys/sem.h中定义)
struct sembuf
{
unsigned short int sem_num; //信号量编号
short int sem_op; //对信号量的操作
short int sem_flg; //通常设为SEM_UNDO
}
*/
void Wait(int sem_id) //P操作
{
struct sembuf sb;
sb.sem_num = 0;
sb.sem_op = -1; //Wait(-1)
sb.sem_flg = SEM_UNDO;
if (semop(sem_id, &sb, 1) < 0) //P操作，判断是否成功
{
perror("P操作失败！\n"); //输出错误信息
exit(1);
}
}
void Signal(int sem_id) //V操作
{
struct sembuf sb;
sb.sem_num = 0;
sb.sem_op = 1; //Signal(+1)
sb.sem_flg = SEM_UNDO;
if (semop(sem_id, &sb, 1) < 0) //V操作，判断是否成功
{
perror("V操作失败！\n");
exit(1);
}
}
void printTime() //输出时间
{
gettimeofday(&now, NULL);
double difference = ((double)now.tv_usec - (double)start.tv_usec) / 1000000;
printf("| %8.4fs | ", difference);
}
//生产者操作
void Producer(struct buf *bf)
{
int i, j;
for (i = 0; i < N_Producer; i++) //使用循环创建进程
{
int pid = fork(); //创建子进程
if (pid < 0)
{
perror("创建进程失败\n");
exit(1);
}
else if (pid == 0)
{
void *addr;
addr = shmat(shm_id, 0, 0);
bf = (struct buf *) addr;
if (bf == (void *) - 1)
{
perror("连接共享内存失败！\n");
exit(1);
}
int count = 0;
for (j = 0; j < N_Worktime; j++) //每个进程操作10次
{
Wait(empty);
Wait(mutex);
//随机等待一段时间(0~99微秒)
usleep(get_rand_num());
//向缓冲区添加数据
char c = get_rand_letter();
bf->str[bf->in] = c;
bf->in = (bf->in + 1) % MAX_BUFFER_SIZE;
bf->is_empty = 0;
bf->num++;
int first, last;
//打印时间
printTime();
//打印缓冲区数据
if (bf->in - 1 >= bf->out)
last = bf->in - 1;
else
last = bf->in - 1 + MAX_BUFFER_SIZE;
for (first = bf->out; bf->num>0&&first <= last; first++)
{
printf("%c ", bf->str[first % MAX_BUFFER_SIZE]);
}
int k;
for (k = bf->num; k <= MAX_BUFFER_SIZE; k++)
{
printf(" ");
}
printf("|");
//打印进程编号和产生数据
printf(" 生产者进程%d | 生产%c |\n", i, c);
//刷新缓冲区
fflush(stdout);
Signal(mutex);
Signal(full);
}
//取消对共享内存段的连接
shmdt(addr);
exit(0);
}
}
}
//消费者操作
void Consumer(struct buf *bf)
{
int i, j;
for (i = 0; i < N_Producer; i++) //使用循环创建进程
{
int pid = fork(); //创建子进程
if (pid < 0)
{
perror("创建进程失败\n");
exit(1);
}
else if (pid == 0)
{
void *addr;
addr = shmat(shm_id, 0, 0);
bf = (struct buf *) addr;
if (bf == (void *) - 1)
{
perror("连接共享内存失败！\n");
exit(1);
}
int count = 0;
for (j = 0; j < N_Worktime; j++) //每个进程操作10次
{
Wait(full);
Wait(mutex);
//随机等待一段时间(0~99微秒)
usleep(get_rand_num());
//从缓冲区取走数据
char c = bf->str[bf->out];
bf->out = (bf->out + 1) % MAX_BUFFER_SIZE;
bf->is_empty = (bf->out == bf->in);
bf->num--;
int first, last;
//打印时间
printTime();
//打印缓冲区数据
if (bf->in - 1 >= bf->out)
last = bf->in - 1;
else
last = bf->in - 1 + MAX_BUFFER_SIZE;
for (first = bf->out; bf->num>0&&first <= last; first++)
{
printf("%c ", bf->str[first % MAX_BUFFER_SIZE]);
}
int k;
for (k = bf->num; k <= MAX_BUFFER_SIZE; k++)
{
printf(" ");
}
printf("|");
//打印进程编号和消费数据
printf(" 消费者进程%d | 消费%c |\n", i, c);
//刷新缓冲区
fflush(stdout);
Signal(mutex);
Signal(empty);
}
//取消对共享内存段的连接
shmdt(addr);
exit(0);
}
}
}
int main()
{
printf("| 进程执行时间 | 缓冲区数据 | 当前执行的进程 | 产生或消费的数据 |\n");
printf("----------------------------------------------------------------------------\n");
gettimeofday(&start, NULL); //获取开始运行的时间（1970年1月1日到现在的秒和微秒）
shm_id = shmget(SHMKEY, MAX_BUFFER_SIZE, IPC_CREAT | SHM_MODE); //创建共享内存
if (shm_id < 0)
{
perror("共享内存创建失败\n");
exit(1);
}
struct buf *bf; //定义一个循环队列指针
//将共享内存段连接到一个进程的地址空间中
void *addr;
addr = shmat(shm_id, 0, 0);
bf = (struct buf *) addr;
if (bf == (void *) -1)
{
perror("连接共享内存失败!\n");
exit(1);
}
//创建3个信号量
if ((full = semget(SEM_FULL, 1, IPC_CREAT | SEM_MODE)) < 0)
{
perror("创建信号量SEM_FULL失败!\n");
exit(1);
}
if ((empty = semget(SEM_EMPTY, 1, IPC_CREAT | SEM_MODE)) < 0)
{
perror("创建信号量SEM_EMPTY失败!\n");
exit(1);
}
if ((mutex = semget(MUTEX, 1, IPC_CREAT | SEM_MODE)) < 0)
{
perror("创建信号量MUTEX失败!\n");
exit(1);
}
//设置3个信号量
if (semctl(full, 0, SETVAL, 0) == -1) //SEM_FULL->0
{
perror("SEM_FULL设置失败!\n");
exit(1);
}
if (semctl(empty, 0, SETVAL, 10) == -1) //SEM_EMPTY->10
{
perror("SEM_EMPTY设置失败!\n");
exit(1);
}
if (semctl(mutex, 0, SETVAL, 1) == -1) //MUTEX->1
{
perror("MUTEX设置失败!\n");
exit(1);
}
//初始化缓冲区
memset(bf->str,'\0',sizeof(bf->str));
bf->out = 0;
bf->in = 0;
bf->is_empty = 1;
bf->num = 0;
Producer(bf);
Consumer(bf);
//等待程序结束
while (wait(0) != -1);
//取消对共享内存的连接
shmdt(addr);
//删除共享内存段
shmctl(shm_id, IPC_RMID, 0);
//撤销信号量
shmctl(full, IPC_RMID, 0);
shmctl(empty, IPC_RMID, 0);
shmctl(mutex, IPC_RMID, 0);
printf("----------------------------------------------------------------------------\n");
printf("程序结束!\n");
fflush(stdout);
exit(0);
return 0;
}
以上是通过进程来实现生产者和消费者问题，考虑如何采用线程来实现。
<pthread.h>中：
pthread_t：声明线程标识符。
pthread_create()函数：用于创建线程。
int pthread_create(pthread_t *tidp,const pthread_attr_t *attr,void *(*start_rtn)(void*),void *arg);
pthread_join用来等待一个线程的结束,线程间同步的操作
int pthread_join(pthread_t thread, void **retval);
<semaphore.h>中：
sem_t：信号量的数据类型。
sem_init()函数：信号量初始化
sem_init(sem_t *sem, int pshared, unsigned int value);
sem_wait()函数：P操作。
int sem_wait(sem_t *sem);
sem_post()函数：V操作
int sem_post(sem_t *sem);
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <semaphore.h>
#include <pthread.h>
#include <sys/time.h>
#include <sys/wait.h>
#include <string.h>
//缓冲区大小
#define MAX_BUFFER_SIZE 10
struct buf //循环队列
{
int in; //指向缓冲中下一个空位
int out; //指向缓冲中第一个满位
char str[MAX_BUFFER_SIZE]; //存放字母
int num; //产品个数
int is_empty; //是否为空
};
struct buf bf; //定义一个循环队列
const int N_Consumer = 2; //消费者数量
const int N_Producer = 2; //生产者数量
const int N_Buffer = 10; //缓冲区大小
const int N_Worktime = 10; //每个进程执行的次数
sem_t full, empty, mutex; //三个信号量
struct timeval start; //开始时间
struct timeval now; //当前时间
int get_rand_num() //0~99 随机数
{
gettimeofday(&now, NULL);
int digit;
srand((unsigned)(now.tv_usec)); //当前时间（微秒）作为随机数种子
digit = rand() % 100;
return digit;
}
char get_rand_letter() //随机生成A~Z
{
gettimeofday(&now, NULL);
char letter;
srand((unsigned)(now.tv_usec));
letter = (char)((rand() % 26) + 'A');
return letter;
}
void Wait(sem_t &sem) //P操作
{
sem_wait(&sem);
}
void Signal(sem_t &sem) //V操作
{
sem_post(&sem);
}
void printTime() //输出时间
{
gettimeofday(&now, NULL);
double difference = ((double)now.tv_usec - (double)start.tv_usec) / 1000000;
printf("| %8.4fs | ", difference);
}
//生产者操作
void *Producer(void *arg)
{
int i,j;
int count = 0;
for (j = 0; j < N_Worktime; j++) //每个线程操作10次
{
Wait(empty);
Wait(mutex);
//随机等待一段时间(0~99微秒)
usleep(get_rand_num());
//向缓冲区添加数据
char c = get_rand_letter();
bf.str[bf.in] = c;
bf.in = (bf.in + 1) % MAX_BUFFER_SIZE;
bf.is_empty = 0;
bf.num++;
int first, last;
//打印时间
printTime();
//打印缓冲区数据
if (bf.in - 1 >= bf.out)
last = bf.in - 1;
else
last = bf.in - 1 + MAX_BUFFER_SIZE;
for (first = bf.out; bf.num>0&&first <= last; first++)
{
printf("%c ", bf.str[first % MAX_BUFFER_SIZE]);
}
int k;
for (k = bf.num; k <= MAX_BUFFER_SIZE; k++)
{
printf(" ");
}
printf("|");
//打印进程编号和产生数据
printf(" 生产者线程%d | 生产%c |\n",*(int*)arg, c);
Signal(mutex);
Signal(full);
}
return (void*)0;
}
//消费者操作
void *Consumer(void *arg)
{
int i, j;
int count = 0;
for (j = 0; j < N_Worktime; j++) //每个线程操作10次
{
Wait(full);
Wait(mutex);
//随机等待一段时间(0~99微秒)
usleep(get_rand_num());
//从缓冲区取走数据
char c = bf.str[bf.out];
bf.out = (bf.out + 1) % MAX_BUFFER_SIZE;
bf.is_empty = (bf.out == bf.in);
bf.num--;
int first, last;
//打印时间
printTime();
//打印缓冲区数据
if (bf.in - 1 >= bf.out)
last = bf.in - 1;
else
last = bf.in - 1 + MAX_BUFFER_SIZE;
for (first = bf.out; bf.num>0&&first <= last; first++)
{
printf("%c ", bf.str[first % MAX_BUFFER_SIZE]);
}
int k;
for (k = bf.num; k <= MAX_BUFFER_SIZE; k++)
{
printf(" ");
}
printf("|");
//打印进程编号和消费数据
printf(" 消费者线程%d | 消费%c |\n", *(int*)arg, c);
Signal(mutex);
Signal(empty);
}
return (void*)0;
}
int main()
{
printf("| 线程执行时间 | 缓冲区数据 | 当前执行的线程 | 产生或消费的数据 |\n");
printf("----------------------------------------------------------------------------\n");
gettimeofday(&start, NULL); //获取开始运行的时间（1970年1月1日到现在的秒和微秒）
//创建并初始化3个信号量
if (sem_init(&full, 0, 0) < 0)
{
perror("创建信号量SEM_FULL失败!\n");
exit(1);
}
if (sem_init(&empty, 0, 10) < 0)
{
perror("创建信号量SEM_EMPTY失败!\n");
exit(1);
}
if (sem_init(&mutex, 0, 1) < 0)
{
perror("创建信号量MUTEX失败!\n");
exit(1);
}
//初始化缓冲区
memset(bf.str, '\0', sizeof(bf.str));
bf.out = 0;
bf.in = 0;
bf.is_empty = 1;
bf.num = 0;
//定义4个线程
pthread_t producer0;
pthread_t producer1;
pthread_t consumer0;
pthread_t consumer1;
int arg0 = 0, arg1 = 1; //传入函数的参数（代表进程编号）
//创建线程
if (pthread_create(&producer0, NULL, Producer, &arg0) != 0)
{
perror("创建生产者线程失败！\n");
exit(1);
}
if (pthread_create(&producer1, NULL, Producer, &arg1) != 0)
{
perror("创建生产者线程失败！\n");
exit(1);
}
if (pthread_create(&consumer0, NULL, Consumer, &arg0) != 0)
{
perror("创建消费者线程失败！\n");
exit(1);
}
if (pthread_create(&consumer1, NULL, Consumer, &arg1) != 0)
{
perror("创建消费者线程失败！\n");
exit(1);
}
//ret:接收函数pthread_join中被等待线程的返回值
int ret1,ret2,ret3,ret4;
pthread_join(producer0, (void **)&ret1);
pthread_join(producer1, (void **)&ret2);
pthread_join(consumer0, (void **)&ret3);
pthread_join(consumer1, (void **)&ret4);
printf("----------------------------------------------------------------------------\n");
printf("程序结束!\n");
return 0;
}
开始看到这个实验的题目时，头绪不多。因为所需要用到的Linux相关库<sys/shm.h>、<sys/sem.h>是之前没有接触过的。之后我通过查找资料，更好地了解如何使用它们实现生产者-消费者问题，具体使用的库函数可详见我上方的实验报告。这个过程中踩的一些坑被我记录下来：
刚开始运行时提示错误“共享内存创建失败”。
这是因为创建共享内存的函数的第三个参数没有或上IPC_CREAT。
如何打印缓冲区中的数据是需要细心考虑的：
in-1<out。这时就需要将last赋值为in-1+MAX_BUFFER_SIZE，做% MAX_BUFFER_SIZE后就可指向缓冲区数组的正确位置。时间的处理主要用到timeval结构体，在前文也已进行说明。错误提示输出perror()和较美观的输出界面体现了人性化设计。
整个问题的实现可以由实验指导书的流程图较好地被标示出来，理解了这几个流程后，多点耐心、细心就可以完成代码实现。
对于思考题中的线程实现，则需要去了解库<pthread.h>和<semaphore.h>。整体框架和进程实现大同小异，不同点在思考题已做说明。
总的来说，这次实验让我对生产者-消费者问题的实现和基于Linux环境的C语言编程有了更加深刻的了解，较好地锻炼了我的逻辑思维和代码设计能力。