使用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) 中插入一个空的 MyEvent。因为你不是 try/catch'ing offer() 你不会知道它。忽略所有异常并且不检查返回值既不是一个好主意也不是一种做法。

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