더 중요한 메시지가 들어오는 경우 메시지 대기열에 추가

Question

우선 순위가 가장 높은 메시지를 먼저 보내야합니다. 따라서 우리는 용도로 PriorityQueue을 사용합니다.

PriorityQueue<MessageData> queue = new PriorityQueue<MessageData>(); 

그러나 우리는 또한 대기열이 정렬 된 집합처럼 행동하기를 원합니다. 따라서 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(); 
    } 
} 

여기서 우리는 데이터 구조가 메시지의 최소 우선 순위 값을 알고 있기 때문에 이에 대한 인스턴스 변수를 선언합니다. 들어오는 메시지의 우선 순위가 검사되고이 우선 순위가 저장된 값보다 낮 으면 저장된 값이 업데이트됩니다. 보낸 메시지를보고하려면 클래스를 사용해야합니다. 데이터 구조의 다른 멤버가 우선 순위가 낮은 멤버가없는 경우 다음 요소의 우선 순위가 저장된 우선 순위가됩니다.

두 스레드가 구현 된 큐를 공유합니다. 한 스레드는 데이터베이스에서 데이터를 가져 와서 큐에 삽입합니다. 다른 하나는 대기열을 읽고 우선 순위 번호가 가장 낮은 우선 순위가 가장 높은 메시지를 보냅니다. 대기열은 최소 우선 순위 값을 0으로 설정하고 데이터베이스에서 데이터를 가져 오는 스레드는 저장된 최소값이 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에 의한 업데이트의 "완전한 세트"만을 볼 수 있기를 원하므로

예를 들어 AtomicInteger를 사용할 수 있습니다. 하지만 몇 가지 작업에 대해 이야기 할 때 ... 다른 것이 필요합니다.

"무력"해결책은 일종의 잠금을 추가하는 것입니다. 의미 : 당신의 쓰레드는 LOCK 객체를 공유합니다. 한 스레드가 "임계 영역"에 진입 할 때마다 먼저 LOCK을 획득해야합니다 (물론 나중에 직접 릴리스해야 함). 그러나 이것은 매우 신중하게 설계해야합니다. 스레드 A가 그 잠금을 너무 길게 잡아서 B가 "굶주 리지"않는지 확인하고 싶습니다.

코드를 다시 보면서 좀 더 자세히 살펴보면 minPriority를 ​​AtomicInteger로 만들 수 있습니다. 문제는 대기열의 "크기"를 작동하는 다른 스레드와 어떻게 관련이 있는지입니다.