队列工作线程停止工作，线程安全问题？

Question

我想先介绍一下我的问题。

我有几个WorkingThreads正在接收一个字符串，处理该字符串，然后将处理后的字符串发送到一个全局队列，如下所示：

class Main {
  public static Queue<String> Q;
  public static void main(String[] args) {
    //start working threads
  }
}

WorkingThread.java：

class WorkingThread extends Thread {
  public void run() {
    String input;
    //do something with input
    Main.q.append(processedString);
}

因此，现在每隔800 to就有另一个名为Inserter的线程排空所有条目，以形成一些sql，但这并不重要。

class Inserter extends Thread {
  public void run() {
    while(!Main.Q.isEmpty()) {
      System.out.print(".");
      // dequeue and formulate some SQL
    }
  }
}

每样东西都工作了5到10分钟，但突然之间，我看不到任何点打印出来(基本上是插入器的心跳)。队列不是空的，我可以保证，但是插入器不能工作，即使它是正常启动的。

我有一个悬念，当一个工人想要插入什么而插入者去排队的时候有问题，这可能是某种“死锁”吗？

我真希望有人能对这种行为作出解释。我期待着学习;)

编辑:我正在使用

Queue<String> Q = new LinkedList<String>();

Answer 1

您没有使用synchronized或线程安全的Queue，因此存在竞争风险。LinkedList的使用显示出对这一事实的了解(有点吓人)。在尝试处理更多线程代码之前，您可能需要阅读更多关于线程和线程安全性的信息。

必须手动同步或使用JDK提供的现有实现之一。生产者/消费者模式通常使用BlockingQueue实现之一实现。

如果队列已满，有界大小的BlockingQueue将阻止生产者尝试put。如果队列为空，BlockingQueue将始终阻塞使用者。

这允许您删除队列上旋转并等待项的所有自定义逻辑。

使用Java 8 lambda的简单示例如下所示：

public static void main(String[] args) throws Exception {
    final BlockingQueue<String> q = new LinkedBlockingQueue<>();
    final ExecutorService executorService = Executors.newFixedThreadPool(4);
    final Runnable consumer = () -> {
        while (true) {
            try {
                System.out.println(q.take());
            } catch (InterruptedException e) {
                return;
            }
        }
    };
    executorService.submit(consumer);
    final Stream<Runnable> producers = IntStream.range(0, 5).mapToObj(i -> () -> {
        final Random random = ThreadLocalRandom.current();
        while (true) {
            q.add("Consumer " + i + " putting " + random.nextDouble());
            try {
                TimeUnit.MILLISECONDS.sleep(random.nextInt(2000));
            } catch (InterruptedException e) {
                //ignore
            }
        }
    });
    producers.forEach(executorService::submit);
}

consumer在BlockingQueue.take方法上阻塞，并且立即有一个可用项，它将被唤醒并打印该项。如果没有项，线程将被挂起--允许物理CPU执行其他操作。

producers分别使用add将一个String推送到队列中。由于队列是无界的，add将始终返回true。在可能存在工作积压的情况下，for consumer可以绑定队列并使用put方法(该方法抛出InterruptedException，因此需要try..catch，这就是为什么使用add更容易)--这将自动创建流控制。

Answer 2

您需要同步访问队列或使用ConcurrentLinkedQueue (请参阅http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ConcurrentLinkedQueue.html )。

或者还建议使用BlockingQueue (取决于您的需求) http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/BlockingQueue.html

有关BlockingQueue的更详细说明，请参见

http://tutorials.jenkov.com/java-util-concurrent/blockingqueue.html

Answer 3

似乎更像是同步问题。您正在尝试做一个- Producer - Consumer problem的模拟.您需要同步队列或使用BlockingQueue。你可能有种族问题。