消費者や労働者のパターン

Question

私は10人の労働者と、以下の二つの方法の一覧のコードを持っている：

public void demoDeques() { 
     int maxSizeOfJobDeque = 3; 
     Producer producer = new ProducerImpl(maxSizeOfJobDeque); 

     Logger.debug("WorkFlowEngineImpl : " + 
       "Creating Workers and adding them to allocator"); 
     List<Worker> workerList = buildWorkerList(producer); 
     Logger.debug("WorkFlowEngineImpl : " + 
       "Assigning some jobs to the workers. " + 
        "The workers have not been started yet"); 

     for (int i=1; i<4; i++) { 
      producer.assign(new JobImpl("job " + i, i)); 
      try { 
       Thread.sleep(4000); 
      } catch(InterruptedException e) { 
       e.printStackTrace(); 
      } 
     } 

     Logger.debug("WorkFlowEngineImpl : " + "Starting the workers"); 
     startWorkersAndWait5Seconds(workerList); 
     Logger.debug("WorkFlowEngineImpl : " + 
       "Assigning some more jobs to the " + 
        "started workers"); 

     for (int i=4; i<7; i++) { 
      producer.assign(new JobImpl("Job " + i, i)); 
      try { 
       Thread.sleep(4000); 
      } catch(InterruptedException e) { 
       e.printStackTrace(); 
      } 
     } 

     Logger.debug("WorkFlowEngineImpl : " + "Assigning More Jobs"); 
     for (int i=7; i<11; i++) { 
      producer.assign(new JobImpl("job" + i, i)); 
      try { 
       Thread.sleep(4000); 
      } catch(InterruptedException e) { 
       e.printStackTrace(); 
      } 
     } 
    } 

プロデューサー：

public synchronized void assign(Job job) { 
    Set<Worker> workerSet = jobMap.keySet(); 
    LinkedBlockingDeque<Job> jobQueue; 
    StringBuffer sb; 

    for (Worker worker : workerSet) { 
     jobQueue = jobMap.get(worker); 

     sb = new StringBuffer(); 
     sb.append("Assigning job "); 
     sb.append(job.getJobNumber()); 
     sb.append(" to "); 
     sb.append(worker); 
     sb.append("'s jobs Deque"); 

     Logger.debug("Producer : " + sb.toString()); 

     if (! jobQueue.offerFirst(job)) { 
      jobQueue.pollLast(); 
      jobQueue.offerFirst(job); 
     } 
    } 
} 

私は私が持っているように、2つの方法を変更しようとしていますアロケータは100のジョブのリストを有し、100人に達するまで時間がたつにつれて10人の作業者のそれぞれに最大3つのジョブが与えられるように割り当てる。すなわち、 6になるので、10人の作業者に到達すると、3人の作業を割り当てた作業者1に戻って、100番目の作業に達するまで停止し、すべてのジョブが割り当てられていることを警告します。助けを借りて助けてください。

Answer 1

私はあなたの仕事の割り当て方法には反対しています。私はあなたのコードを掛けていると思います。

一般に、プロデューサは作業を作成してキューに割り当てる必要があります。雇用を減らすことを心配するのは、生産者の責任ではありません。作業員が利用可能になると、キューからジョブが引き出されます。彼らが1つの仕事か3つを引くかどうかは、それが引っ張ってくると単にワーカースレッド内の論理の問題です。

おそらく、キューのシングルトンクラスを作成したいと思うでしょう。受信ジョブと要求ジョブを処理する必要があります。

このようにアプローチすると、生産者から作業者を切り離すことができます。ワークロード/ボトルネックの指示に従って、プロデューサまたはワーカーを追加できます。また、労働者がプロデューサーに作業を進める準備ができていることを知らせる、何らかの信号メカニズムを生成しないようにします。最後に、これにより、あなたはdemoDequeue内の所定の位置にあるマジックナンバーを使って、ジョブが1つしか利用できないか、ジョブの番号が順不同になってしまうようなさまざまな競合状態を心配する必要がなくなります。

HTH申し訳ありませんが、直接的な回答ではありませんが、問題を解決するためのより良い方法であなたを得る必要があります。

Answer 2

あなたの質問には特定の実行順序が記載されていますが、あなたのコメントはこれが真の要件ではないことを示しているようです。独自のスレッドプールを実装する代わりに、Java 1.5で提供されるExecutorsを使用することをお勧めします。たとえば：

ExecutorService threadPool = 
    Executors.newFixedThreadPool(NUMBER_THREADS_TO_RUN_JOBS); 
// your producers then just have to submit jobs to the pool 
for (int i=1; i<4; i++) { 
    threadPool.submit(new JobImpl("Job " + i, i)); 
} 

私は処理がそれぞれの「仕事」ではなくJobImplが動作するように上記のコードのためRunnableを実装する必要がある場合に発生する必要がまさにあなたの例から見ることができません。仕事の価値を返す必要がある場合は、JobImplをCallableに変更し、call()によって返される結果を得るためにメソッドによって返されたFutureを使用して返すことができます。