南昌大学实验报告

学生姓名：    王维      学    号：     6100212202       专业班级：  网工121班 

实验类型：□ 验证 □ 综合 □ 设计 □ 创新   实验日期：    5/14    实验成绩：     

一、    实验项目名称

实验8 多线程“生产者消费者”实验

二、    实验目的

 “生产者消费者”问题是一个著名的同时性编程问题的集合。通过学习经典的“生产者消费者”问题的实验，读者可以进一步熟悉Linux中的多线程编程，并且掌握用信号量处理线程间的同步和互斥问题。

三、    实验内容

“生产者—消费者”问题描述如下。

有一个有限缓冲区和两个线程：生产者和消费者。他们分别不停地把产品放入缓冲区和从缓冲区中拿走产品。一个生产者在缓冲区满的时候必须等待，一个消费者在缓冲区空的时候也必须等待。另外，因为缓冲区是临界资源，所以生产者和消费者之间必须互斥执行。它们之间的关系如图1所示。

图1  生产者消费者问题描述

这里要求使用有名管道（将在下一章介绍，可提前了解）来模拟有限缓冲区，并且使用信号量来解决“生产者—消费者”问题中的同步和互斥问题。

四、    实验步骤

（1）信号量的考虑。

这里使用3个信号量，其中两个信号量avail和full分别用于解决生产者和消费者线程之间的同步问题，mutex是用于这两个线程之间的互斥问题。其中avail表示有界缓冲区中的空单元数，初始值为N；full表示有界缓冲区中非空单元数，初始值为0；mutex是互斥信号量，初始值为1。

（2）画出流程图。

本实验流程图如图2所示。

图2  “生产者—消费者”实验流程图

（3）编写代码

本实验的代码中采用的有界缓冲区拥有3个单元，每个单元为5个字节。为了尽量体现每个信号量的意义，在程序中生产过程和消费过程是随机（采取0～5s的随机时间间隔）进行的，而且生产者的速度比消费者的速度平均快两倍左右（这种关系可以相反）。生产者一次生产一个单元的产品（放入“hello”字符串），消费者一次消费一个单元的产品。

/*producer-customer.c*/

#include <stdio.h>

#include <stdlib.h>

#include <unistd.h>

#include <fcntl.h>

#include <pthread.h>

#include <errno.h>

#include <semaphore.h>

#include <sys/ipc.h>

#define MYFIFO            "myfifo"     /* 缓冲区有名管道的名字 */

#define BUFFER_SIZE       3            /* 缓冲区的单元数 */

#define UNIT_SIZE         5             /* 每个单元的大小 */

#define RUN_TIME          30            /* 运行时间 */

#define DELAY_TIME_LEVELS     5.0     /* 周期的最大值 */

int fd;

time_t end_time;

sem_t mutex, full, avail;              /* 3个信号量 */

/*生产者线程*/

void *producer(void *arg)

{

     int real_write;

     int delay_time = 0;

    

     while(time(NULL) < end_time)

     {

          delay_time = (int)(rand() * DELAY_TIME_LEVELS/(RAND_MAX) / 2.0) + 1;

          sleep(delay_time);

          /*P操作信号量avail和mutex*/

          sem_wait(&avail);

          sem_wait(&mutex);

          printf("\nProducer: delay = %d\n", delay_time);

          /*生产者写入数据*/

          if ((real_write = write(fd, "hello", UNIT_SIZE)) == -1)

          {

               if(errno == EAGAIN)

               {

                    printf("The FIFO has not been read yet.Please try later\n");

               }

          }

          else

          {

               printf("Write %d to the FIFO\n", real_write);

          }

         

          /*V操作信号量full和mutex*/

          sem_post(&full);

          sem_post(&mutex);

     }    

     pthread_exit(NULL);

}

/* 消费者线程*/

void *customer(void *arg)

{

     unsigned char read_buffer[UNIT_SIZE];

     int real_read;

     int delay_time;

    

     while(time(NULL) < end_time)

     {

          delay_time = (int)(rand() * DELAY_TIME_LEVELS/(RAND_MAX)) + 1;

          sleep(delay_time);

          /*P操作信号量full和mutex*/

          sem_wait(&full);

          sem_wait(&mutex);

          memset(read_buffer, 0, UNIT_SIZE);

                    printf("\nCustomer: delay = %d\n", delay_time);

          if ((real_read = read(fd, read_buffer, UNIT_SIZE)) == -1)

          {

               if (errno == EAGAIN)

               {

                   printf("No data yet\n");

               }

          }

          printf("Read %s from FIFO\n", read_buffer);

          /*V操作信号量avail和mutex*/

          sem_post(&avail);

          sem_post(&mutex);

     }

     pthread_exit(NULL);

}

int main()

{

     pthread_t thrd_prd_id,thrd_cst_id;

     pthread_t mon_th_id;

     int ret;

    

     srand(time(NULL));

     end_time = time(NULL) + RUN_TIME;

     /*创建有名管道*/

     if((mkfifo(MYFIFO, O_CREAT|O_EXCL) < 0) && (errno != EEXIST))

     {

          printf("Cannot create fifo\n");

          return errno;

     }              

     /*打开管道*/

     fd = open(MYFIFO, O_RDWR);

     if (fd == -1)

     {

          printf("Open fifo error\n");

          return fd;

     }    

     /*初始化互斥信号量为1*/

     ret = sem_init(&mutex, 0, 1);

     /*初始化avail信号量为N*/

     ret += sem_init(&avail, 0, BUFFER_SIZE);

     /*初始化full信号量为0*/

     ret += sem_init(&full, 0, 0);

     if (ret != 0)

     {

          printf("Any semaphore initialization failed\n");

          return ret;

     }

     /*创建两个线程*/

     ret = pthread_create(&thrd_prd_id, NULL, producer, NULL);

     if (ret != 0)

     {

          printf("Create producer thread error\n");

          return ret;

     }

     ret = pthread_create(&thrd_cst_id, NULL, customer, NULL);

     if(ret != 0)

     {

          printf("Create customer thread error\n");

          return ret;

     }

     pthread_join(thrd_prd_id, NULL);

     pthread_join(thrd_cst_id, NULL);

     close(fd);

     unlink(MYFIFO);

     return 0;

}

五、    实验结果

六、    实验心得

通过这次实验，我深入理解了生产者消费者的过程，这是经典的操作系统理论中的进程/线程同步模型。本次实验使用了信号量来完成线程中的同步过程。注意编译的时候gcc要加参数-lpthread。

本实验使用的信号量是POSIX的信号量。函数sem_wait是P操作，而sem_post是V操作。此外对于pthread线程库的函数也有了新的认识，pthread_join函数使当前线程阻塞，等待指定线程结束。其概念类似于进程中的wait/waitpid函数。

另外测试代码中有一处错误：if((mkfifo(MYFIFO, O_CREAT|O_EXCL) < 0) && (errno != EEXIST))。生成有名管道的mkfifo函数的第二个参数，并不是相当于open函数的第二个参数（标志），而是相当于open的第三个参数，即权限。可以是S_IRWXU、S_IRUSR等宏的或组合，或者直接使用八进制数字来表示权限，如0644。

测试代码虽然能编译通过，但是运行的时候会提示缺少文件权限。我改成了下面的代码：

if((mkfifo(MYFIFO, 0644) < 0) && (errno != EEXIST))

第二篇：通过多线程解决生产者,消费者问题

进程是一个静态的概念。

线程是一个进程里面的不同的执行路径。

在java技术中，线程通常是抢占式的而不需要时间片分配进程（分配给每个线程相等的CPU时间的进程）。抢占式调度模型就是许多线程处于可以运行状态（等待状态），但实际上只有一个线程在运行。该线程一直运行到它终止进入可运行状态（等待状态），或者另一个具有更高优先级的线程变成可运行状态。在后一种情况下，低优先级的线程被高优先级的线程抢占，高优先级的线程获得运行的机会。Java线程调度器支持不同优先级线程的抢先方式，Java线程调度器本身不支持相同优先级线程的时间片轮换。它依赖于Java运行时系统所在的操作系统：如果操作系统支持时间片的轮换，则线程调度器就支持相同优先级线程的时间片轮换。

在同一个时间点，一个cpu只能支持一个线程。Java运行时系统实现了一个用于调度线程执行的线程调度器，用于确定某一时刻由哪一个线程在CPU上运行。

线程的启动必须使用Thread类的start（）方法。不能在主线程main里直接调用run（）。

创建线程的方法：

1. 定义线程实现Runnable接口

2. 从Thread类继承。

class Runner1 implements Runnable

class Runner1 extends Thread

启动线程的方式：

1. 通过Thread类

Thread t = new Thread(r);

t.start();

如果是使用class Runner1 extends Thread来创建线程，可以通过如下方法来启动线程。因为只有Thread类有start方法。

2. 直接Runner1 r = new Runner1();

r.start();

3. 通过Thread类 (父类指向子类的对象)

Thread t = new Runner1();

t.start();

sleep()会抛出interruptedException必须try catch。

可以用interrupt打断。

使用join（）的时候，相当于方法调用。也就是说，使用了join方法的时候，先执行完使用了join的线程。相当于是说，使用了join线程的优先级最高。必须等到使用了join的线程执行完。join（）同样会抛出interruptedException。必须try catch。

yield（）就是执行到yield的时候让出一下cpu时间，但是只让出一次。

线程同步：线程同步的含义就是在于，不同的线程要访问同一个对象，这部分被称为critical sections。

同步有两种方式：同步块和同步方法

Synchronized可以锁定当前对象。一个线程已经进入到被锁定的区域，不可以有别的线程进入。

解决死锁的方式：把锁的粒度加粗一些。也就是说，尽量锁定一个对象，不要锁定两个对象。

Synchronized锁定的是代码，而不是某个变量。也就是说，别的线程依然可以访问这个变量，只是不能通过锁定的代码部分对变量进行值的改变。但是可以通过别的方法改变变量的值。

锁定问题：加了锁定，效率变低。不加锁定，可能会出现数据不一致的错误。

程序中让线程的终止，一般不用stop方法，常用的是设置一个flag变量，然后结合interrupt（）方法。

如果线程已经运行起来，又阻塞了，为了让线程停止：

首要考虑采用join（）方法，正常停止。

再考虑使用exception的方式停止。

生产者，消费者问题。其实就是一个堆栈问题当中用到了wait，notify，notifyAll。 有生产者，负责生产。

消费者，负责消费。

//每个生产者或者消费者都是一个线程。

//主类。

publicclassProducerConsumer {

}

//产品

classWoTou {

int id; WoTou(int id) { } this.id = id; publicstaticvoid main(String[] args) { } SyncStackss = newSyncStack(); Producer p = newProducer(ss); Consumer c = newConsumer(ss); //三个生产者，一个消费者 new Thread(p).start(); new Thread(p).start(); new Thread(p).start(); new Thread(c).start();

}

public String toString() { } return"WoTou : " + id;

//堆栈类

classSyncStack {

//栈顶指针 int index = 0; //栈数组，大小为6 WoTou[] arrWT = newWoTou[6]; //压栈由生产者调用。必须是同步方法。 publicsynchronizedvoid push(WoTouwt) { } //出栈由消费者调用。必须是同步方法。 publicsynchronizedWoTou pop() { //空了，栈顶指针指向栈最底部。就阻塞。必须是同步方法才可以wait。 //必须用while，不能用if while(index == 0) { try { this.wait(); e.printStackTrace(); } catch (InterruptedException e) { //满了，栈顶指针指向栈最顶部。就阻塞。 //必须是同步方法才可以wait。如果方法没有注明synchronized，会编译出错。 while(index == arrWT.length) { } //唤醒消费者的线程 this.notifyAll(); //压栈 arrWT[index] = wt; //指针上升 index ++; try { } //this是当前对象正在访问的线程，wait //wait必须被人唤醒，和sleep不一样，自己醒不了。 this.wait(); e.printStackTrace(); } catch (InterruptedException e) {

}

} } } this.notifyAll(); //指针下移 index--; //拿到产品 returnarrWT[index];

//生产者实现runnable

class Producer implements Runnable {

}

//消费者

class Consumer implements Runnable {

SyncStackss = null; Consumer(SyncStackss) { } this.ss = ss; SyncStackss = null; Producer(SyncStackss) { } publicvoid run() { } //每个人最多生产20个馒头 for(int i=0; i<20; i++) { } //生产一个产品，并指定一个id WoTouwt = newWoTou(i); //生产好了以后压栈 ss.push(wt); try { //随机睡眠一段时间 //控制睡眠的时间 Thread.sleep((int)(Math.random() * 200)); e.printStackTrace(); this.ss = ss; System.out.println("生产了：" + wt); } catch (InterruptedException e) { }

publicvoid run() { //每个人最多消费20个馒头 for(int i=0; i<20; i++) { //消费就是出栈 WoTouwt = ss.pop(); try { //随机睡眠一段时间 //控制睡眠的时间 System.out.println("消费了: " + wt);

}

}

} Thread.sleep((int)(Math.random() * 1000)); catch (InterruptedException e) { e.printStackTrace(); } }