重要なメッセージが届いた場合にメッセージキューに追加

Question

優先度が最も高いメッセージを最初に送信する必要がありますので、PriorityQueueを使用します。

ただし、キューもソートセットのように動作します。したがって、PriorityQueueは、既存のメンバーを繰り返す挿入を無視するように変更しました。

import java.util.Comparator; 
import java.util.PriorityQueue; 

public class PrioritySet<E> extends PriorityQueue<E> { 

    private static final long serialVersionUID = 34658778L; 

    public PrioritySet() { 
     super(); 
    } 

    public PrioritySet(int initialCapacity, Comparator<? super E> comparator) { 
     super(initialCapacity, comparator); 
    } 

    @Override 
    public boolean offer(E e) { 
     boolean isAdded = false; 
     if(!super.contains(e)) { 
      isAdded = super.offer(e); 
     } 
     return isAdded; 
    } 
} 

私たちのアプリ固有のデータ構造の実装。ここで

import java.util.Comparator; 

public class MessagePrioritySet extends PrioritySet<MessageData> { 

    private static final long serialVersionUID = 34658779L; 

    private int minPriorityNumber; 

    public MessagePrioritySet() { 
     super(); 
    } 

    public MessagePrioritySet(int initialCapacity, Comparator<MessageData> comparator) { 
     super(initialCapacity, comparator); 
    } 

    public synchronized int getMinPriorityNumber() { 
     return minPriorityNumber; 
    } 

    public synchronized void setMinPriorityNumber(int minPriorityNumber) { 
     this.minPriorityNumber = minPriorityNumber; 
    } 

    @Override 
    public synchronized boolean offer(MessageData notification) { 
     boolean isAdded = super.offer(notification); 
     if (notification.getPriority() < minPriorityNumber) 
      minPriorityNumber = notification.getPriority(); 
     return isAdded; 
    } 

    public synchronized void reportSent(MessageData notification) { 
     MessageData nextMessageData = peek(); 
     if (nextMessageData == null) 
      minPriorityNumber = 0; 
     else if (nextMessageData.getPriority() > notification.getPriority()) 
      minPriorityNumber = nextMessageData.getPriority(); 
    } 
} 

、我々はそのためのインスタンス変数を宣言するように、データ構造がメッセージの最低優先度の値を認識することにしたいです。着信メッセージの優先度がチェックされ、この優先度が格納された値よりも低い場合、格納された値が更新されます。送信されたメッセージを報告するには、クラスの使用が必要です。データ構造の他のメンバが削除されたものと同じ優先度を持たない場合、次の要素の優先度が格納された優先度になります。

2つのスレッドが実装されたキューを共有します。 1つのスレッドがデータベースからデータをフェッチし、それらをキューに挿入します。もう一方はキューを読み取り、最も低い優先度の番号を持つ最優先のメッセージを送信します。キューは最小優先度値を0に設定し、データベースからデータをフェッチするスレッドは、格納された最小値がゼロでない場合、キューに格納されている最小値以下の優先度値を持つローを読み取ります。キュー内の現在のメッセージが送信されている間は、すでにキューに入っているものよりも重要な新しいメッセージだけがキューに追加されます。

スレッド内のwhileループの操作はアトミックでなければならず、アトムにする方法を教えてくれる人には感謝しています。

private void startMptSender() { 
    sleepInterval = 1000; 
    final MessagePrioritySet messagePrioritySet = new MessagePrioritySet(); 

    Runnable mptReader = new Runnable() { 

     @Override 
     public void run() { 
      while (true) { 
       List<MessageData> messageDataList; 

       if (messagePrioritySet.getMinPriorityNumber() == 0) 
        messageDataList = messageDao.readSMSMpt(); 
       else 
        messageDataList = messageDao.readSMSMpt(messagePrioritySet.getMinPriorityNumber()); 

       for (MessageData messageData : messageDataList) { 
        messagePrioritySet.offer(messageData); 
       } 
       try { 
        Thread.sleep(sleepInterval); 
       } catch (InterruptedException ie) { 

       } 
      } 
     } 
    }; 

    executor.execute(mptReader); 

    Runnable mptPusher = new Runnable() { 

     @Override 
     public void run() { 
      while (status) { 
       if (messagePrioritySet.size() > 0) { 

        while (messagePrioritySet.size() != 0) { 
         MessageData noti = messagePrioritySet.remove(); 
         mptSender.sendSms(noti); 
         messageDao.markNotificationAsRead(noti.getSyskey()); 
         messagePrioritySet.reportSent(noti); 
         try { 
          Thread.sleep(sleepInterval); 
         } catch (InterruptedException e) { 
          e.printStackTrace(); 
         } 
        } 
       } else { 
        try { 
         Thread.sleep(sleepInterval); 
        } catch (InterruptedException e) { 
         e.printStackTrace(); 
        } 
       } 
      } 
     } 
    }; 

    executor.execute(mptPusher); 
} 

}

Answer 1

私はあなたに変換原子で何を意味するかと仮定します。あなたは、各スレッドが他のスレッドによって中断されずに一回の反復ため、その作業のすべてをやっていることを望みます。

つまり、（おそらく複数の）操作があります。スレッドAが操作を行っている間はスレッドBは何もしてはいけません.BがAによって行われた更新の「完全なセット」しか見ないようにしたいからです。

確かに、たとえば、1つのintに書き込む場合、たとえば、AtomicIntegerを使用することができます。しかし、いくつかの操作について話しているときには、別のものが必要です。

"強力な"ソリューションは、何らかのロックを追加することです。意味：あなたのスレッドはいくつかのLOCKオブジェクトを共有します。 1つのスレッドが「クリティカルセクション」に入るたびに、最初にそのLOCKを獲得する必要があります（そして、その後は直接解放されます）。しかし、これは非常に慎重に設計する必要があります。そのスレッドを長時間保持することによってスレッドAが「飢えている」Bにならないようにします。

あなたのコードをもう一度見てみると、おそらくminPriorityをAtomicIntegerにしようとする可能性があります。問題は、キューの「サイズ」を操作している他のスレッドにどう関連するかということです。