使用 ConcurrentLinkedQueue 的 Java 线程问题

Question

我对以下代码片段有疑问。它旨在处理添加到事件队列 (ConcurrentLinkedQueue) 的事件（通过调用 processEvent 方法提供）。事件被添加到事件队列中并在 run 方法中定期进行处理。

几乎总是一切都很好。但有时在调用 processEvent 方法后，当事件添加到队列中时，运行部分无法看到有新事件。

知道哪里出了问题吗？除了使用字符串常量作为锁的明显错误之外？

import java.util.concurrent.ConcurrentLinkedQueue;

public class MyCommunicator implements Runnable {

private ConcurrentLinkedQueue<MyEvent> eventQueue = null;

private boolean stopped = false;

private String lock = "";
private Thread thread = null;

public MyCommunicator() {

    eventQueue = new ConcurrentLinkedQueue<MyEvent>();
}

public void start() {
    thread = new Thread(this, "MyCommunicatorThread");
    thread.start();
}

public void stop() {
    stopped = true;
    synchronized (lock) {
        lock.notifyAll();
    }
    eventQueue.clear();
}

public void run() {
    while (!stopped) {
        try {

            MyEvent event = null;
            while (!stopped && ((event = eventQueue.peek()) != null)) {
                sendEvent(event);
                eventQueue.poll();
            }

            if (!stopped) {
                synchronized (lock) {
                    lock.wait(10000L);
                }
            }
        }

        catch (Exception e) {

        }
    }
}

/**
 * START EVENT JOB - ADD A NEW EVENT TO BE PROCESSED
 */
public void processEvent(MyEvent event) {
    eventQueue.offer(event);
    synchronized (lock) {
        lock.notifyAll();
    }
}

/**
 * END EVENT JOB
 */
private void sendEvent(MyEvent event) {
    // do send event job
}

}

Answer 1

为什么要使用锁和通知？

使用 LinkedBlockingQueue 代替并保存自己所有的麻烦。

通过 poll() 超时将完成您想要做的一切。

---

编辑：关于当前代码；

您需要定义“无法看到新事件”。您的 run() 方法每 10 秒查看一次队列；如果队列中有东西，它会“看到它”并将其拉出。

• 如果您的意思是“它在收到通知时没有立即看到它，只有 10 秒后”，那么这很容易回答，因为您有一个竞争条件，很容易导致这种情况发生。当该线程完成检查/处理队列和获取锁之间时，可以将某些内容插入队列中。如果 wait() 没有超时，您将阻塞，直到插入下一个事件。如果在此期间调用 stop() 方法，您将丢失队列中的所有事件。使用 LinkedBlockingQueue 而不是所有不必要的锁定和通知可以解决此问题。这不是一个“简单”的解决方案，它是针对此用例和问题的正确解决方案。

• 如果情况并非如此，那么您根本没有向队列中插入任何内容，问题出在您未在此处发布的代码中。在不了解该代码的情况下猜测您正在尝试在 eventQueue.offer(event) 处插入 null MyEvent。因为你没有尝试/捕获 offer() 你不会知道它。忽略所有异常并且不检查返回值既不是一个好主意，也不是一个好做法。

• 第三种可能性是您在某处有一些其他代码锁定了同一个确切的内部字符串文字引用，这将导致该代码挂起。你提到了它，但我会在这里重申 - 这是一件非常糟糕的事情，特别是考虑到它是空字符串。 java.util.concurrent 包提供了真正的锁定 与 条件（如果您坚持在此处使用它们）。请注意，这仍然无法消除您有时会错过某个事件 10 秒的竞争条件，但它至少会更干净。为了消除竞争条件，您需要放弃并发队列而使用常规队列，并在访问它之前简单地获取锁（以及获取插入锁）。这将同步您的线程，因为插入器将被阻止插入，除非该线程正在等待锁定条件。在同一代码块中混合使用锁和无锁方法来进行线程同步通常会导致这些问题。

Answer 2

您遇到了所谓的丢失信号。您轮询队列，然后等待监视器（获取锁定）。生产者线程添加事件，然后调用notifyAll()（获取锁）。事件排队/轮询和条件等待/通知之间不存在happens-before关系。

因此，线程 A 可以在空时进行轮询，然后尝试获取锁，同时线程 B 添加一个元素并获取锁，通知所有等待线程然后释放锁。然后线程 A 获取锁并等待它，但信号已丢失。

由于您纯粹使用锁来发送信号，因此您可能会考虑另一种机制，例如可重复使用的锁存器，例如 Doug Lea 的新 jdk7 Phaser，或者直接使用BlockingQueue。

或者，我们有几个 ReusableLatch 例如 BooleanLatch 用于单个读取器线程或 PhasedLatch 用于多方支持。

Answer 3

乍一看没有什么特别的想法，但在您不知情的情况下，许多事情都可能出错，因为：

    catch (Exception e) {

    }

捕获任何 Exception 的处理程序（其中包括 RuntimeException 及其各种子类）然后忽略它通常是一个坏主意。如果这是为了捕获特定类型的异常（例如，可能由 lock.wait() 引发的 InterruptedException），那么您应该限制它到该异常类型。如果您出于某种原因捕获任何异常，那么您至少应该在异常发生时记录一些内容。

Answer 4

我在使用 ConcurrentLinkedQueue 时遇到的一个问题，我真的怀疑这是一个真正的错误，因为它并不是真正完全同步的证明。

我还没有完全测试过这一点，但我查看了代码，我很确定如果队列实际上是空的，.isEmpty() 不会同步。当一个线程调用 .isEmpty() 并返回 true 时，队列可能已经包含元素。