乐观锁思想和AtomicInteger对队列push操作进行保护

Question

请问使用AtomicInteger提供的方法和乐观锁思想对队列的push操作做保护，这样写安全吗？

    /**
     * 循环队列，保留一个槽位用于判断是否满了，所以实际最大容量为 SIZE - 1
     */
    public static final Integer SIZE = 1001;

    private AtomicInteger head = new AtomicInteger(0);
    private AtomicInteger tail = new AtomicInteger(0);

    private Object[] objects;


    public QueueTest2() {
        objects = new Object[SIZE];
    }

    public boolean isEmpty() {
        return head.get() == tail.get();
    }

    public boolean isFull() {
        return head.get() + 1 % SIZE == tail.get();
    }


    /**
     * 直到成功push才会退出，否则会一直Push直到容量满了
     *
     * @param t
     */
    public void push(T t) {
        //非空并且该槽位没有被更新
        if (isFull()) {
            return;
        }
        int expectHead = head.get();
        int updateHead = (head.get() + SIZE + 1) % SIZE;
        if (head.compareAndSet(expectHead, updateHead)) {
            objects[updateHead] = t;
        } else {
            push(t);
        }
    }

    public T peek() {
        return (T) objects[tail.get()];
    }

    public T pop() {
        if (isEmpty()) {
            return null;
        }

        int expectTail = tail.get();
        int updateTail = (tail.get() + SIZE + 1) % SIZE;
        T t = null;
        if (tail.compareAndSet(expectTail, updateTail)) {
            t = (T) objects[updateTail];
            objects[updateTail] = null;
        }
        return t;
    }

    public static void main(String[] args) throws InterruptedException {

        Set<Integer> set = new HashSet<>();

        QueueTest2<Integer> test2 = new QueueTest2();
        ExecutorService executorService = Executors.newFixedThreadPool(6);
        for (int i = 0; i < 1000; i++) {
            final int j = i;
            executorService.execute(() -> {
                test2.push(j);
            });
        }

        executorService.awaitTermination(10, TimeUnit.SECONDS);
        for (int i = 0; i < 1000; i++) {
            Integer pop = test2.pop();
            System.out.println(pop);
            set.add(pop);
        }
        System.out.println(test2.head + " " + test2.tail);
        System.out.println("--------------------------------------------------");

    }

Answer 1

引自免费微信小程序：皆我百晓生

使用AtomicInteger和乐观锁思想对队列的push操作进行保护是一种有效的方式，可以提供线程安全性。但是上述的代码实现并没有完全符合乐观锁的思想，下面是对代码的详细分析和改进建议：

1. isFull方法计算错误：在isFull方法中，计算是否队列已满的表达式不正确，应该将加法操作的结果用括号括起来，即应该是(head.get() + 1) % SIZE == tail.get()。原来的表达式中，加法操作的优先级高于求模操作，导致计算结果错误。

2. 乐观锁思想的实现：在push和pop方法中，应该使用CAS操作来更新head和tail变量，这样可以实现乐观锁的思想。代码中使用的是AtomicInteger类，其中的compareAndSet方法可以用于原子地进行比较和设置操作。

3. 递归调用的问题：在push方法的else分支中，代码使用递归调用来实现重试操作，但这样做可能会导致栈溢出。推荐的做法是使用循环来重试，直到成功为止。

下面是根据上述改进建议进行的代码修改：

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class QueueTest2<T> {
    private static final Integer SIZE = 1001;
    private AtomicInteger head = new AtomicInteger(0);
    private AtomicInteger tail = new AtomicInteger(0);
    private Object[] objects;

    public QueueTest2() {
        objects = new Object[SIZE];
    }

    public boolean isEmpty() {
        return head.get() == tail.get();
    }

    public boolean isFull() {
        return (head.get() + 1) % SIZE == tail.get();
    }

    public void push(T t) {
        while (true) {
            if (isFull()) {
                return;
            }
            int expectHead = head.get();
            int updateHead = (head.get() + SIZE + 1) % SIZE;
            if (head.compareAndSet(expectHead, updateHead)) {
                objects[updateHead] = t;
                return;
            }
        }
    }

    public T peek() {
        return (T) objects[tail.get()];
    }

    public T pop() {
        while (true) {
            if (isEmpty()) {
                return null;
            }
            int expectTail = tail.get();
            int updateTail = (tail.get() + SIZE + 1) % SIZE;
            if (tail.compareAndSet(expectTail, updateTail)) {
                T t = (T) objects[updateTail];
                objects[updateTail] = null;
                return t;
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Set<Integer> set = new HashSet<>();

        QueueTest2<Integer> test2 = new QueueTest2<>();
        ExecutorService executorService = Executors.newFixedThreadPool(6);
        for (int i = 0; i < 1000; i++) {
            final int j = i;
            executorService.execute(() -> {
                test2.push(j);
            });
        }

        executorService.awaitTermination(10, TimeUnit.SECONDS);
        for (int i = 0; i < 1000; i++) {
            Integer pop = test2.pop();
            System.out.println(pop);
            set.add(pop);
        }
        System.out.println(test2.head + " " + test2.tail);
        System.out.println("--------------------------------------------------");
    }
}

希望上述解决方案对您有所帮助！如果还有任何问题，请随时提问。