java中使用LinkedBlockingQueue和异步线程消费该队列

LinkedBlockingQueue是Java集合框架中的一种阻塞队列，它支持多线程并发操作，常用于生产者和消费者模式中。使用LinkedBlockingQueue可以保证生产者和消费者之间的数据传输是线程安全的。

异步线程消费该队列的具体实现方式如下：

1. 创建一个LinkedBlockingQueue对象，指定队列容量。

LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>(100);

2. 创建一个生产者线程，向队列中不断添加数据。

Thread producer = new Thread(() -> {
    while (true) {
        try {
            String data = produceData();
            queue.put(data);
            System.out.println("Produced data: " + data);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
});
producer.start();

3. 创建一个消费者线程池，使用异步线程消费队列中的数据。

ExecutorService executorService = Executors.newFixedThreadPool(2);
for (int i = 0; i < 2; i++) {
    executorService.execute(() -> {
        while (true) {
            try {
                String data = queue.take();
                consumeData(data);
                System.out.println("Consumed data: " + data);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    });
}

在上述代码中，我们创建了一个固定大小为2的线程池，用于异步消费队列中的数据。每个线程都不断从队列中取出数据，并进行消费。由于LinkedBlockingQueue是阻塞队列，如果队列为空，消费线程将会阻塞等待，直到队列中有新的数据。

完整代码如下：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class LinkedBlockingQueueExample {

    public static void main(String[] args) {
        LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>(100);

        Thread producer = new Thread(() -> {
            while (true) {
                try {
                    String data = produceData();
                    queue.put(data);
                    System.out.println("Produced data: " + data);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        producer.start();

        ExecutorService executorService = Executors.newFixedThreadPool(2);
        for (int i = 0; i < 2; i++) {
            executorService.execute(() -> {
                while (true) {
                    try {
                        String data = queue.take();
                        consumeData(data);
                        System.out.println("Consumed data: " + data);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

    private static String produceData() {
        return "data-" + System.currentTimeMillis();
    }

    private static void consumeData(String data) {
        // do something with the data
    }
}