如何实现一个可以被多个线程处理的队列？

Question

我认为我做错了。我正在创建线程，以便从共享队列中剔除一些数据。我的问题是程序速度慢，内存不足，我怀疑队列可能不像我希望的那样共享。我怀疑这是因为在我的代码中，我添加了一行显示队列的大小，如果我启动2个线程，然后我得到两个完全不同数字的输出，似乎自己增加（我认为它可能是相同的数字，但也许它从100跳到2，等等，但看完它后显示105和5，并以不同的速率运行，如果我有4个线程，那么我看到4个不同的数字）。

这是相关部分的片段。我在节目

static class queue_class { 
     int number; 
     int[] data; 
     Context(int number, int[] data) { 
      this.number = number; 
      this.data = data; 
     } 
    } 

然后我将一些职位的调用后创建队列的顶部创建队列我想要的数据静态类..

static class process_threaded implements Callable<Void> { 
    // queue with contexts to process 
    private Queue<queue_class> queue; 

    process_threaded(queue_class request) { 
     queue = new ArrayDeque<queue_class>(); 
     queue.add(request); 
    } 

    public Void call() { 
     while(!queue.isEmpty()) { 
      System.out.println("in contexts queue with a size of " + queue.size()); 
      Context current = contexts.poll(); 
      //get work and process it, if it work great then the solution goes elsewhere 
      //otherwise, depending on the data, its either discarded or parts of it is added back to queue 
      queue.add(new queue_class(k, data_list)); 

，你可以看到，数据有3个选项，如果数据是好的，就会被发送出去，如果数据完全可怕或丢回队列就丢弃。我认为队列正在发送，但我怀疑是因为每个线程都在自己的队列中工作，而不是共享队列。

这是猜测正确的，我在做这错了吗？

Answer 1

你在你的评估，即每个线程（可能）拥有自己的队列工作，因为你正在创建你的Callable的构造队列正确。 （它实际上很奇怪有一个Callable<Void> - 是不是只是一个Runnable？）

还有其他问题，例如，您正在处理一个非线程安全的队列，或者你的代码在编写时不会编译的事实。

但是，重要的问题是你真的需要显式创建一个队列吗？为什么没有ExecutorService你提交Callable（或Runnables，如果你决定进行切换）：将执行者的引用传递到你的Callable，他们可以将新的Callable添加到执行者的任务队列中以运行。没有必要重新发明轮子。

例如：

static class process_threaded implements Runnable { 
    // Reference to an executor 
    private final ExecutorService exec; 
    // Reference to the job counter 
    private final AtomicInteger jobCounter; 
    // Request to process 
    private queue_class request; 

    process_threaded(ExecutorService exec, AtomicInteger counter, queue_class request) { 
     this.exec = exec; 
     this.jobCounter = counter; 
     this.jobCounter.incrementAndGet(); // Assuming that you will always 
              // submit the process_threaded to 
              // the executor if you create it. 
     this.request = request; 
    } 

    public run() { 
     //get work and process **request**, if it work great then the solution goes elsewhere 
     //otherwise, depending on the data, its either discarded or parts of are added back to the executor 
     exec.submit(new process_threaded(exec, new queue_class(k, data_list))); 

     // Can do some more work 

     // Always run before returning: counter update and notify the launcher 
     synchronized(jobCounter){ 
      jobCounter.decrementAndGet(); 
      jobCounter.notifyAll(); 
     } 
    } 
} 

编辑：

为了解决您的时候关闭执行的问题，我觉得最简单的办法是有一个工作计数器，关机时达到0.对于螺纹安全，AtomicInteger可能是最佳选择。我在上面添加了一些代码来包含更改。然后你的启动代码看起来像这样：

void theLauncher() { 

    AtomicInteger jobCounter = new AtomicInteger(0); 

    ExecutorService exec = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcesses()); 

    exec.submit(new process_threaded(exec, jobCounter, someProcessRequest)); 
    // Can submit some other things here of course... 

    // Wait for jobs to complete: 
    for(;;jobCounter.get() > 0){ 
     synchronized(jobCounter){ // (I'm not sure if you have to have the synchronized block, but I think this is safer. 
      if(jobCounter.get() > 0) 
       jobCounter.wait(); 
     } 
    } 

    // Now you can shutdown: 
    exec.shutdown(); 
} 

Answer 2

不要重新发明轮子！如何使用ConcurrentLinkedQueue？从javadocs：

基于链接节点的无界线程安全队列。该队列命令元素FIFO（先进先出）。队列头是最长时间在队列中的元素。队列的尾部是已经在队列上的最短时间的元素。新元素插入到队列尾部，队列检索操作获取队列头部的元素。当许多线程将共享对共同集合的访问时，ConcurrentLinkedQueue是一个合适的选择。