Java线程间通信之wait/notify

// 在Java中调用wait方法的正确姿势
synchronized (sharedObject) {
   while (condition) {
      sharedObject.wait(); // 在循环体中调用wait方法：线程会释放对象锁，等待被唤醒
   }
   // 在这里执行一些操作：例如将消息写入队列或者从队列中获取消息
}

// 生产者线程
public class Producer extends Thread {

    private final static Logger logger = LoggerFactory.getLogger(Producer.class);
    private final Queue<Integer> queue;
    private final int maxSize;
    private int messageCount = 1;

    public Producer(Queue<Integer> queue, int maxSize, String threadName) {
        super(threadName);
        this.queue = queue;
        this.maxSize = maxSize;
    }

    @Override
    public void run() {
        while (true) {
            synchronized (queue) {  // 在条件判断之前给共享资源加锁
                while (queue.size() >= maxSize) {
                    try {
                        logger.info("消息队列已满: 生产者线程调用wait方法进入等待状态 ...");
                        queue.wait(); // 在循环体中：使用共享对象来调用wait方法，释放共享资源的锁
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
               
                sleep(1000);  // 让生产者每1秒钟生产一条消息
              
                int messageId = messageCount++;
                logger.info("生产消息: {}", messageId);
                queue.add(messageId);  // 将消息写入队列
                queue.notifyAll();     // 通知消费者线程，对消息进行消费
            }
        }
    }
}


// 消费者线程
public class Consumer extends Thread {

    private final static Logger logger = LoggerFactory.getLogger(Consumer.class);
    private final Queue<Integer> queue;

    public Consumer(Queue<Integer> queue, String threadName) {
        super(threadName);
        this.queue = queue;
    }

    @Override
    public void run() {
        while (true) {
            synchronized (queue) {  // 在条件判断之前给共享资源加锁
                while (queue.isEmpty()) {
                    try {
                        logger.info("消息队列为空: 消费者线程调用wait方法进入等待状态 ...");
                        queue.wait();  // 在循环体中：使用共享对象来调用wait方法，释放共享资源的锁
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                logger.info("消费消息: {}", queue.remove());
                queue.notifyAll();     // 通知生产者线程，可以继续生产消息了
            }
        }
    }
}

// 测试代码
public class Launcher {
    
    public static void main(String[] args) throws Exception {
        Queue<Integer> queue = new LinkedList<>();
        int maxSize = 5;
        new Producer(queue, maxSize, "producer-thread").start();
        new Consumer(queue, "consumer-thread").start();
    }
    
}