1 用synchronized+wait()+notifyAll实现
思路:当生产者达到最大数量的时候,让生产者线程wait阻塞等待,否则继续生产,同时通知消费者进行消费。
当消费者消费到0个的时候,让消费者线程wait阻塞等待,否则继续消费,同时通知生产者进行生产。
public class Container1<T> { final private LinkedList<T> lists = new LinkedList<T>(); final private int MAX = 10;//最多10个元素 private int count = 0; public synchronized void put(T t) { while (lists.size() == MAX) { try { this.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } lists.add(t); ++count; this.notifyAll();//通知消费者线程进行消费 } public synchronized T get() { T t = null; while (lists.size() == 0) { try { this.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } t = lists.removeFirst(); count--; this.notifyAll();//通知生产者进行生产 return t; } public static void main(String[] args) { Container1<String> c = new Container1<>(); //启动消费者线程 for (int i = 0; i < 10; i++) { new Thread(() -> { for (int j = 0; j < 5; j++) { System.out.println(Thread.currentThread().getName() + "get="+c.get()); } }, "c" + i).start(); } try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); } //启动生产者线程 for (int i = 0; i < 2; i++) { new Thread(() -> { for (int j = 0; j < 25; j++) { c.put(Thread.currentThread().getName() + "set=" + j); } }, "p" + i).start(); } } } 这种实现方式有个问题:就是生产者线程调用notifyAll通知的时候,有可能通知到生产者线程,消费者线程调用notifyAll通知的时候有可能通知到消费者线程,所以还可以改进,下面用第二种方式实现,解决这个问题。
2 使用Lock和Condition来实现
在进行锁处理的时候还有一个接口:Condition,这个接口可以由用户来自己进行锁的对象创建。
Condition的作用是对锁进行更精确的控制。
Condition的await()方法相当于Object的wait()方法,Condition的signal()方法相当于Object的notify()方法,Condition的signalAll()方法相当于Object的notifyAll()方法。
不同的是Object的wait(), notify(), notifyAll() 方法是和“同步锁”(synchronized关键字)捆绑使用的;而Condition是需要与“互斥锁/共享锁”捆绑使用。
使用Condition,分别用两个等待队列来实现生产者和消费者队列
public class Container2<T> { final private LinkedList<T> lists = new LinkedList<>(); final private int MAX = 10;//最多10个元素 private int count = 0; private Lock lock = new ReentrantLock(); private Condition producer = lock.newCondition(); private Condition consumber = lock.newCondition(); public void put(T t) { try { lock.lock(); while (lists.size() == MAX) { producer.await(); } lists.add(t); ++count; consumber.signalAll();//通知消费者线程进行消费 } catch (InterruptedException e) { e.printStackTrace(); } finally { lock.unlock(); } } public T get() { T t = null; try { lock.lock(); while (lists.size() == 0) { consumber.await(); } t = lists.removeFirst(); count--; producer.signalAll();//通知生产者进行生产 } catch (InterruptedException e) { e.printStackTrace(); } finally { lock.unlock(); } return t; } public static void main(String[] args) { Container2 c = new Container2(); for (int i = 0; i <10 ; i++) { new Thread(()->{ for (int j = 0; j <5 ; j++) { System.out.println(Thread.currentThread().getName()+"get="+c.get()); } },"c"+i).start(); } try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); } //启动生产者线程 for (int i = 0; i <2 ; i++) { new Thread(()->{ for (int j = 0; j <25 ; j++) { c.put(Thread.currentThread().getName()+"set="+j); } },"p"+i).start(); } } }