맵을 쓰레드 방식으로 채우고 그 맵을 백그라운드 스레드의 다른 메소드로 전달 하시겠습니까?

Question

add 메서드는 스레드 안전 방식으로 channelMessageHolder CHM을 채우는 다중 스레드에 의해 호출됩니다. 같은 클래스에서

, 나는 30 초마다 실행되는 backgrond에 스레드를 가지고 있고 그것은 channelMessageHolder에서 데이터를 전달하여 send 메소드를 호출합니다.

public class Processor { 
    private final ScheduledExecutorService executorService = Executors 
     .newSingleThreadScheduledExecutor(); 
    private final AtomicReference<ConcurrentHashMap<Channel, ConcurrentLinkedQueue<Message>>> channelMessageHolder = 
       new AtomicReference<>(new ConcurrentHashMap<Channel, ConcurrentLinkedQueue<Message>>()); 

    private Processor() { 
    executorService.scheduleAtFixedRate(new Runnable() { 
     @Override 
     public void run() { 
     send(); 
     } 
    }, 0, 30, TimeUnit.SECONDS); 
    } 

    // this will be called by only single background thread 
    private void send(ConcurrentHashMap<Channel, ConcurrentLinkedQueue<Message>> messageByChannels) { 
    for(Entry<Channel, ConcurrentLinkedQueue<Message>> entry : messageByChannels.entrySet()) { 
     Channel channel = entry.getKey(); 
     ConcurrentLinkedQueue<Message> messageHolder = entry.getValue(); 

     while (!messageHolder.isEmpty()) { 
     Message message = messageHolder.poll(); 
     .... 
     // process this and send to database 
     }  
    } 
    } 

    // called by multiple threads 
    public void add(final Channel channel, final Message message) { 
    // populate channelMessageHolder in a thread safe way 
    } 
} 

질문

는 channelMessageHolder 이미 내 Processor 클래스에 존재 볼 수 있듯이 그래서 명시 적으로 30 초마다이 방법을 보내려면이 맵에서 데이터를 전달해야합니까? 또는 직접 보내기 메서드에서 사용할 수 있습니까?

혼란은 직접 보내기 메서드에서 사용하면 동시에 다중 스레드로 채워집니다. 그래서 send 메서드로 전달하기 위해 AtomicReference의 getAndSet 메서드를 사용하고 있습니다.

내가하는 일이 잘못되었다는 사실을 알려주고 더 좋은 방법이 있습니까?

Answer 1

channelMessageHolder가 내 프로세서 클래스에 이미 존재하므로 30 초마다 메서드에서 데이터를 명시 적으로 전달해야합니까? 또는 직접 보내기 메서드에서 사용할 수 있습니까?

당신은 확실히 send() 방법에서 직접 사용할 수 있으며 ConcurrentHashMap가 이미 동기화되어 있기 때문에 당신은 AtomicReference 래퍼가 필요하지 않습니다. 걱정할 필요가있는 것은지도의 키와 값 개체가 제대로 동기화되고 있다는 것입니다. 나는 Channel이 불변이고 ConcurrentLinkedQueue이 동시라고 가정하므로 잘해야합니다. 생산자 스레드가 보낸 스레드가 충돌없이 항목을 보내는 동시에 그것으로 항목을 추가 할 수 있도록

// no need for AtomicReference 
private final ConcurrentHashMap<Channel, ConcurrentLinkedQueue<Message>> channelMessageHolder = 
    new ConcurrentHashMap<Channel, ConcurrentLinkedQueue<Message>>(); 

ConcurrentHashMap 당신을위한 동기화 처리합니다. AtomicReference은 여러 스레드간에 동기화되지 않은 클래스를 공유하려는 경우에만 필요합니다.

혼란은 내가 직접 내 send 메소드에서 사용한다면, 동시에 여러 스레드에 의해 채워질 것이다. 그래서 나는 메소드를 보내기 위해 AtomicReference의 getAndSet 메소드를 사용하고있다.

그래도 괜찮습니다. 여러 스레드가 ConcurrentLinkedQueue이라는 메시지를 추가합니다. 백그라운드 스레드가 시작될 때마다 30 초마다 Channel이 대기열에서 제외되고 그 순간 대기열에있는 메시지가 전송됩니다. ConcurrentLinkedQueue은 생산자와 소비자의 경쟁 조건을 보호합니다. 하나 개의 스레드 디큐있을 보이기 때문에 그것은 당신의 경우에 작동

while (!messageHolder.isEmpty()) { 
    Message message = messageHolder.poll(); 

:

코드에서이 문제는 큐를 여러 번 호출에 의존하기 때문에이 재진입되지 않는 것입니다 하지만 다음 코드가 더 좋습니다 :

while (true) { 
    // only one call to the concurrent queue 
    Message message = messageHolder.poll(); 
    if (message == null) { 
     break; 
    } 
    ... 
} 

Answer 2

아니면 내가 직접 문제없이 send 방법의 시작 부분에 channelMessageHolder.getAndSet(new ConcurrentHashMap<Channel, ConcurrentLinkedQueue<Message>>())을 말함으로써 당신은 직접 send 방법을 사용할 수 있어야합니다 아무것도

을 통과하지 않고 내 send 메소드에서 사용할 수 있습니다 . 말했다

, 자바 8 당신이 정말로 안 사용되는 AtomicReference 것을해야한다는 의미 ConcurrentHashMap 클래스에 computeIfAbsent라는 새로운 방법을 추가했습니다.