Java线程安全的队列实现

Java线程安全的队列实现

线程安全队列的重要性

在多线程编程环境中，数据的共享和同步是关键问题。队列作为一种常用的数据结构，经常用于在不同线程之间传递数据。如果队列不是线程安全的，多个线程同时访问和修改队列时可能会导致数据不一致、竞态条件等问题，进而引发程序的不稳定和错误。线程安全的队列能够确保在多线程环境下，对队列的操作是原子性的、符合预期的，避免数据冲突，保证程序的正确性和稳定性。

Java中线程安全队列的常用类型及原理

ConcurrentLinkedQueue

1. 原理
   • ConcurrentLinkedQueue是一个基于链接节点的无界线程安全队列，它采用了乐观锁的策略。在该队列中，节点之间通过next指针链接。由于队列是无界的，理论上可以不断添加元素而不会受到容量限制。
   • 它使用CAS（Compare - And - Swap）操作来实现线程安全。CAS是一种硬件级别的原子操作，它比较内存中的值与预期值，如果相等则将内存中的值替换为新值。在ConcurrentLinkedQueue中，offer和poll等操作利用CAS来修改队列的节点引用，避免了使用传统的锁机制，从而提高了并发性能。
2. 代码示例

import java.util.concurrent.ConcurrentLinkedQueue;

public class ConcurrentLinkedQueueExample {
    public static void main(String[] args) {
        ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>();

        Thread producer1 = new Thread(() -> {
            for (int i = 0; i < 5; i++) {
                queue.offer(i);
                System.out.println("Producer1 added: " + i);
            }
        });

        Thread consumer1 = new Thread(() -> {
            while (true) {
                Integer value = queue.poll();
                if (value != null) {
                    System.out.println("Consumer1 removed: " + value);
                } else {
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

        producer1.start();
        consumer1.start();

        try {
            producer1.join();
            consumer1.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在上述代码中，我们创建了一个ConcurrentLinkedQueue，并启动了一个生产者线程producer1和一个消费者线程consumer1。生产者线程不断向队列中添加元素，消费者线程不断从队列中取出元素。由于ConcurrentLinkedQueue是线程安全的，两个线程可以同时操作队列而不会出现数据不一致的问题。

ArrayBlockingQueue

1. 原理
   • ArrayBlockingQueue是一个有界的阻塞队列，它使用数组来存储元素。它内部使用一把锁（ReentrantLock）来保证线程安全，同时有两个条件变量notEmpty和notFull来实现阻塞和唤醒机制。
   • 当队列已满时，调用put方法添加元素的线程会被阻塞，直到队列有空间可用；当队列已空时，调用take方法获取元素的线程会被阻塞，直到队列中有元素可用。这种阻塞机制确保了在多线程环境下，对队列的操作是有序且安全的。
2. 代码示例

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class ArrayBlockingQueueExample {
    public static void main(String[] args) {
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(3);

        Thread producer1 = new Thread(() -> {
            for (int i = 0; i < 5; i++) {
                try {
                    queue.put(i);
                    System.out.println("Producer1 added: " + i);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        Thread consumer1 = new Thread(() -> {
            while (true) {
                try {
                    Integer value = queue.take();
                    System.out.println("Consumer1 removed: " + value);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        producer1.start();
        consumer1.start();

        try {
            producer1.join();
            consumer1.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在这个示例中，我们创建了一个容量为3的ArrayBlockingQueue。生产者线程producer1尝试向队列中添加5个元素，当队列满时，put方法会阻塞线程，直到队列有空间。消费者线程consumer1不断从队列中取出元素，当队列为空时，take方法会阻塞线程，直到队列中有元素。

LinkedBlockingQueue

1. 原理
   • LinkedBlockingQueue是一个基于链表的有界或无界阻塞队列。它内部使用两把锁（takeLock和putLock）来分别控制取元素和放元素的操作，这样可以提高并发性能，因为读操作和写操作可以并行进行，只要它们不竞争同一个锁。
   • 同样，它也使用了条件变量notEmpty和notFull来实现阻塞和唤醒机制。当队列满时，put操作会阻塞；当队列空时，take操作会阻塞。如果构造时不指定容量，它是一个无界队列，理论上可以无限添加元素。
2. 代码示例

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class LinkedBlockingQueueExample {
    public static void main(String[] args) {
        BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(3);

        Thread producer1 = new Thread(() -> {
            for (int i = 0; i < 5; i++) {
                try {
                    queue.put(i);
                    System.out.println("Producer1 added: " + i);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        Thread consumer1 = new Thread(() -> {
            while (true) {
                try {
                    Integer value = queue.take();
                    System.out.println("Consumer1 removed: " + value);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        producer1.start();
        consumer1.start();

        try {
            producer1.join();
            consumer1.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

此代码创建了一个容量为3的LinkedBlockingQueue。生产者线程向队列中添加元素，消费者线程从队列中取出元素。由于LinkedBlockingQueue的线程安全机制，多线程操作队列时不会出现数据错误。

PriorityBlockingQueue

1. 原理
   • PriorityBlockingQueue是一个无界的阻塞队列，它按照元素的自然顺序或者自定义的比较器顺序对元素进行排序。它内部使用堆数据结构来维护元素的顺序，保证每次取出的元素都是队列中优先级最高的（根据排序规则）。
   • 它使用一把锁（ReentrantLock）来保证线程安全，在添加和移除元素时，会对堆进行相应的调整以维护元素的顺序。当队列中没有元素时，take方法会阻塞等待新元素的到来。
2. 代码示例

import java.util.concurrent.PriorityBlockingQueue;

class PriorityElement implements Comparable<PriorityElement> {
    private int value;
    private int priority;

    public PriorityElement(int value, int priority) {
        this.value = value;
        this.priority = priority;
    }

    @Override
    public int compareTo(PriorityElement other) {
        return this.priority - other.priority;
    }

    @Override
    public String toString() {
        return "Value: " + value + ", Priority: " + priority;
    }
}

public class PriorityBlockingQueueExample {
    public static void main(String[] args) {
        PriorityBlockingQueue<PriorityElement> queue = new PriorityBlockingQueue<>();

        Thread producer1 = new Thread(() -> {
            queue.add(new PriorityElement(10, 2));
            queue.add(new PriorityElement(20, 1));
            queue.add(new PriorityElement(30, 3));
            System.out.println("Producer1 added elements.");
        });

        Thread consumer1 = new Thread(() -> {
            while (true) {
                try {
                    PriorityElement element = queue.take();
                    System.out.println("Consumer1 removed: " + element);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        producer1.start();
        consumer1.start();

        try {
            producer1.join();
            consumer1.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在这个示例中，我们定义了一个PriorityElement类，它实现了Comparable接口，根据优先级进行排序。生产者线程向PriorityBlockingQueue中添加元素，消费者线程按照优先级顺序从队列中取出元素。

自定义线程安全队列

基于锁的实现

1. 思路
   • 我们可以通过使用ReentrantLock和Condition来实现一个简单的线程安全队列。ReentrantLock用于保证对队列操作的线程安全，Condition用于实现阻塞和唤醒机制，类似于ArrayBlockingQueue的原理。
2. 代码示例

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class CustomThreadSafeQueue<T> {
    private final List<T> queue;
    private final int capacity;
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();

    public CustomThreadSafeQueue(int capacity) {
        this.queue = new LinkedList<>();
        this.capacity = capacity;
    }

    public void put(T item) throws InterruptedException {
        lock.lock();
        try {
            while (queue.size() == capacity) {
                notFull.await();
            }
            queue.add(item);
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public T take() throws InterruptedException {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                notEmpty.await();
            }
            T item = queue.remove(0);
            notFull.signal();
            return item;
        } finally {
            lock.unlock();
        }
    }
}

可以使用以下代码测试这个自定义队列：

public class CustomQueueTest {
    public static void main(String[] args) {
        CustomThreadSafeQueue<Integer> queue = new CustomThreadSafeQueue<>(3);

        Thread producer1 = new Thread(() -> {
            for (int i = 0; i < 5; i++) {
                try {
                    queue.put(i);
                    System.out.println("Producer1 added: " + i);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        Thread consumer1 = new Thread(() -> {
            while (true) {
                try {
                    Integer value = queue.take();
                    System.out.println("Consumer1 removed: " + value);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        producer1.start();
        consumer1.start();

        try {
            producer1.join();
            consumer1.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在上述代码中，CustomThreadSafeQueue通过ReentrantLock和Condition实现了一个有界的线程安全队列。put方法在队列满时阻塞，take方法在队列空时阻塞。

基于CAS的实现

1. 思路
   • 基于CAS的实现更加复杂，需要仔细处理节点的链接和更新操作。我们可以借鉴ConcurrentLinkedQueue的思想，使用AtomicReference来存储队列的头和尾节点，通过CAS操作来更新这些引用。
2. 代码示例

import java.util.concurrent.atomic.AtomicReference;

class Node<T> {
    T item;
    AtomicReference<Node<T>> next;

    Node(T item) {
        this.item = item;
        this.next = new AtomicReference<>(null);
    }
}

public class CustomConcurrentQueue<T> {
    private final AtomicReference<Node<T>> head = new AtomicReference<>(null);
    private final AtomicReference<Node<T>> tail = new AtomicReference<>(null);

    public void offer(T item) {
        Node<T> newNode = new Node<>(item);
        while (true) {
            Node<T> currentTail = tail.get();
            if (currentTail == null) {
                if (head.compareAndSet(null, newNode)) {
                    tail.compareAndSet(null, newNode);
                    return;
                }
            } else {
                Node<T> next = currentTail.next.get();
                if (next == null) {
                    if (currentTail.next.compareAndSet(null, newNode)) {
                        tail.compareAndSet(currentTail, newNode);
                        return;
                    }
                } else {
                    tail.compareAndSet(currentTail, next);
                }
            }
        }
    }

    public T poll() {
        while (true) {
            Node<T> currentHead = head.get();
            if (currentHead == null) {
                return null;
            }
            Node<T> next = currentHead.next.get();
            if (next == null) {
                if (head.compareAndSet(currentHead, null)) {
                    tail.compareAndSet(currentHead, null);
                    return currentHead.item;
                }
            } else {
                if (head.compareAndSet(currentHead, next)) {
                    return currentHead.item;
                }
            }
        }
    }
}

以下是测试代码：

public class CustomConcurrentQueueTest {
    public static void main(String[] args) {
        CustomConcurrentQueue<Integer> queue = new CustomConcurrentQueue<>();

        Thread producer1 = new Thread(() -> {
            for (int i = 0; i < 5; i++) {
                queue.offer(i);
                System.out.println("Producer1 added: " + i);
            }
        });

        Thread consumer1 = new Thread(() -> {
            while (true) {
                Integer value = queue.poll();
                if (value != null) {
                    System.out.println("Consumer1 removed: " + value);
                } else {
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

        producer1.start();
        consumer1.start();

        try {
            producer1.join();
            consumer1.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在这个基于CAS的自定义队列中，offer和poll方法通过AtomicReference和CAS操作实现了线程安全的入队和出队操作，避免了使用锁带来的性能开销。

选择合适的线程安全队列

1. 性能考虑
   • 如果追求高并发性能，且对队列容量没有严格限制，ConcurrentLinkedQueue是一个不错的选择，因为它基于乐观锁（CAS）机制，避免了锁竞争带来的性能损耗。
   • 对于有界队列，如果读操作和写操作的并发度较高，LinkedBlockingQueue由于使用两把锁分别控制读写，性能会优于ArrayBlockingQueue（使用一把锁）。但如果读写操作的竞争不激烈，ArrayBlockingQueue的简单锁机制可能也能满足需求且实现相对简单。
2. 功能需求
   • 如果需要按照元素的优先级取出元素，PriorityBlockingQueue是必须的选择。它能够根据元素的自然顺序或自定义比较器顺序进行排序，保证每次取出的是优先级最高的元素。
   • 如果需要在队列满或空时进行阻塞操作，ArrayBlockingQueue、LinkedBlockingQueue和PriorityBlockingQueue等阻塞队列都能满足需求，而ConcurrentLinkedQueue是非阻塞队列，不具备这种特性。
3. 内存考虑
   • ConcurrentLinkedQueue和PriorityBlockingQueue是无界队列，如果在内存有限的环境中使用，需要注意防止内存溢出。而ArrayBlockingQueue和LinkedBlockingQueue（有界构造时）可以通过设置容量来控制内存的使用。

总结不同实现的特点及适用场景

1. ConcurrentLinkedQueue
   • 特点：基于链表结构，无界队列，采用乐观锁（CAS）机制，非阻塞。
   • 适用场景：适用于高并发读写场景，对队列容量没有严格限制，且不希望因锁竞争导致性能下降的情况。例如，在分布式系统中作为消息传递队列，多个节点可以同时快速地向队列中添加和取出消息。
2. ArrayBlockingQueue
   • 特点：基于数组结构，有界队列，使用一把ReentrantLock保证线程安全，支持阻塞操作。
   • 适用场景：适用于需要严格控制队列容量，且读写操作竞争不特别激烈的场景。比如在一些资源有限的系统中，作为任务队列，限制任务的数量，以避免资源过度消耗。
3. LinkedBlockingQueue
   • 特点：基于链表结构，可设置为有界或无界队列，使用两把锁（takeLock和putLock）分别控制读写，支持阻塞操作。
   • 适用场景：适用于读写操作并发度较高的场景，特别是在有界队列的情况下，能够在保证线程安全的同时，提高并发性能。例如在多线程处理数据的管道中，作为数据传递的中间队列。
4. PriorityBlockingQueue
   • 特点：基于堆结构，无界队列，按照元素优先级排序，使用ReentrantLock保证线程安全，支持阻塞操作。
   • 适用场景：适用于需要按照元素优先级处理的场景，如在任务调度系统中，根据任务的优先级分配执行顺序。
5. 自定义队列
   • 基于锁的实现：实现相对简单，通过ReentrantLock和Condition实现线程安全和阻塞机制。适用于对性能要求不是极高，但需要定制化队列功能的场景。
   • 基于CAS的实现：实现复杂，但性能较高，避免了锁竞争。适用于对性能极度敏感，且对队列操作有深入理解和定制需求的场景。

在实际的Java多线程编程中，根据具体的业务需求和系统环境，选择合适的线程安全队列是保证程序性能和正确性的关键。通过深入理解各种队列的实现原理和特点，开发者能够更加高效地构建稳定、高性能的多线程应用程序。