核心思想:
- 组成:1、生产者;2、消费者;3、主题队列;
- 互助:生产者生产主题加入主题队列;消费者从主题队列中取出主题,消费主题;
- 互斥:生产者和消费者都持有主题队列,当队列中主题数达到队列上限,生产者停止生产,等待消费者消费;当队列中没有主题,消费者停止消费,等待生产者生产。
wait()和notifyAll()
wait()
/ nofity()
方法是基类Object
的两个方法,也就意味着所有Java类都会拥有这两个方法,我们可以为任何对象实现同步机制。
-
wait()
:当主题队列已满/空时,生产者/消费者线程停止自己的执行,放弃锁,使自己处于等待状态,让其他线程执行。 -
notify()
:当生产者/消费者向生产队列放入/取出一个产品后,向其他等待的线程发出可执行的通知,同时放弃锁,使自己处于等待状态。
public class ProducerAndConsumerByObject {
public static void main(String[] args) {
ProducerAndConsumerByObject producerAndConsumerByObject = new ProducerAndConsumerByObject();
}
private static final int MAX_SIZE = 10;
final LinkedList<Integer> subjects = new LinkedList<>();
ProducerAndConsumerByObject() {
doExample();
}
private void doExample() {
Subject subject = new Subject();
Producer p1 = new Producer(subject, "P1");
Producer p2 = new Producer(subject, "P2");
Producer p3 = new Producer(subject, "P3");
Customer c1 = new Customer(subject, "C1");
Customer c2 = new Customer(subject, "C2");
p1.start();
p2.start();
p3.start();
c1.start();
c2.start();
}
public class Subject {
private int i;
Subject() {
}
// 生产者生产主题,加入队列
void produce(String name) {
while(true) {
synchronized (subjects) {
while (subjects.size() == MAX_SIZE) {
try {
subjects.wait();
System.out.println("Queue is full, Producer wait");
} catch (Exception ex) {
ex.printStackTrace();
}
}
System.out.println(name + " Producer produces " + i);
subjects.offer(i++);
subjects.notifyAll();
}
try {
Thread.sleep(100);
} catch (Exception ex) {
ex.printStackTrace();
}
}
}
// 消费者消费主题,取出队列
void customer(String name) {
while(true) {
synchronized (subjects) {
while (subjects.isEmpty()) {
try {
subjects.wait();
System.out.println("Queue is empty, Customer wait");
} catch (Exception ex) {
ex.printStackTrace();
}
}
Integer i = subjects.poll();
subjects.notifyAll();
System.out.println(name + " Customer cust " + i);
}
try {
Thread.sleep(100);
} catch (Exception ex) {
ex.printStackTrace();
}
}
}
}
// 生产者,独立线程,在run方法中生产
public class Producer extends Thread {
private Subject subject;
private String name;
Producer(Subject subject, String name) {
this.subject = subject;
this.name = name;
}
@Override
public void run() {
System.out.println(name + " is running");
subject.produce(name);
}
}
// 消费者,独立线程,在run方法中消费
public class Customer extends Thread {
private Subject subject;
private String name;
Customer(Subject subject, String name) {
this.subject = subject;
this.name = name;
}
@Override
public void run() {
System.out.println(name + " is running");
subject.customer(name);
}
}
}
await()和signalAll()
上面的方式实现进程间同步依赖于synchronized
,在简单场景下可以满足要求,但是在复杂场景下,对于锁需要更加细致的操作(条件满足,及时释放锁),因此需要用到lock
.
await()
和signalAll()
是lock
中实现同步的一种方式,基本和上述的wait()
和notifyAll()
方法相同。通过lock.newCondition()
获取condition
,通过condition
的await()
和signlaAll()
实现同步。
public class ProducerConsumerByLock {
public static void main(String[] args) {
ProducerConsumerByLock producerConsumerByLock = new ProducerConsumerByLock();
}
private static final int MAX_SIZE = 5;
private static final Lock lock = new ReentrantLock();
private static final Condition emptyCondition = lock.newCondition();
private static final Condition fullCondition = lock.newCondition();
private Queue<Integer> subject = new LinkedList<>();
ProducerConsumerByLock() {
doExample();
}
private void doExample() {
Subject subject = new Subject();
Producer p1 = new Producer(subject, "P1");
Producer p2 = new Producer(subject, "P2");
Producer p3 = new Producer(subject, "P3");
Consumer c1 = new Consumer(subject, "C1");
Consumer c2 = new Consumer(subject, "C2");
p1.start();
p2.start();
p3.start();
c1.start();
c2.start();
}
public class Subject {
int i = 0;
public Subject(){
}
public void produce(String name) {
while(true) {
lock.lock();
while (subject.size() == MAX_SIZE) {
try {
fullCondition.await();
System.out.println( name + " Queue is full");
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
System.out.println(name + " produce " + i);
subject.offer(i++);
fullCondition.signalAll();
emptyCondition.signalAll();
lock.unlock();
try {
Thread.sleep(100);
} catch (Exception ex) {
ex.printStackTrace();
}
}
}
public void consume(String name) {
while (true) {
lock.lock();
while (subject.size() == 0) {
try {
emptyCondition.await();
System.out.println(name + " Queue is empty");
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
int x = subject.poll();
System.out.println(name + " consume " + x);
fullCondition.signalAll();
emptyCondition.signalAll();
lock.unlock();
try {
Thread.sleep(100);
} catch (Exception ex) {
ex.printStackTrace();
}
}
}
}
public class Producer extends Thread {
private Subject subject;
private String name;
Producer(Subject subject, String name) {
this.subject = subject;
this.name = name;
}
@Override
public void run() {
System.out.println(name + " is running");
subject.produce(name);
}
}
public class Consumer extends Thread {
private Subject subject;
private String name;
Consumer(Subject subject, String name) {
this.subject = subject;
this.name = name;
}
@Override
public void run() {
System.out.println(name + " is running");
subject.consume(name);
}
}
}
BlockingQueue
LinkedBlockingQueue
是一个已经在内部实现了同步的队列,它可以在生成对象时指定容量大小。它利用await()
和signalAll()
实现阻塞的入队出队操作,具体的方法为:
-
put()
方法:容量达到最大时,自动阻塞,生产者。 -
take()
方法:容量为0时,自动阻塞,消费者。
public class ProducerConsumerByBlockingQueue {
public static void main(String[] args) {
ProducerConsumerByBlockingQueue producerConsumerByBlockingQueue = new ProducerConsumerByBlockingQueue();
}
private static final int MAX_SIZE = 5;
private LinkedBlockingQueue<Integer> subject = new LinkedBlockingQueue<>(MAX_SIZE);
ProducerConsumerByBlockingQueue() {
doExample();
}
private void doExample() {
Subject subject = new Subject();
Producer p1 = new Producer(subject, "P1");
Producer p2 = new Producer(subject, "P2");
Producer p3 = new Producer(subject, "P3");
Consumer c1 = new Consumer(subject, "C1");
Consumer c2 = new Consumer(subject, "C2");
p1.start();
p2.start();
p3.start();
c1.start();
c2.start();
}
public class Subject {
int i = 0;
public Subject(){
}
public void produce(String name) {
while(true) {
try {
System.out.println(name + " produce " + i + " with size = " + subject.size());
subject.put(i++);
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
Thread.sleep(100);
} catch (Exception ex) {
ex.printStackTrace();
}
}
}
public void consume(String name) {
while (true) {
try {
int x = subject.take();
System.out.println(name + " consumer " + x + " with size = " + subject.size());
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
Thread.sleep(100);
} catch (Exception ex) {
ex.printStackTrace();
}
}
}
}
public class Producer extends Thread {
private Subject subject;
private String name;
Producer(Subject subject, String name) {
this.subject = subject;
this.name = name;
}
@Override
public void run() {
System.out.println(name + "is running");
subject.produce(name);
}
}
public class Consumer extends Thread {
private Subject subject;
private String name;
Consumer(Subject subject, String name) {
this.subject = subject;
this.name = name;
}
@Override
public void run() {
System.out.println(name + "is running");
subject.consume(name);
}
}
}
Tips
需要注意到,生产者Producer
和消费者Consumer
在上述三种实现方式中完全一样,不同的只是Subject
,这就做的好处:
- 使得生产者和消费者只有生产和消费的功能,而不需要知道如何生产和消费,提高了生产者和消费者的复用率(可以将
Subject
抽象化,传入不同的Subject
子类) - 使得
Subject
的逻辑变得完整,增加代码可读性,如何生产出,如何消费掉,一目了然。