produce日志之后，即：同一任务的消费行为一定发生在生产行为之后。缓冲区的容量留给读者验证。符合两个验证条件。
BlockingQueue写法的核心只有两行代码，并发和容量控制都封装在了BlockingQueue中，正确性由BlockingQueue保证。面试中首选该写法，自然美观简单。
实现二：wait && notify
如果不能将并发与容量控制都封装在缓冲区中，就只能由消费者与生产者完成。最简单的方案是使用朴素的wait && notify机制。
public class WaitNotifyModel implements Model {
private final Object BUFFER_LOCK = new Object();
private final Queue<Task> buffer = new LinkedList<>();
private final int cap;
private final AtomicInteger increTaskNo = new AtomicInteger(0);
public WaitNotifyModel(int cap) {
this.cap = cap;
}
@Override
public Runnable newRunnableConsumer() {
return new ConsumerImpl();
}
@Override
public Runnable newRunnableProducer() {
return new ProducerImpl();
}
private class ConsumerImpl extends AbstractConsumer implements Consumer, Runnable {
@Override
public void consume() throws InterruptedException {
synchronized (BUFFER_LOCK) {
while (buffer.size() == 0) {
BUFFER_LOCK.wait();
}
Task task = buffer.poll();
assert task != null;
// 固定时间范围的消费，模拟相对稳定的服务器处理过程
Thread.sleep(500 + (long) (Math.random() * 500));
System.out.println("consume: " + task.no);
BUFFER_LOCK.notifyAll();
}
}
}
private class ProducerImpl extends AbstractProducer implements Producer, Runnable {
@Override
public void produce() throws InterruptedException {
// 不定期生产，模拟随机的用户请求
Thread.sleep((long) (Math.random() * 1000));
synchronized (BUFFER_LOCK) {
while (buffer.size() == cap) {
BUFFER_LOCK.wait();
}
Task task = new Task(increTaskNo.getAndIncrement());
buffer.offer(task);
System.out.println("produce: " + task.no);
BUFFER_LOCK.notifyAll();
}
}
}
public static void main(String[] args) {
Model model = new WaitNotifyModel(3);
for (int i = 0; i < 2; i++) {
new Thread(model.newRunnableConsumer()).start();
}
for (int i = 0; i < 5; i++) {
new Thread(model.newRunnableProducer()).start();
}
}
}
验证方法同上。
朴素的wait && notify机制不那么灵活，但足够简单。synchronized、wait、notifyAll的用法可参考【Java并发编程】之十：使用wait/notify/notifyAll实现线程间通信的几点重要说明，着重理解唤醒与锁竞争的区别。
实现三：简单的Lock && Condition
我们要保证理解wait && notify机制。实现时可以使用Object类提供的wait()方法与notifyAll()方法，但更推荐的方式是使用java.util.concurrent包提供的Lock && Condition。
public class LockConditionModel1 implements Model {
private final Lock BUFFER_LOCK = new ReentrantLock();
private final Condition BUFFER_COND = BUFFER_LOCK.newCondition();
private final Queue<Task> buffer = new LinkedList<>();
private final int cap;
private final AtomicInteger increTaskNo = new AtomicInteger(0);
public LockConditionModel1(int cap) {
this.cap = cap;
}
@Override
public Runnable newRunnableConsumer() {
return new ConsumerImpl();
}
@Override
public Runnable newRunnableProducer() {
return new ProducerImpl();
}
private class ConsumerImpl extends AbstractConsumer implements Consumer, Runnable {
@Override
public void consume() throws InterruptedException {
BUFFER_LOCK.lockInterruptibly();
try {
while (buffer.size() == 0) {
BUFFER_COND.await();
}
Task task = buffer.poll();
assert task != null;
// 固定时间范围的消费，模拟相对稳定的服务器处理过程
Thread.sleep(500 + (long) (Math.random() * 500));
System.out.println("consume: " + task.no);
BUFFER_COND.signalAll();