java - 线程化的BlockingQueue，需要澄清

Question

在使用BlockingQueue时，我实现了以下逻辑来从中读取数据，直到另有通知。不幸的是，以下情况间歇性地发生:

问题:

• 即使将 shouldContinueReading 设置为 false，循环也不会持续中断
• 问题是间歇性的，有时一切正常

作为 QThread 类的一部分，我声明:

public static volatile boolean          shouldContinueReading   = true;

Run(确认正在执行)方法包含:

    while (shouldContinueReading) {
        try {

            String retrieved = qIn.poll(2, TimeUnit.MILLISECONDS);
            if (retrieved != null)
                consume(retrieved);

        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    System.out.println("I am out");  // <-- not always seen
    if (qIn.remainingCapacity() > 0) {
        try {
            consume(qIn.take());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

当这种情况发生时，在另一个线程中，当某些事情发生时，shouldContinueReading 会更改状态

    while (stillReading) {
        // do nothing
    }

    QThread.shouldContinueReading = false;

更新:问题已解决

事实证明问题出在更远的地方:

private void consume(String take) {
            // some processing

    produce(newData.toString());
}

private void produce(String newData) {
    System.out.println(newData);
    try {
        qOut.put(newData);          // <-- Problem is here. Should use offer instead of put
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

qIn(队列输入)和 qOut(队列输出)均声明为:

private volatile BlockingQueue<String>  qIn;
private volatile BlockingQueue<String>  qOut;

对象本身在其他地方创建，如下所示并传递给构造函数:

    BlockingQueue<String> q1 = new SynchronousQueue<String>();
    BlockingQueue<String> q2 = new SynchronousQueue<String>();

    QThread qThread = new QThread(q1, q2);

有什么建议吗？我应该用 qOut 做什么？我声明得不正确吗？

Answer 1

我敢打赌QThread.shouldContinueReading = false;不会总是被执行，或者读取线程一开始就没有执行。 IE。您所看到的问题可能出在上游的某个地方——而不是这里。我要做的第一件事就是以 100% 的信心确定问题到底出在哪里(多放一些打印语句)。

除了这个问题之外，我建议使用线程中断机制，而不是滚动你自己的标志(这反过来只是一个美化的标志，但这样你就可以影响第三方代码，例如 BlockedQueue 并使实现更简单、更高效)特别是如果这是生产代码。