Java 多线程——生产者与消费者

Java 多线程——生产者与消费者

生产者消费者问题是一个典型的线程同步问题，需要满足以下条件

生产者将物品放置到一个队列中，消费者按顺序取走

当队列满时，生产者需要停止生产，当队列为空时，消费者只能等待

当生产者向队列中加入产品时，需要通知消费者。当消费者取走物品时，需要通知生产者

原理

入队操作

synchronized(this){
while(!cond){
wait()
}
add(e)
notifyAll()
}

出队操作

synchronized(this){
while(!cond){
wait()
}
poll()
notifyAll()
}

注：此处的判断条件不能使用

if

，避免线程被唤醒时不满足条件。故线程被唤醒后需要
再次检查条件，看是否满足条件。当存在多个消费者线程和生产者线程时，线程被唤醒时
就可能不满足条件。读者可以自己尝试，多个消费者和生产者时使用

if

进行判断，会出现什么现象

Java实现

下面看一下Java代码实现

Factory.java

使用了Java的ArrayDeque 实现同步队列

package com.hive;

import java.util.ArrayDeque;
import java.util.Queue;

/**
* Created by albert on 16-12-11.
*/
public class Factory {
private final static int MAX_SIZE = 10;
private Queue<String> pool;

public Factory() {
pool = new ArrayDeque<>();
}

synchronized public String poll() {
while (pool.size() == 0) {    // 此处不应用 if判断，避免被错误唤醒
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
notifyAll(); // 通知生产者线程
return pool.poll();
}

synchronized public void add(String e) {
while (pool.size() > MAX_SIZE) {   // 此处不应用 if判断，避免被错误唤醒
try {
wait();
} catch (InterruptedException e1) {
e1.printStackTrace();
}
}
pool.add(e);
notifyAll(); // 通知消费者线程
}
}

Producer.java

生产者类

package com.hive;

import java.lang.String;

/**
* Created by albert on 16-12-11.
*/
public class Producer implements Runnable {
private String name;
private Factory goods;

public Producer(String name, Factory goods) {
this.name = name;
this.goods = goods;
}

@Override
public void run() {
while (true) {
for (int i = 0; i < 20; i++) {
System.out.println(name + " yield " + i);
goods.add(name + ">" + i);
try {
Thread.sleep(1 * 100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}

Customer.java

消费者类

package com.hive;

/**
* Created by albert on 16-12-11.
*/
public class Customer implements Runnable {
private String name;
private Factory goods;

public Customer(String name, Factory goods) {
this.name = name;
this.goods = goods;
}

@Override
public void run() {
while (true) {
System.out.println(name + " use " + goods.poll());
try {
Thread.sleep(10 * 100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}

Main.java

测试方法

package com.hive;

public class Main {

public static void main(String[] args) {
Factory goods = new Factory();
Producer p1 = new Producer("P1", goods);
Producer p2 = new Producer("P2", goods);
Customer c1 = new Customer("C1", goods);
Customer c2 = new Customer("C2", goods);
new Thread(p1).start();
new Thread(p2).start();
new Thread(c1).start();
new Thread(c2).start();
}
}

##　运行结果

P1 yield 0
P2 yield 0
C1 use P1>0
C2 use P2>0
P1 yield 1
P2 yield 1
P1 yield 2
P2 yield 2
P1 yield 3
P2 yield 3
P1 yield 4
P2 yield 4
P1 yield 5
P2 yield 5
P1 yield 6
P2 yield 6
P2 yield 7
C1 use P1>1
C2 use P2>1
P2 yield 8
P1 yield 7
C1 use P1>2
C2 use P2>2
P1 yield 8
P2 yield 9
C1 use P1>3
C2 use P2>3
P2 yield 10
P1 yield 9
C2 use P2>4
C1 use P1>4
P1 yield 10
P2 yield 11
C2 use P1>5
C1 use P2>5
P1 yield 11
P2 yield 12
C2 use P2>6
C1 use P2>7
P1 yield 12
P2 yield 13
C2 use P1>6
C1 use P1>7
P2 yield 14
P1 yield 13
C2 use P2>8
C1 use P2>9