在Java中如何使用ConcurrentLinkedQueue实现无锁队列_ConcurrentLinkedQueue集合操作

P粉602998670

发布时间：2025-11-25 09:51:24

495人浏览过

来源于php中文网

原创

ConcurrentLinkedQueue是Java中基于链表的无锁线程安全队列，使用CAS实现高并发性能，遵循FIFO原则，支持高效入队（offer/add）、出队（poll/peek）操作，适用于非阻塞、高吞吐场景，但size()不精确、不支持null元素和阻塞操作，迭代器为弱一致性，适合生产者-消费者模型。

在java中如何使用concurrentlinkedqueue实现无锁队列_concurrentlinkedqueue集合操作

在Java中，ConcurrentLinkedQueue 是一个基于链表结构的线程安全队列，它使用无锁（lock-free）算法实现高并发环境下的高效操作。这个类位于 java.util.concurrent 包下，适用于多线程环境中需要高性能队列的场景。

ConcurrentLinkedQueue 的基本特性

ConcurrentLinkedQueue 实现了 Queue 接口，遵循 FIFO（先进先出）原则。它的主要特点包括：

线程安全：内部通过 CAS（Compare-And-Swap）操作保证线程安全，无需显式加锁。
无界队列：容量不受限，理论上可无限添加元素（受限于内存）。
弱一致性迭代器：遍历时不保证反映最新的修改，适合非严格一致性的场景。
高性能：在高并发读写场景下性能优于 synchronized 队列或 BlockingQueue（如 ArrayBlockingQueue）。

常用集合操作方法

以下是 ConcurrentLinkedQueue 提供的核心操作方法及使用示例：

1. 添加元素（入队）

add(E e) 或 offer(E e)：将元素插入队尾，成功返回 true。两者在此类中效果相同。

示例：

立即学习“Java免费学习笔记（深入）”；

Copy Leaks

AI内容检测和分级，帮助创建和保护原创内容

下载

ConcurrentLinkedQueue queue = new ConcurrentLinkedQueue<>();
queue.offer("task1");
queue.add("task2");

2. 获取并移除元素（出队）

poll()：获取并移除队首元素，队列为空时返回 null。
remove()：与 poll 类似，但队列为空时抛出 NoSuchElementException。

推荐使用 poll() 避免异常处理。

String task = queue.poll(); // 安全获取
if (task != null) {
    System.out.println("处理任务：" + task);
}

3. 查看队首元素（不出队）

peek()：返回队首元素但不移除，队列为空时返回 null。

String head = queue.peek();
System.out.println("当前队首：" + head);

4. 查询队列状态

size()：返回当前队列元素数量。注意：该方法需遍历整个链表，高并发下可能不精确且性能较差，应避免频繁调用。
isEmpty()：判断队列是否为空，推荐用于空检查。

if (!queue.isEmpty()) {
    System.out.println("队列中有 " + queue.size() + " 个元素");
}

多线程环境下的使用示例

以下是一个生产者-消费者模型示例，展示 ConcurrentLinkedQueue 在并发环境中的应用：

import java.util.concurrent.ConcurrentLinkedQueue;

public class CLQExample {
    private static final ConcurrentLinkedQueue queue = new ConcurrentLinkedQueue<>();

    public static void main(String[] args) {
        // 启动生产者线程
        Thread producer = new Thread(() -> {
            for (int i = 1; i <= 5; i++) {
                String task = "任务-" + i;
                queue.offer(task);
                System.out.println("生产：" + task);
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        });

        // 启动消费者线程
        Thread consumer = new Thread(() -> {
            while (true) {
                String task = queue.poll();
                if (task != null) {
                    System.out.println("消费：" + task);
                } else if (Thread.currentThread().isInterrupted()) {
                    break; // 生产结束且队列为空
                }
                // 小休避免过度占用CPU
                try {
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        });

        producer.start();
        consumer.start();

        try {
            producer.join();
            consumer.interrupt();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}