锁和多线程:生产者与消费者3种实现(四)

生产者 消费者 Condition BlockingDeque 

 

• 1．wait notifyAll
• 2．ReentrantLock Condition
• 3．BlockingDeque

 

搞明白 线程 锁和多线程系列

1．wait notifyAll

首先使用最传统的wait notifyAll synchronized方式,生产食物后唤醒消费者,食物充足时就等待,不生产食物.

• wait() 让线程等待
• notifyAll 唤醒等待线程

package com.lyf.procon;

/**
 * @Author lyf
 * @Date 2018/11/18 0018 21:09
 */
public class Food01 {
  private static int num = 0; // 食物数量
  
  public synchronized void add() {
    if (num < 1) {// 判断食物是否充足
      num++;// 生产食物
      notifyAll();// 唤醒消费者
    } else {
      try {
        wait();// 生产者等待
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }

  public synchronized void remove() {
    if (num > 0) {// 判断食物是否充足
      num--; // 消费食物
      notifyAll();// 唤醒生产者
    } else {
      try {
        wait();// 消费者等待
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }

}

class Producer01 extends Thread{
  private Food01 food01;

  public Producer01(Food01 proCon1) {
    this.food01 = proCon1;
  }

  @Override
  public void run() {
    while (true){
      food01.add();
      System.out.println(Thread.currentThread().getName() + "生产食物...");
    }
  }
}

class Consumer01 extends Thread{
  private Food01 food01;

  public Consumer01(Food01 proCon1) {
    this.food01 = proCon1;
  }

  @Override
  public void run() {
    while (true){
      food01.remove();
      System.out.println(Thread.currentThread().getName() + "消费食物...");
    }
  }
}

class Test01{
  public static void main(String[] args) throws InterruptedException {
    Food01 food01 = new Food01();
    for (int i = 0; i < 3; i++) {
      new Producer01(food01).start();
      new Consumer01(food01).start();
    }
    
    Thread.sleep(10);
    System.exit(0);// 强制终止程序
  }
}

2．ReentrantLock

使用ReentrantLock重入锁和Condition条件实现,创建重入锁lock变量.然后,创建生产者的触发条件producer和消费者触发条件consumer.调用

• consumer.signalAll()唤醒消费者
• consumer.await()让消费者等待
• producer.signalAll()唤醒生产者
• producer.await()让生产者等待

package com.lyf.procon;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Author lyf
 * @Date 2018/11/18 0018 21:09
 */
public class Food02 {
  private static int num = 0; // 食物数量
  
  private ReentrantLock lock;
  private Condition producer;// 生产者状态
  private Condition consumer;// 消费者状态

  public Food02() {
    lock = new ReentrantLock();
    producer = lock.newCondition();
    consumer = lock.newCondition();
  }

  public void add() {
    lock.lock();// 获得锁
    try {
      if (num < 1) {
        num++;
        consumer.signalAll();// 通知消费者
      }else{
        producer.await();// 生产者等待
      }
    } catch (InterruptedException e) {
      e.printStackTrace();
    } finally {
      lock.unlock();// 释放锁
    }
  }

  public void remove() {
    lock.lock();// 获得锁
    try {
      if (num > 0) {
        num--;
        producer.signalAll();// 通知生产者
      }else{
        consumer.await();// 消费者等待
      }
    } catch (InterruptedException e) {
      e.printStackTrace();
    } finally {
      lock.unlock();// 释放锁
    }
  }
}

class Producer02 extends Thread{
  private Food02 food02;

  public Producer02(Food02 proCon1) {
    this.food02 = proCon1;
  }

  @Override
  public void run() {
    while (true){
      food02.add();
      System.out.println(Thread.currentThread().getName() + "生产食物...");
    }
  }
}

class Consumer02 extends Thread{
  private Food02 food02;

  public Consumer02(Food02 proCon1) {
    this.food02 = proCon1;
  }

  @Override
  public void run() {
    while (true){
      food02.remove();
      System.out.println(Thread.currentThread().getName() + "消费食物...");
    }
  }
}

class Test02{
  public static void main(String[] args) throws InterruptedException {
    Food02 food02 = new Food02();
    for (int i = 0; i < 3; i++) {
      new Producer02(food02).start();
      new Consumer02(food02).start();
    }
    
    Thread.sleep(10);
    System.exit(0);// 强制终止程序
  }
}

3．BlockingDeque

使用阻塞队列BlockingDeque实现,调用

• put()在队列尾部添加元素
• take()在队列头部删除元素

package com.lyf.procon;

import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * @Author lyf
 * @Date 2018/11/18 0018 21:09
 */
public class Food03 {
  // 创建阻塞队列
  private BlockingDeque<Integer> blockingDeque = new LinkedBlockingDeque<>(10);

  public void add() {
    try {
      blockingDeque.put(1);// 在队列尾部添加元素
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }

  public void remove() {
    try {
      blockingDeque.take();// 在队列头部删除元素
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }

}

class Producer03 extends Thread{
  private Food03 food03;

  public Producer03(Food03 proCon1) {
    this.food03 = proCon1;
  }

  @Override
  public void run() {
    while (true){
      food03.add();
      System.out.println(Thread.currentThread().getName() + "生产食物...");
    }
  }
}

class Consumer03 extends Thread{
  private Food03 food03;

  public Consumer03(Food03 proCon1) {
    this.food03 = proCon1;
  }

  @Override
  public void run() {
    while (true){
      food03.remove();
      System.out.println(Thread.currentThread().getName() + "消费食物...");
    }
  }
}

class Test03{
  public static void main(String[] args) throws InterruptedException {
    Food03 food03 = new Food03();
    for (int i = 0; i < 3; i++) {
      new Producer03(food03).start();
      new Consumer03(food03).start();
    }
    
    Thread.sleep(10);
    System.exit(0);// 强制终止程序
  }
}

参考资料

• https://www.cnblogs.com/fankongkong/p/7339848.html