黑马程序员技术交流社区

标题: 【上海校区】Java并发（二）线程间通信-生产者消费者模型 [打印本页]

---

作者: 不二晨    时间: 2018-10-16 09:06
标题: 【上海校区】Java并发（二）线程间通信-生产者消费者模型
1 线程通信 - wait、notify、notifyAll

1.1 wait方法

public final void wait() throws InterruptedException {
        wait(0);
}
public final native void wait(long timeout) throws InterruptedException;
1
2
3
4
和上一篇文章一样，咱们先看官方文档：

Causes the current thread to wait until another thread invokes the {@link java.lang.Object#notify()} method or the {@link java.lang.Object#notifyAll()} method for this object, or some other thread interrupts the current thread, or a certain amount of real time has elapsed.
The current thread must own this object’s monitor. The thread releases ownership of this monitor and waits until another thread notifies threads waiting on this object’s monitor to wake up either through a call to the notify method or the notifyAll method. The thread then waits until it can re-obtain ownership of the monitor and resumes execution.
As in the one argument version, interrupts and spurious wakeups are possible, and this method should always be used in a loop:
synchronized (obj) {
while (<condition does not hold>)
obj.wait(timeout, nanos);
… //Perform action appropriate to condition
}
This method should only be called by a thread that is the owner of this object’s monitor. See the notify method for a description of the ways in which a thread can become the owner of a monitor.
核心内容归纳如下：

1、wait会导致当前线程等待，直到被其他线程调用notify、notifyAll方法唤醒。

注意：

通常我们说“对象调用方法”，但是在多线程中喜欢说“线程调用方法”。实际上线程中的对象在调用方法，都可以理解为线程在调用方法。
wait方法是由锁对象来调用的，也就是monitor，而且唤醒线程的notify方法也是由同一个锁对象调用的。
2、当前线程必须拥有此对象的monitor，也就是对象锁。当前线程调用wait方法后，会释放这个对象锁并等待，直到另外一个线程调用notify方法去唤醒。

注意：当前线程被唤醒不代表立即获取了对象的monitor，只有等另一线程调用完notify()并退出synchronized块，释放对象锁后，当前线程才可获得锁执行。

3、wait方法应始终在循环中使用。

4、wait方法只能被拥有对象锁的线程调用，即wait方法只能在同步方法或同步语句块中调用。

1.2 notify方法

public final native void notify();
1
官方解释：

Wakes up a single thread that is waiting on this object’s monitor. If any threads are waiting on this object, one of them is chosen to be awakened. The choice is arbitrary and occurs at the discretion of the implementation. A thread waits on an object’s monitor by calling one of the wait methods.
The awakened thread will not be able to proceed until the current thread relinquishes the lock on this object. The awakened thread will compete in the usual manner with any other threads that might be actively competing to synchronize on this object; for example, the awakened thread enjoys no reliable privilege or disadvantage in being the next thread to lock this object.
This method should only be called by a thread that is the owner of this object’s monitor. A thread becomes the owner of the object’s monitor in one of three ways:
By executing a synchronized instance method of that object.
By executing the body of a synchronized statement that synchronizes on the object.
For objects of type Class, by executing a synchronized static method of that class.
Only one thread at a time can own an object’s monitor.
核心内容解释如下：

1、调用notify方法，可唤醒在对象锁（monitor）上等待的单个线程。如果有很多线程等待此对象锁，则随机选择一个去唤醒。

2、在当前线程放弃对该对象的锁定之前，被唤醒的线程将无法继续运行。

3、notify()方法只应由作为此对象锁（monitor）所有者的线程调用。即notify方法只能在同步方法或同步语句块中调用。

4、每一次只有一个线程可以拥有对象的锁（monitor）。

1.3 notifyAll方法

notifyAll方法与notify方法基本相似，只是notifyAll方法，用于唤醒在该对象锁（monitor）上等待的所有线程。

2 生产者 - 消费者模型多种实现方式

生产者消费者问题是研究多线程程序绕不开的经典问题之一。生产者-消费者模型，简单概括如下：

生产者持续生产，直到缓冲区满，阻塞；缓冲区不满后，生产者继续生产
消费者持续消费，直到缓冲区空，阻塞；缓冲区不空后，消费者持续消费
生产者可以有多个，消费者可以有多个
可通过如下条件验证模型的正确性：

同一产品的消费行为一定发生在生产行为之后
任意时刻，缓冲区的大小不得小于0，也不得大于其最大容量
在实现模型之前，首先构思一下基本的类结构。


主要设计如下：

Storage（仓库）接口，接口中设计了produce()、consume()方法。由于后期produce方法、consumer方法都会改动。所以在设计的时候将可能发生的变化集中到一个类中，不影响原有的构架设计，同时无需修改其他业务代码。
Storage实现类，实现具体的produce()、consume()方法
Producer（生产者）类
Consumer（消费者类）类
2.1 wait、notify实现

/**
* 仓库接口，象征着生产者和消费者中间的队列
* @Author: cherry
* @Date: Created in 2018/9/27 15:36
*/
public interface Storage {

    /**
     * 生产行为
     * @param num 生产商品的数量
     */
    void produce(int num);

    /**
     * 消费行为
     * @param num 消费商品的数量
     */
    void consume(int num);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
/**
* Storage接口的实现类，produce、consume方法采用wait、notify实现
* @Author: cherry
* @Date: Created in 2018/9/27 15:41
*/
public class StorageWaitNotifyImpl implements Storage {
    /**
     * 队列最大容量
     */
    public static final int MAX_SIZE = 100;

    /**
     * 仓库中，存放产品的容器。也可以用作同步对象锁
     */
    private LinkedList<Object> queue = new LinkedList<>();


    @Override
    public void produce(int num) {
        synchronized (queue){
            //判断缓存队列大小是否会超出最大容量
            while(queue.size() + num > MAX_SIZE){
                System.out.println("要生产的数量："+ num + "，现库存量：" + queue.size() + "，总容量：100，暂停生产！");
                try {
                    queue.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            //线程被唤醒或未超出最大容量
            for (int i = 0; i < num; i++) {
                queue.add(new Object());
            }

            //生产者生产完毕，唤醒消费者
            queue.notifyAll();
            System.out.println("已生产数量："+ num + "，现库存量：" + queue.size());
        }
    }

    @Override
    public void consume(int num) {
        synchronized (queue){
            //判断缓存队列大小是否小于0
            while(queue.size() - num < 0){
                System.out.println("要消费的数量："+ num + "，现库存量：" + queue.size() + "，暂停消费！");
                try {
                    queue.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            //线程被唤醒或队列不为空
            for (int i = 0; i < num; i++) {
                queue.remove();
            }

            //消费者消费完毕，唤醒生产者
            queue.notifyAll();
            System.out.println("已消费数量："+ num + "，现库存量：" + queue.size());
        }
    }
}
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
/**
* 生产者
* @Author: cherry
* @Date: Created in 2018/9/27 15:47
*/
public class Producer implements Runnable {
    /**
     * 聚合仓库对象，produce方法中会调用仓库对象的produce方法
     */
    private Storage storage;

    /**
     * 生产数量
     */
    private int num;

    public Producer(){

    }

    public Producer(Storage storage,int num){
        this.storage = storage;
        this.num = num;
    }


    @Override
    public void run() {
        this.produce(num);
    }

    /**
     * 生产者produce方法中调用storage.produce方法
     * @param num 生产数量
     */
    private void produce(int num){
        this.storage.produce(num);
    }

    public Storage getStorage() {
        return storage;
    }

    public void setStorage(Storage storage) {
        this.storage = storage;
    }

    public int getNum() {
        return num;
    }

    public void setNum(int num) {
        this.num = num;
    }
}
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
/**
* 消费者
* @Author: cherry
* @Date: Created in 2018/9/27 16:02
*/
public class Consumer implements Runnable {
    /**
     * 聚合仓库对象，this.consume()方法中会调用仓库对象的consume方法
     */
    private Storage storage;

    /**
     * 消费数量
     */
    private int num;

    public Consumer(){

    }

    public Consumer(Storage storage,int num){
        this.storage = storage;
        this.num = num;
    }

    @Override
    public void run() {
        this.consume(num);
    }

    private void consume(int num){
        this.storage.consume(num);
    }

    public Storage getStorage() {
        return storage;
    }

    public void setStorage(Storage storage) {
        this.storage = storage;
    }

    public int getNum() {
        return num;
    }

    public void setNum(int num) {
        this.num = num;
    }
}
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
/**
* @Author: cherry
* @Date: Created in 2018/9/27 16:37
*/
public class TestQueue {
    public static void main(String[] args){
        Storage storage = new StorageWaitNotifyImpl();

        //初始化生产者
        Producer producer1 = new Producer(storage,30);
        Producer producer2 = new Producer(storage,40);
        Producer producer3 = new Producer(storage,50);
        Producer producer4 = new Producer(storage,60);
        Producer producer5 = new Producer(storage,70);


        //初始化消费者
        Consumer consumer1 = new Consumer(storage,30);
        Consumer consumer2 = new Consumer(storage,40);
        Consumer consumer3 = new Consumer(storage,50);
        Consumer consumer4 = new Consumer(storage,60);
        Consumer consumer5 = new Consumer(storage,70);

        new Thread(producer1).start();
        new Thread(producer2).start();
        new Thread(producer3).start();
        new Thread(producer4).start();
        new Thread(producer5).start();
        new Thread(consumer1).start();
        new Thread(consumer2).start();
        new Thread(consumer3).start();
        new Thread(consumer4).start();
        new Thread(consumer5).start();
    }

}
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
运行结果：

已生产数量：30，现库存量：30
已生产数量：40，现库存量：70
要生产的数量：50，现库存量：70，总容量：100，暂停生产！
要生产的数量：60，现库存量：70，总容量：100，暂停生产！
要生产的数量：70，现库存量：70，总容量：100，暂停生产！
已消费数量：30，现库存量：40
已消费数量：40，现库存量：0
已生产数量：70，现库存量：70
要生产的数量：60，现库存量：70，总容量：100，暂停生产！
已消费数量：60，现库存量：10
已生产数量：50，现库存量：60
要消费的数量：70，现库存量：60，暂停消费！
要生产的数量：60，现库存量：60，总容量：100，暂停生产！
已消费数量：50，现库存量：10
已生产数量：60，现库存量：70
已消费数量：70，现库存量：0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
2.2 await、signal实现

在实现之前，咱们分析一个简单的性能问题：在2.1小节中的queue.notifyAll()，该方法会唤醒所有等待queue对象锁的线程，也就是说如果生产者刚刚生产完毕，调用notifyAll()方法想唤醒的是消费者，结果却唤醒了生产者，被唤醒的生产者（因为超出容量）又将等待，这样效率会很低！

如何解决此效率问题呢？后面会有解释，咱们先往下走。

在JDK5.0之后，java.util.concurrent包提供了Lock && Condition。

Lock实现提供了比使用synchronized方法和语句块可获得的更广泛的锁定操作，它能以更优雅的方式处理线程同步问题。利用lock()和unlock()来锁定需同步的语句。当然Lock的强大之处不仅仅在于此，读写锁（ReadWriteLock）更彰显其强大。读写锁（ReadWriteLock）可实现：读与写互斥、写与写互斥、而读与读不互斥。

Condition用来替代传统的Object的wait()、notify()实现线程间的通信。在Condition中，用await()替换wait()，用signal()替换notify()，用signalAll()替换notifyAll()，传统线程的通信方式，Condition都可以实现，这里注意，Condition是被绑定到Lock上的，要创建一个Lock的Condition必须用newCondition()方法。调用Condition的await()、signal()和signalAll()方法，都必须在lock保护之内，就是说必须在lock.lock()和lock.unlock之间才可以使用。

现在要使用Lock、以及Condition的await、signal方法去改写模型。那么只需重新实现Storage接口即可，其他代码几乎不需要改变。

/**
* Storage的实现类。采用Lock、多个Condition去实现同步
* @Author: cherry
* @Date: Created in 2018/9/27 19:44
*/
public class StorageConditionImpl implements Storage {

    /**
     * 队列最大容量
     */
    public static final int MAX_SIZE = 100;

    /**
     * 仓库中，存放产品的容器
     */
    private LinkedList<Object> queue = new LinkedList<>();

    private Lock lock = new ReentrantLock();

    /**
     * 作为生产者使用的同步锁
     */
    private Condition notFull = lock.newCondition();

    /**
     * 作为消费者使用的同步锁
     */
    private Condition notEmpty = lock.newCondition();

    @Override
    public void produce(int num) {
        lock.lock();
        try{
            //判断缓存队列大小是否会超出最大容量
            while(queue.size() + num > MAX_SIZE){
                System.out.println("要生产的数量："+ num + "，现库存量：" + queue.size() + "，总容量：100，暂停生产！");
                try {
                    //生产者等待
                    notFull.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            //线程被唤醒或未超出最大容量
            for (int i = 0; i < num; i++) {
                queue.add(new Object());
            }

            //生产者生产完毕，唤醒消费者
            notEmpty.signalAll();
            System.out.println("已生产数量："+ num + "，现库存量：" + queue.size());
        }finally {
            lock.unlock();
        }
    }

    @Override
    public void consume(int num) {
        lock.lock();
        try{
            //判断缓存队列大小是否小于0
            while(queue.size() - num < 0){
                System.out.println("要消费的数量："+ num + "，现库存量：" + queue.size() + "，暂停消费！");
                try {
                    //消费者等待
                    notEmpty.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            //线程被唤醒或队列不为空
            for (int i = 0; i < num; i++) {
                queue.remove();
            }

            //消费者消费完毕，唤醒生产者
            notFull.signalAll();
            System.out.println("已消费数量："+ num + "，现库存量：" + queue.size());
        }finally {
            lock.unlock();
        }
    }
}
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
测试结果：

已生产数量：30，现库存量：30
已生产数量：40，现库存量：70
要生产的数量：50，现库存量：70，总容量：100，暂停生产！
要生产的数量：60，现库存量：70，总容量：100，暂停生产！
要生产的数量：70，现库存量：70，总容量：100，暂停生产！
已消费数量：30，现库存量：40
已生产数量：50，现库存量：90
已消费数量：50，现库存量：40
已生产数量：60，现库存量：100
要生产的数量：70，现库存量：100，总容量：100，暂停生产！
已消费数量：40，现库存量：60
要生产的数量：70，现库存量：60，总容量：100，暂停生产！
已消费数量：60，现库存量：0
已生产数量：70，现库存量：70
已消费数量：70，现库存量：0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
消费者等待使用notEmpty.await()，而唤醒生产者使用notFull.signalAll()。这样就解决之前使用notifyAll的性能问题。使用notEmpty notFull两个Condition条件可以使得当队列存满时，那么阻塞的肯定是生产者线程，唤醒的肯定是消费者线程，相反，当队列为空时，阻塞的肯定是消费者线程，唤醒的肯定是生产者线程。

2.3 BlockingQueue实现

BlockingQueue是JDK5.0的新增内容，它是一个已经在内部实现了同步的队列，实现方式采用的是await()、signal()方法。它用于阻塞的操作是put()和take()方法。

put()：类似于我们上面的生产者线程，容量达到最大时，自动阻塞。
take()：类似于我们上面的消费者线程，容量为0时，自动阻塞。
实现代码非常简单，跟以往一样，我们只需要重新实现Storage接口即可：

/**
* @Author: cherry
* @Date: Created in 2018/9/27 21:25
*/
public class StorageBlockQueueImpl implements Storage {
    /**
     * 队列最大容量
     */
    public static final int MAX_SIZE = 100;

    /**
     * 阻塞队列
     */
    private LinkedBlockingQueue<Object> queue = new LinkedBlockingQueue<>(MAX_SIZE);

    @Override
    public void produce(int num) {
        for (int i = 0; i < num; i++) {
            try {
                //自动阻塞
                queue.put(new Object());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public void consume(int num) {
        for (int i = 0; i < num; i++) {
            try {
                //自动阻塞
                queue.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
---------------------
【转载】
作者：cherry-peng
原文：https://blog.csdn.net/xsp_happyb ... 417?utm_source=copy

---

作者: 不二晨    时间: 2018-10-16 14:13
奈斯

---

作者: 魔都黑马少年梦    时间: 2018-10-25 14:46

---

欢迎光临 黑马程序员技术交流社区 (http://bbs.itheima.com/)

黑马程序员IT技术论坛 X3.2