Java 生产者消费者模型的三种实现过程

java

生产者一边在生产,消费者一边消耗。当库存满的时候生产者暂停生产,直到有空位;当库存空的时候消费者暂停消费,直到有产品。

关键点:

  1. 生产者和消费者都是在不断生产和消费的,是同时并发的,不应该等满了再消费 / 空了再生产。
  2. 前提条件是生产速度和消费速度不同。

设计方法:

  1. 生产者和消费者分别使用一个线程模拟。将其抽象分别继承Runnable接口,生产和消费的行为放在Runnable的run()方法中.
  2. 控制生产和消费的速度不同。可以使每次生产 / 每次消费 之间的间隔不同,达到 生产比消费快 或者 消费比生产快 的效果.
  3. 生产者拿到仓库锁,判断是否还有空位

    • 如果没有空位,就让自己先别再去试图抢占锁了(将自己放入等待池,wait());
    • 如果有空位,就生产一个,然后唤醒(notify())等待池中的线程(包括先前因为仓库空而wait()掉的线程)来竞争锁。

  4. 消费者拿到仓库锁,判断是否还有可以消费的商品

    • 如果没有商品了,就让自己先别再去试图抢占锁了(将自己放入等待池,wait());
    • 如果有空位,就消费一个,然后唤醒(notify())等待池中的线程(包括先前因为仓库满而wait()掉的线程)来竞争锁。

  5. 当某个生产者线程拿到锁,发现仓库满了,会wait()自己,而不notify()(防止又唤醒一个生产者线程),直到有一个消费者线程拿到锁,消费后,才notify()等待池中的生产者线程,才继续生产。(消费者同理)

1.使用synchronized 关键字

public class ProducerCustomerModel {

static int MAX_SIZE = 10;

static List<Integer> list;

public static void main(String[] args) {

list = new ArrayList<>();

new Thread(new Producer()).start();

new Thread(new Customer()).start();

}

private static class Producer implements Runnable{

@Override

public void run() {

while(true){

try {

Thread.sleep(250); // 生产效率:每250ms一个

} catch (InterruptedException e) {

e.printStackTrace();

}

synchronized (list){

// 第一步:拿到锁

while(list.size() >= MAX_SIZE) {

// 第二步,判断能不能生产。

// 如果发现库存满了,就进入等待池,放弃锁,线程暂停。

try {

list.wait();

} catch (InterruptedException e) {

e.printStackTrace();

}

}

// 如果有,生产一个

list.add(1);

System.out.println("生产了一个产品,现在有"+list.size()+"个产品。");

// 唤醒消费线程,可以试图消费了。

list.notifyAll();

}

}

}

}

private static class Customer implements Runnable{

@Override

public void run() {

while(true){

try {

Thread.sleep(500); // 消费效率:每500ms一个

} catch (InterruptedException e) {

e.printStackTrace();

}

synchronized (list){

// 第一步:拿到锁

while(list.size()==0){

// 第二步,判断能不能消费。

// 如果发现没有商品了,就进入等待池,放弃锁,线程暂停。

try {

list.wait();

} catch (InterruptedException e) {

e.printStackTrace();

}

}

// 如果有,消费一个

list.remove(0);

System.out.println("消费了一个产品,现在有"+list.size()+"个产品。");

// 唤醒生产线程,可以试图生产了。

list.notifyAll();

}

}

}

}

}

2.使用ReenterantLock

ReenterantLock与synchronized类似,synchronized在语句块结束后会自动释放锁,ReenterantLock需要unlock()。ReenterantLock还支持公平锁(等待最久的先拿锁)。

需要设置两个条件,分别用于空/满时候的信号。

public class ProducerCustomerModelReentrantLock {

static ReentrantLock lock = new ReentrantLock(true);

static int MAX_SIZE = 10;

static List<Integer> list;

static Condition ableToProduce,ableToCustom;

public static void main(String[] args) {

list = new ArrayList<>();

ableToCustom = lock.newCondition();

ableToProduce = lock.newCondition();

new Thread(new Producer()).start();

new Thread(new Customer()).start();

}

private static class Producer implements Runnable{

@Override

public void run() {

while(true) {

try {

Thread.sleep(250);

} catch (InterruptedException e) {

e.printStackTrace();

}

lock.lock();

while(list.size()>=MAX_SIZE) {

try {

ableToProduce.await();

} catch (InterruptedException e) {

e.printStackTrace();

}

}

list.add(1);

System.out.println("生产了一个产品,现在有"+list.size()+"个产品。");

ableToCustom.signal();

lock.unlock();

}

}

}

private static class Customer implements Runnable{

@Override

public void run() {

while(true) {

try {

Thread.sleep(500);

} catch (InterruptedException e) {

e.printStackTrace();

}

lock.lock();

while(list.size()==0) {

try {

ableToCustom.await();

} catch (InterruptedException e) {

e.printStackTrace();

}

}

list.remove(0);

System.out.println("消费了一个产品,现在有"+list.size()+"个产品。");

ableToProduce.signal();

lock.unlock();

}

}

}

}

3.使用BlockingQueue

阻塞队列本身就是当队满的时候阻塞线程,故直接使用就可以实现生产者/消费者模型。

阻塞队列内部也是使用ReenterantLock实现的。

public class ProducerCustomerModelUsingBlockedQueue {

static BlockingQueue<Integer> queue;

static int MAX_SIZE = 10;

public static void main(String[] args) {

queue = new ArrayBlockingQueue<>(MAX_SIZE);

new Thread(new Producer()).start();

new Thread(new Customer()).start();

}

private static class Producer implements Runnable{

@Override

public void run() {

while(true){

try {

Thread.sleep(300);

} catch (InterruptedException e) {

e.printStackTrace();

}

try {

queue.put(1);

System.out.println("生产了一个产品,现在有"+queue.size()+"个产品。");

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

}

private static class Customer implements Runnable{

@Override

public void run() {

while(true){

try {

Thread.sleep(400);

} catch (InterruptedException e) {

e.printStackTrace();

}

try {

queue.take();

System.out.println("消费了一个产品,现在有"+queue.size()+"个产品。");

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

}

}

以上是 Java 生产者消费者模型的三种实现过程 的全部内容, 来源链接: utcz.com/z/390631.html

回到顶部