隊列中的多線程作業 - 作業太多？

Question

假設我有一個很大的隊列，就像10,000個對象一樣。我想用5個工作線程創建一個線程池，每個工作線程從隊列中移除一個項並對其進行處理，直到隊列爲空。

我擔心的是，通過使用我在不同地方看到的設置，我最終創建了10,000個工作，但是通過5名工作人員執行。我覺得這不是真正可擴展的 - 隊列已經有10,000個項目，現在我在堆棧上還有10,000個作業（即使它們沒有被正在執行，這看起來像是一個內存問題）。

這似乎是這個答案的建議：https://stackoverflow.com/a/9916299/774359 - 這是「// now submit our jobs」部分，讓我很擔心。是否有效地將隊列倒入工作中是一個問題？

這裏是什麼，我至今一個簡單的例子：）

在main（：

ExecutorService executor = Executors.newFixedThreadPool(5); 
while(!hugeQueue.isEmpty()) { 
    String work = hugeQueue.remove(); 
    System.out.println("Creating job for " + work); 
    Runnable worker = new Worker(work); 
    executor.execute(worker); 
} 

在Worker類：​​

public Worker(String itemFromQueue) { this.job = itemFromQueue; } 

@Override 
public void run() { 
    System.out.println("Working on " + this.itemFromQueue); 
    //Do actual work 
} 

當hugeQueue包含萬個的數字，我查看所有10,000個「創建作業」消息，然後查看所有10,000個「正在處理」消息。我認爲如果只有5個職位同時創建，然後他們就可以開展工作 - 當一個線程打開時，它會創建另一個工作，然後工作。這樣，堆棧中就不會有10,000個作業。我將如何實現這一目標？我是否正確地思考這個架構？

---

編輯，包括基於一個答案更新的信息：

@ seneque的代碼沒有編譯通俗易懂，所以我做了一些細微的變化 - 不幸的是，這個輸出是工人剛剛創建，而沒有實際的工作。

在main（）：

int numOfThreads = 5; 
BlockingQueue<Integer> hugeQueue = new LinkedBlockingQueue<>(); 
for(int x = 0; x < 1000; x++) { hugeQueue.add(x); } 

ExecutorService executor = Executors.newFixedThreadPool(numOfThreads); 
LongRunningWorker longRunningWorker = new LongRunningWorker(); 

for(int i = 0; i < numOfThreads ; i++) { 
    System.out.println("Created worker #" + i); 
    executor.submit(longRunningWorker); 
} 
System.out.println("Done"); 

在LongRunningWorker：

public class LongRunningWorker implements Runnable { 
    BlockingQueue<Integer> workQueue; 
    void spiderExmaple(BlockingQueue<Integer> workQueue) { 
     this.workQueue = workQueue; 
    } 

    @Override 
    public void run() { 
     try { 
      while(workQueue.poll(3, TimeUnit.SECONDS) != null) { 
       Integer work = workQueue.remove(); 
       System.out.println("Working on " + work); 
       new Worker(work).run(); 
      } 
     } catch (InterruptedException e) { e.printStackTrace(); } 
    } 
} 

在工人：

public class Worker implements Runnable{ 
    Integer work; 
    Worker(Integer x) { this.work = x; } 

    @Override 
    public void run() { 
     System.out.println("Finished work on " + this.work); 

    } 
} 

Answer 1

一種解決方案將是有你五個直接輪詢隊列。

BlockingQueue<String> hugeQueue = ... 
ExecutorService executor = Executors.newFixedThreadPool(5); 
LongRunningWorker longRunningWorker = new LongRunningWorker(hugeQueue); 
for(int i = 0; i < 5 ; i++) { 
    executor.submit(longRunningWorker) 
} 

然後LongRunningWorker的定義如下：

class LongRunningWorker(BlockingQueue<String> workQueue) extends Runnable { 
    final BlockingQueue<String> workQueue; 
    LongRunningWorker(BlockingQueue<String> workQueue) { 
     this.workQueue = workQueue; 
    } 

    public void run() { 
     while((String work = workQueue.poll(3, TimeUnit.Second) != null) { 
      try { 
       new Worker(work).run(); 
      } catch (Exception e) { 
       // 
      } 
     } 
    } 
}