Java-BlockingQueue阻塞队列

说明

LinkedBlockingQueue 是 Java 并发包 java.util.concurrent 中的一个阻塞队列。它内部使用链表实现，并且是线程安全的

代码示例

package org.jeecg.modules.netty.tcp;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

public class LinkedBlockingQueueExample {
    /**
     * 阻塞方式：put/take
     * 非阻塞方式：offer/poll
     * @param args
     */
    public static void main(String[] args) {
        // 创建一个容量为10的 LinkedBlockingQueue
        LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>(10);

        // 启动生产者线程
        Thread producerThread = new Thread(new Producer(queue));
        producerThread.start();

        try {
            TimeUnit.MILLISECONDS.sleep(5000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        // 启动消费者线程
        Thread consumerThread = new Thread(new Consumer(queue));
        consumerThread.start();
    }

    // 生产者类
    static class Producer implements Runnable {
        private final LinkedBlockingQueue<String> queue;

        Producer(LinkedBlockingQueue<String> queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            try {
                for (int i = 0; i < 20; i++) {
                    String message = "Message-" + i;
                    System.out.println("Producing: " + message);

                    // 阻塞方式：如果队列满了，则此操作将阻塞
                    // queue.put(message);

                    // 非阻塞方式：超时后继续执行
                    queue.offer(message, 1000, TimeUnit.MILLISECONDS);

                    // 模拟生产时间
                    TimeUnit.MILLISECONDS.sleep(1000);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    // 消费者类
    static class Consumer implements Runnable {
        private final LinkedBlockingQueue<String> queue;

        Consumer(LinkedBlockingQueue<String> queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            try {
                while (true) {
                    // 阻塞方式：如果队列为空，则此操作将阻塞
                    // String message = queue.take();

                    // 非阻塞方式：超时后继续执行
                    String message = queue.poll(1000, TimeUnit.MILLISECONDS);
                    System.out.println("Consuming: " + message);

                    // 模拟消费时间
                    TimeUnit.MILLISECONDS.sleep(500);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}