ConcurrentMapにスレッドを格納するのは安全ですか？

Question

私は自分のサービスへのREST呼び出しによって新しいスレッドを作成するバックエンドサービスを構築しています。スレッドが5分かかると何も受け取らない場合、スレッドは別のREST呼び出しを待機します。 すべてのスレッドを追跡するために、現在実行中のすべてのスレッドを追跡するコレクションがあるため、ユーザーがアクションを受け入れたり却下したりするなど、REST呼び出しが最後に来ると、userIDを使用してそのスレッドを識別できます。そのスレッドが受け入れられた場合、そのスレッドが次のアクションを実行することができる場合は、そのスレッドをコレクションから削除します。私は同時実行性の問題を避けるためにこれをConcurrentMapを使って実装しました。

これは私がスレッドで作業して初めてのので、発生する可能性のある問題を見落とさないようにしたいと考えています。私のコードを見て、私がそれをより良くすることができるか、または欠陥があるかどうかを教えてください。

public class UserAction extends Thread { 

    int  userID; 
    boolean isAccepted = false; 
    boolean isDeclined = false; 
    long timeNow  = System.currentTimeMillis(); 
    long timeElapsed = timeNow + 50000; 

    public UserAction(int userID) { 
     this.userID = userID; 
    } 

    public void declineJob() { 
     this.isDeclined = true; 
    } 

    public void acceptJob() { 
     this.isAccepted = true; 
    } 

    public boolean waitForApproval(){ 
     while (System.currentTimeMillis() < timeElapsed){ 
     System.out.println("waiting for approval"); 
     if (isAccepted) { 
      return true; 
     } else if (declined) { 
      return false; 
     } 

    } 
     return isAccepted; 
    } 

    @Override 
    public void run() { 
     if (!waitForApproval) { 
      // mustve timed out or user declined so remove from list and return thread immediately 
      tCollection.remove(userID); 
      // end the thread here 
      return; 
     } 
     // mustve been accepted so continue working 
    } 
} 

public class Controller { 

    public static ConcurrentHashMap<Integer, Thread> tCollection = new ConcurrentHashMap<>(); 

    public static void main(String[] args) { 
     int barberID1 = 1; 
     int barberID2 = 2; 

     tCollection.put(barberID1, new UserAction(barberID1)); 
     tCollection.put(barberID2, new UserAction(barberID2)); 

     tCollection.get(barberID1).start(); 
     tCollection.get(barberID2).start(); 

     Thread.sleep(1000); 
     // simulate REST call accepting/declining job after 1 second. Usually this would be in a spring mvc RESTcontroller in a different class. 
     tCollection.get(barberID1).acceptJob(); 
     tCollection.get(barberID2).declineJob(); 

    } 
} 

Answer 1

スレッドには（明示的な）スレッドは必要ありません。最初の残りの呼び出しで作成されたタスクオブジェクトの共有プール。

2回目の残りの呼び出しが来ると、既に使用するスレッド（残りの呼び出しを処理しているスレッド）があります。ユーザーIDに基づいてタスクオブジェクトを取得するだけで済みます。例えば、DelayQueueで実行できる期限切れのタスクを取り除く必要もあります。

擬似コード：

public void rest1(User u) { 
    UserTask ut = new UserTask(u); 
    pool.put(u.getId(), ut); 
    delayPool.put(ut); // Assuming UserTask implements Delayed with a 5 minute delay 
} 

public void rest2(User u, Action a) { 
    UserTask ut = pool.get(u.getId()); 
    if(!a.isAccepted() || ut == null) 
     pool.remove(u.getId()); 
    else 
     process(ut); 

    // Clean up the pool from any expired tasks, can also be done in the beginning 
    // of the method, if you want to make sure that expired actions aren't performed 
    while((UserTask u = delayPool.poll()) != null) 
     pool.remove(u.getId()); 
} 

Answer 2

あなたの旗isAcceptedとクラスAtomicBooleanのisDeclinedを行う必要があり、同期の問題があります。 重要な概念は、あるスレッドのメモリへの変更が、そのデータを必要とする他のスレッドに伝達されるようにする必要があることです。これらはメモリフェンスと呼ばれ、同期呼び出しの間に暗黙的に発生することがよくあります。

「中央メモリ」を備えた（単純な）フォンノイマンアーキテクチャの考え方は、最新のマシンでは偽であり、データがキャッシュ/スレッド間で正しく共有されていることを知る必要があります。

また、他の人が示唆するように、各タスクのスレッドを作成することは貧弱なモデルです。それはひどくスケールされ、あまりにも多くのタスクが提出された場合、あなたのアプリケーションはキリングに脆弱になります。メモリにはいくつかの制限がありますので、一度に多数の保留中のタスクしか持てませんが、スレッドの上限はずっと低くなります。 あなたがスピンしているので、それはすべて悪化します。スピンウェイトは、スレッドが条件を待つループに入る。より良いモデルは、ConditionVariableで待機するので、待機中のもの以外のスレッドは、待機中のものが準備ができている（または準備されている）ことが通知されるまで、オペレーティングシステムによって中断される可能性があります。

多くの場合、スレッドを作成および破棄するための時間とリソースに大きなオーバーヘッドがあります。ほとんどのプラットフォームは同時に比較的少数のスレッドしか実行できないので、大部分の時間を費やすために多くの「高価な」スレッドを作成し、スワップアウト（中断）しても何もしないことは非常に非効率的です。

正しいモデルは固定数のスレッド（または比較的固定された数）のプールを起動し、タスクを共有キューに配置して、スレッドの「取る」作業と処理を行います。

そのモデルは一般に「スレッドプール」として知られています。 ご覧になるエントリレベルの実装は、ThreadPoolExecutorです。 https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ThreadPoolExecutor.html