java - 多消费者无锁队列的实现

Question

最近我遇到了这个问题:有3个消费者线程，需要实现一个无锁队列(不能使用同步)，这样就不会消费线程被阻塞。假设队列已包含数据。

我想了一会儿，发现了原子操作，如果仔细使用的话会有所帮助。我的实现如下所示。由于数据已经在队列中，我还没有实现 enqueue 方法并在构造函数内填充数组。

public class SPMCQueue {

    private AtomicInteger index = new AtomicInteger(0);

    public int[] arr;

    public SPMCQueue(int size) {
        arr = IntStream.range(0, size).toArray();
    }

    public Integer dequeue() {
        Integer ret = null;
        int x = index.getAndIncrement();
        if (x < arr.length) {
            ret = arr[x];
            System.out.println(arr[x] + " by " + Thread.currentThread().getName());
        }
        else {
            throw new RuntimeException("Queue is empty");
        }
        return ret;
    }
}

class QueueTest {
    public static void main(String[] args) {

        SPMCQueueq = new SPMCQueue(40);

        Runnable t1 = () -> {
            try {
            while (true) {
                q.dequeue();
            }
            }catch(Exception e) {

            }
        };

        Runnable t2 = () -> { 
            try {
            while(true) { q.dequeue(); }
            }catch(Exception e) {

            }
        };

        Runnable r3 = () -> { 

            try {
                while(true) { q.dequeue(); }
            } catch (Exception e) {
                // TODO Auto-generated catch block
                //e.printStackTrace();
            }

        };

        Thread thread1 = new Thread(t1);
        Thread thread2 = new Thread(t2);
        Thread thread3 = new Thread(r3);

        thread1.start();
        thread2.start();
        thread3.start();

    }
}

我已经执行了上面的程序，结果显示所有 3 个消费者都在消耗数据，尽管是乱序的，并且某些线程比其他线程消耗更多的数据，但我没有看到任何数据出现多次在 o/p 中。

我有以下问题:

1. 上面的实现有什么问题吗？

2. 还有哪些其他方法可以实现无锁消费者队列？

Answer 1

我想同时给出我的回复和答案:https://stackoverflow.com/a/21101904/7340499因为这与您的问题类似。

那么，你的问题:

but I don't see any of the data appearing multiple times in the o/p.

这是预期的。因为 getAndIncrement() 是原子的，并且每当访问该函数时，您都会得到不同的 x 值，因此会得到不同的输出。 但是，由于在单个非原子出列操作中组合了“getAndIncrement()”和“System.out.print()”函数，您的输出有时可能会乱序，例如你在一个线程上得到 x=1，另一个线程中断，得到 x=2 并打印它，然后你的初始线程完成打印。我相信这也指出了您的实现问题，如(1)中所问。您的应用程序可以接受无序处理队列吗？

What are the other ways to implement lock-free consumer queue?

嗯，正如您所注意到的，原子操作是一种方法。但本质上，无论如何，它们非常像锁。它们只是在较小的规模上运作(尽管存在一些实现差异)。因此，很难将它们概念化为不同的方法，至少对我自己来说是这样。除此之外，我相信这里还有一些不错的资源:Lock Free Queue -- Single Producer, Multiple Consumers