使用 ConcurrentLinkedQueue<T> 对 Collection 进行序列化访问答案

【问题标题】：Serialized access to Collection with the use of ConcurrentLinkedQueue<T>使用 ConcurrentLinkedQueue<T> 对 Collection 进行序列化访问
【发布时间】：2015-04-12 22:06:02
【问题描述】：

我有以下关于 Java 7 ConcurrentLinkedQueue 的问题。让我们假设我有以下课程：

public class Blah {
    private ConcurrentLinkedQueue<String> queue;

    public Blah() {
        queue = new ConcurrentLinkedQueue<String>();
    }

    public void produce(String action, String task) throws InterruptedException {
        synchronized(queue) {
            while(queue.size() >= 8) 
                queue.wait();
            queue.add(action + "#" + task);
            queue.notifyAll();
        }
    }

    public void consume() throws InterruptedException {
        synchronized(queue) {
            while(queue.size() <= 0)
                queue.wait();
            String element = queue.poll();
            StringTokenizer strTok = new StringTokenizer(element, "#");
            String action = strTok.nextToken();
            String task = strTok.nextToken();
            /**
             * Operate on request
             */
            queue.notifyAll();
        }
    }
}

生产和消费函数将由并发线程调用，以便为列表/从列表中生成/删除线程。我实现了前面的函数consume() 和produce()，以便序列化队列中元素的添加/删除。以上是必需的，还是 ConcurrentLinkedQueue 负责？我问是因为我不想降低代码的性能。

谢谢你，尼克

【问题讨论】：

你到底为什么要同步？？
@Boris 因为我希望关键部分内的所有操作都以原子方式执行。
这是一个ConcurrentLinkedQueue。如果使用正确，它可以保证原子性和可见性。您将非阻塞 Queue 强制转换为 BlockingQueue。不要。

标签： java multithreading java.util.concurrent java-threads

【解决方案1】：

TL;DR：您正在使用专门设计为非阻塞的 Queue 作为 BlockingQueue。

你的代码可以改写为：

public class Blah {
    private BlockingQueue<String> queue;

    public Blah() {
        queue = new LinkedBlockingQueue<>(8);
    }

    public void produce(String action, String task) throws InterruptedException {
        while (true) {
            queue.put(action + "#" + task);
        }
    }

    public void consume() throws InterruptedException {
        while (true) {
            final String[] data = queue.take().split("#");
            final String action = data[0];
            final String task = data[1];
        }
    }
}

BlockingQueue 以8 元素为界。如果队列已满，put 将阻塞。如果队列为空，take 将阻塞。

不需要同步。

另外，StringTokenizer 已被弃用。我建议你使用class 类似：

public class Data {
    private final String action;
    private final String task;

    public Data(final String action, final String task) {
        this.action = action;
        this.task = task;
    }

    public String getAction() {
        return action;
    }

    public String getTask() {
        return task;
    }
}

交换数据。没有理由创建和解析Strings。

【讨论】：

【解决方案2】：

您确实降低了代码的性能，因为您使用的是同步机制中“最慢”的内置同步。

您有一个完美的BlockingQueue 用例。它为您提供put 和take 操作，这些操作会阻塞直到空间/元素可用。

ConcurrentLinkedQueue 只为您提供线程安全而不是同步。这意味着您可以在多线程应用程序中安全地从队列中添加/删除元素，但它没有为您提供等待空间/元素的机制，因此您正确地使用了 wait()、notify() 方法（尽管您可以在任何其他共享对象上也同步，它不必是那个队列）。在 Java 中使用 Lock 而不是 synchronized 会快得多。

【讨论】：

我想要一个同步块的主要原因是因为我希望代码的按请求操作部分以原子方式执行。
我不知道你的请求部分代码是什么意思，我没有看到这样的东西。 BlockingQueue add() 正是实现了这部分代码： while(queue.size() >= 8) queue.wait(); queue.add(action + "#" + task); queue.notifyAll();但以更有效的方式

【解决方案3】：

假设 produce 和 consume 将被独立调用，您可能会以无限循环结束，因为两者都试图锁定队列对象，因此一旦 consume 找不到任何元素，它将永远不会释放锁，防止产品添加元素。同步逻辑大概应该在调用consume和produce的方法中实现。

【讨论】：

他们在 wait() 操作中释放锁。