线程之间的协作(二)生产者与消费者

考虑这样一个饭店,它有一个厨师(Chef)和一个服务员(Waiter)。这个服务员必须等待厨师准备好菜品。当厨师准备好时,他会通知服务员,之后服务员上菜,然后返回继续等待。这是一个任务协作的示例:厨师代表生产者,而服务员代表消费者。两个任务必须在菜品被生产和消费时进行握手,而系统必须以有序的方式关闭。下面是对这个叙述建模的代码:

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
class Meal {
    private final int orderNum;
    public Meal(int orderNum) {
        this.orderNum = orderNum;
    }
    @Override
    public String toString() {
        return "Meal " + orderNum;
    }
}
class Waiter implements Runnable {
    private Restaurant r;
    public Waiter(Restaurant r) {
        this.r = r;
    }
    @Override
    public void run() {
        try {
            while(!Thread.interrupted()) {
                synchronized (this) {
                    while(r.meal == null) {
                        wait();//等待厨师做菜
                    }
                }
                System.out.println("Waiter got " + r.meal);
                synchronized (r.chef) {
                    r.meal = null;//上菜
                    r.chef.notifyAll();//通知厨师继续做菜
                }
            }
        } catch (InterruptedException e) {
            System.out.println("Waiter task is over.");
        }
    }
}
class Chef implements Runnable {
    private Restaurant r;
    private int count = 0;//厨师做的菜品数量
    public Chef(Restaurant r) {
        this.r = r;
    }
    @Override
    public void run() {
        try {
            while(!Thread.interrupted()) {
                synchronized (this) {
                    while(r.meal != null) {
                        wait();//等待服务员上菜
                    }
                }
                if (++count > 10) {
                    System.out.println("Meal is enough, stop.");
                    r.exec.shutdownNow();
                }
                System.out.print("Order up! ");
                synchronized (r.waiter) {
                    r.meal = new Meal(count);//做菜
                    r.waiter.notifyAll();//通知服务员上菜
                }
                TimeUnit.MILLISECONDS.sleep(100);
            }
        } catch (InterruptedException e) {
            System.out.println("Chef task is over.");
        }
    }
}
public class Restaurant {
    Meal meal;
    ExecutorService exec = Executors.newCachedThreadPool();
    //厨师和服务员都服务于同一个饭店
    Waiter waiter = new Waiter(this);
    Chef chef = new Chef(this);
    public Restaurant() {
        exec.execute(waiter);
        exec.execute(chef);
    }
    public static void main(String[] args) {
        new Restaurant();
    }
}

执行结果:

Order up! Waiter got Meal 1
Order up! Waiter got Meal 2
Order up! Waiter got Meal 3
Order up! Waiter got Meal 4
Order up! Waiter got Meal 5
Order up! Waiter got Meal 6
Order up! Waiter got Meal 7
Order up! Waiter got Meal 8
Order up! Waiter got Meal 9
Order up! Waiter got Meal 10
Meal is enough, stop.
Order up! Waiter task is over.
Chef task is over.

Restaurant是Waiter和Chef的焦点,它们都必须知道在为哪个饭店工作,因为他们必须和这家饭店的窗口打交道,一边放置或拿取菜品r.meal。在run()中,waiter进入wait()模式,停止其任务,直至被Chef的notifyAll()唤醒。由于这是一个非常简单的程序,因此我们知道只有一个任务将在Waiter的锁上等待:即Waiter任务自身。出于这个原因,理论上可以调用notify()而不是notifyAll()。但是,在更复杂的情况下,可能会有多个任务在某个特定对象锁上等待,因此你不知道哪个任务应该被唤醒。因此调用notifyAll()要更安全一些,这样可以唤醒等待这个锁的所有任务,而每个任务都必须决定这个通知是否与自己相关。

一旦Chef送上Meal并通知Waiter,这个Chef就将等待,知道Waiter收集到订单并通知Chef,之后Chef就可以做下一份菜品了。

注意,wait()被包装在一个while()字句中,这个语句在不断的测试正在等待的事物。乍一看有点怪——如果在等待一个订单,一单你被唤醒,这个订单就必定是可获得的,对吗?正如前面注意到的,在更复杂的并发应用中,某个其他的任务可能在Waiter被唤醒时突然插足并拿走订单。因此唯一安全的方式是使用下面这种wait()的惯用法:

while(conditionIsNotMet) {
    wait();
}

这可以保证在你退出等待循环之前,条件将得到满足,并且如果你收到了关于某事物的通知,而它与这个条件并无关系,或者在你完全退出等待循环之前,这个条件发生了变化,都可以确保你重返等待状态。

请注意观察,对notifyAll()的调用必须首先捕获Waiter上的锁,而在Waiter.run()中的对wait()的调用会自动的释放这个所,因此这是由可能实现的。因为调用notifyAll()必然拥有这个锁,所以这可以保证两个试图在同一个对象上调用notifyAll()的任务不会互相冲突。

通过把整个run()方法体放到一个try语句块中,可以使得这两个run()方法都被设计为可以有序的关闭。catch子句将紧挨着run()方法的括号之前结束,因此,如果这个任务收到了InterruptedException,它将在捕获到异常后立即结束。

注意,在Chef中,在调用shutdownNow()之后,你应该直接从run()返回,并且通常这就是你应该做的。但是,以这种方式执行还有一些更有趣的东西。记住,shutdownNow()将向所有有ExecutorService启动的任务发送interrupt(),但是在Chef中,任务并没有在获得该interrupt()立即结束,因为当任务试图进入一个(可中断的)阻塞操作时,这个中断只能抛出InterruptedException。因此你将首先看到“Order up!”,然后Chef试图调用sleep()方法时,抛出了InterruptedException。如果你移除对sleep()的调用,那么这个任务将回到run()循环的顶部,并由于Thread.interrupted()测试而退出,同时并不抛异常。

在这两个示例中,对于一个任务而言,只有一个单一的地方用于存放对象,从而使得另一个任务稍后可以使用这个对象。但是,在典型的生产者-消费者实现中,应使用先进先出队列来存储被生产和消费的对象。

时间: 2024-10-05 18:35:57

线程之间的协作(二)生产者与消费者的相关文章

java并发系列(二)-----线程之间的协作(wait、notify、join、CountDownLatch、CyclicBarrier)

在java中,线程之间的切换是由操作系统说了算的,操作系统会给每个线程分配一个时间片,在时间片到期之后,线程让出cpu资源,由其他线程一起抢夺,那么如果开发想自己去在一定程度上(因为没办法100%控制它)让线程之间互相协作.通信,有哪些方式呢? wait.notify.notifyAll 1.void wait( ) 导致当前的线程等待,直到其他线程调用此对象的notify( ) 方法或 notifyAll( ) 方法 2.void wait(long timeout) 导致当前的线程等待,直到

深入理解并发(二)--生产者及消费者

生产者及消费者问题,是线程操作中的一个经典案列.但由于线程运行的不确定性,生产者及消费者可能会产生一些问题: 试想,如果生产者线程向存储数据空间添加了部分信息,但没有添加全部,这时就切换到消费者线程,这时消费者线程将会把已经添加了的部分信息,后上一次的信息混淆了,导致出错. 或者,若生产者放数据,与消费者取数据的速度不匹配,也会出现问题:即可能会出现,生产者放了多条数据,消费者才取了一条,导致数据丢失:或生产者只放了一条数据,但消费者已经取了多条,这会导致重复取出数据. 举例说明: class

C# 线程(四):生产者和消费者

From : http://kb.cnblogs.com/page/42530/ 前面说过,每个线程都有自己的资源,但是代码区是共享的,即每个线程都可以执行相同的函数.这可能带来的问题就是几个线程同时执行一个函数,导致数据的混乱,产生不可预料的结果,因此我们必须避免这种情况的发生. C#提供了一个关键字lock,它可以把一段代码定义为互斥段(critical section),互斥段在一个时刻内只允许一个线程进入执行,而其他线程必须等待.在C#中,关键字lock定义如下: lock(expres

java 中多线程之间的通讯之生产者和消费者 (多个线程之间的通讯)

在真实开发 中关于多线程的通讯的问题用到下边的例子是比较多的 不同的地方时if 和while 的区别 如果只是两个线程之间的通讯,使用if是没有问题的. 但是在多个线程之间就会有问题 1 /* 2 * 这个例子用来解释多个生产者和多个消费者的情况 3 */ 4 5 /* 6 * 资源库 7 */ 8 class Resource 9 { 10 private String name; 11 private int count = 1; 12 private boolean flag = fals

线程的基本协作和生产者消费者

协作基础(wait/notify) Java的根父类是Object,Java在Object类而非Thread类中,定义了一些线程协作的基本方法,使得每个对象都可以调用这些方法,这些方法有两类,一类是wait,另一类是notify. wait方法主要有两个: public final void wait() throws InterruptedException public final native void wait(long timeout) throws InterruptedExcept

漫谈并发编程(五):线程之间的协作

编写多线程程序需要进行线程协作,前面介绍的利用互斥来防止线程竞速是来解决线程协作的衍生危害的.编写线程协作程序的关键是解决线程之间的协调问题,在这些任务中,某些可以并行执行,但是某些步骤需要所有的任务都结束之后才能开动. wait()与notifyAll() wait()使你可以等待某个条件发生变化,wait()会在等待外部世界产生变化的时候将任务挂起,并且只有在notify()或notifyAll()发生时,即表示发生了某些感兴趣的事物,这个任务才会被唤醒并去检查所产生的变化. 调用sleep

java 线程之间的协作 wait()与notifyAll()

package org.rui.thread.block; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; // wax蜡 电气自己主动方式 public class WaxOMatic { public static void main(String[] args) throws Interrupte

ZooKeeper典型应用(二) 生产者与消费者

In this tutorial, we show simple implementations of barriers and producer-consumer queues using ZooKeeper. We call the respective classes Barrier and Queue. These examples assume that you have at least one ZooKeeper server running. 本文将告诉你如何使用 Zookeep

生产者与消费者模式(线程的同步与互斥)

死锁产生的四个条件: 1.互斥使用(资源独占) 一个资源每次只能给一个进程使用 .2.不可强占(不可剥夺) 资源申请者不能强行的从资源占有者手中夺取资源,资源只能由占有者自愿释放 .3.请求和保持(部分分配,占有申请) 一个进程在申请新的资源的同时保持对原有资源的占有(只有这样才是动态申请,动态分配) .4.循环等待 存在一个进程等待队列 {P1 , P2 , - , Pn}, 其中P1等待P2占有的资源,P2等待P3占有的资源,-,Pn等待P1占有的资源,形成一个进程等待环路 生产者:生产数据