实现生产者和消费者

1. wait() / notifyAll()

  1. public class ProducerConsumer {
  2.     private static class Producer implements Runnable {
  3.         private List<Integer> list;
  4.         private int capacity;
  5.         public Producer(List list, int capacity) {
  6.             this.list = list;
  7.             this.capacity = capacity;
  8.         }
  10.         @Override
  11.         public void run() {
  12.             while (true) {
  13.                 synchronized (list) {
  14.                     try {
  15.                         String producer = Thread.currentThread().getName();
  16.                         while (list.size() == capacity) {
  17.                             System.out.println("生产者 " + producer + ":list 已达到最大容量，进行wait");
  18.                             list.wait();
  19.                             System.out.println("生产者 " + producer + ":退出 wait");
  20.                         }
  21.                         Random random = new Random();
  22.                         int i = random.nextInt();
  23.                         System.out.println("生产者 " + producer + ":生产数据" + i);
  24.                         list.add(i);
  25.                         list.notifyAll();
  26.                     } catch (InterruptedException e) {
  27.                         e.printStackTrace();
  28.                     }
  29.                 }
  30.             }
  31.         }
  32.     }
  34.     private static class Consumer implements Runnable {
  35.         private List<Integer> list;
  36.         public Consumer(List list) {
  37.             this.list = list;
  38.         }
  40.         @Override
  41.         public void run() {
  42.             while (true) {
  43.                 synchronized (list) {
  44.                     try {
  45.                         String consumer = Thread.currentThread().getName();
  46.                         while (list.isEmpty()) {
  47.                             System.out.println("消费者 " + consumer + ":list 为空，进行 wait");
  48.                             list.wait();
  49.                             System.out.println("消费者 " + consumer + ":退出wait");
  50.                         }
  51.                         Integer element = list.remove(0);
  52.                         System.out.println("消费者 " + consumer + ":消费数据：" + element);
  53.                         list.notifyAll();
  54.                     } catch (InterruptedException e) {
  55.                         e.printStackTrace();
  56.                     }
  57.                 }
  58.             }
  59.         }
  60.     }
  62.     public static void main(String[] args) {
  63.         final LinkedList linkedList = new LinkedList();
  64.         final int capacity = 5;
  65.         ThreadPoolExecutor executor = new ThreadPoolExecutor(
  66.                 5, 10, 1, TimeUnit.SECONDS,
  67.                 new ArrayBlockingQueue<>(100), new ThreadPoolExecutor.CallerRunsPolicy());
  68.         executor.execute(new Producer(linkedList, capacity));
  69.         executor.execute(new Consumer(linkedList));
  70.         executor.shutdown();
  71.     }
  72. }

2. await() / sigalAll()

  1. public class ProducerConsumer {
  2.     private static ReentrantLock lock = new ReentrantLock();
  3.     private static Condition full = lock.newCondition();
  4.     private static Condition empty = lock.newCondition();
  6.     private static class Producer implements Runnable{
  7.         private List<Integer> list;
  8.         private int capacity;
  9.         public Producer(List list, int capacity) {
  10.             this.list = list;
  11.             this.capacity = capacity;
  12.         }
  13.         @Override
  14.         public void run() {
  15.             while (true){
  16.                 lock.lock();
  17.                 try {
  18.                     String producer = Thread.currentThread().getName();
  19.                     while (list.size() == capacity){
  20.                         System.out.println("生产者 " + producer + ":list 已达到最大容量，进行wait");
  21.                         full.await();
  22.                         System.out.println("生产者 " + producer + ":退出 wait");
  23.                     }
  24.                     Random random = new Random();
  25.                     int i = random.nextInt();
  26.                     System.out.println("生产者 " + producer + ":生产数据" + i);
  27.                     list.add(i);
  28.                     empty.signalAll();
  29.                 } catch (InterruptedException e) {
  30.                     e.printStackTrace();
  31.                 } finally {
  32.                     lock.unlock();
  33.                 }
  34.             }
  35.         }
  36.     }
  37.     private static class Consumer implements Runnable{
  38.         private List<Integer> list;
  39.         public Consumer(List list) {
  40.             this.list = list;
  41.         }
  42.         @Override
  43.         public void run() {
  44.             while (true) {
  45.                 lock.lock();
  46.                 try {
  47.                     String consumer = Thread.currentThread().getName();
  48.                     while (list.isEmpty()) {
  49.                         System.out.println("消费者 " + consumer + ":list 为空，进行 wait");
  50.                         empty.await();
  51.                         System.out.println("消费者 " + consumer + ":退出wait");
  52.                     }
  53.                     Integer element = list.remove(0);
  54.                     System.out.println("消费者 " + consumer + ":消费数据：" + element);
  55.                     full.signalAll();
  56.                 } catch (InterruptedException e) {
  57.                     e.printStackTrace();
  58.                 } finally {
  59.                     lock.unlock();
  60.                 }
  61.             }
  62.         }
  63.     }
  65.     public static void main(String[] args) {
  66.         final LinkedList linkedList = new LinkedList();
  67.         final int capacity = 5;
  68.         ThreadPoolExecutor executor = new ThreadPoolExecutor(
  69.                 5, 10, 1, TimeUnit.SECONDS,
  70.                 new ArrayBlockingQueue<>(100), new ThreadPoolExecutor.CallerRunsPolicy());
  71.         executor.execute(new Producer(linkedList,capacity));
  72.         executor.execute(new Consumer(linkedList));
  73.     }
  74. }

3. 阻塞队列

  1. public class ProducerConsumer {
  2.     private static class Producer implements Runnable {
  3.         private BlockingQueue<Integer> queue;
  4.         public Producer(BlockingQueue<Integer> queue) {
  5.             this.queue = queue;
  6.         }
  7.         @Override
  8.         public void run() {
  9.             while (true) {
  10.                 try {
  11.                     String producer = Thread.currentThread().getName();
  12.                     Random random = new Random();
  13.                     int i = random.nextInt();
  14.                     System.out.println("生产者 " + producer + ":生产数据" + i);
  15.                     queue.put(i);
  16.                 } catch (InterruptedException e) {
  17.                     e.printStackTrace();
  18.                 }
  19.             }
  20.         }
  21.     }
  22.     private static class Consumer implements Runnable {
  23.         private BlockingQueue<Integer> queue;
  24.         public Consumer(BlockingQueue<Integer> queue) {
  25.             this.queue = queue;
  26.         }
  27.         @Override
  28.         public void run() {
  29.             while (true) {
  30.                 try {
  31.                     String consumer = Thread.currentThread().getName();
  32.                     Integer element = queue.take();
  33.                     System.out.println("消费者 " + consumer + ":消费数据：" + element);
  35.                 } catch (InterruptedException e) {
  36.                     e.printStackTrace();
  37.                 }
  38.             }
  39.         }
  40.     }
  41.     public static void main(String[] args) {
  42.         final BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();
  43.         ThreadPoolExecutor executor = new ThreadPoolExecutor(
  44.                 5, 10, 1, TimeUnit.SECONDS,
  45.                 new ArrayBlockingQueue<>(100), new ThreadPoolExecutor.CallerRunsPolicy());
  46.         executor.execute(new Producer(queue));
  47.         executor.execute(new Consumer(queue));
  48.         executor.shutdown();
  49.     }
  50. }