Executor/Queue 프로세스가 마지막으로 알려진 작업 만

Question

이벤트를 처리 할 동시 코드를 작성하려고합니다. 이 처리에는 오랜 시간이 걸릴 수 있습니다.

이벤트가 처리되는 동안 들어오는 이벤트를 기록한 다음 다시 실행하도록 요청할 때 마지막 들어오는 이벤트를 처리해야합니다. (다른 이벤트는 버려 질 수 있습니다.) 이것은 FILO 큐와 조금 비슷하지만 큐에 하나의 요소 만 저장하면됩니다.

필자는 아래에 표시된 내 이벤트 처리 아키텍처에 새로운 Executor를 연결하고 싶습니다.

public class AsyncNode<I, O> extends AbstractNode<I, O> { 
    private static final Logger log = LoggerFactory.getLogger(AsyncNode.class); 
    private Executor executor; 

    public AsyncNode(EventHandler<I, O> handler, Executor executor) { 
     super(handler); 
     this.executor = executor; 
    } 

    @Override 
    public void emit(O output) { 
     if (output != null) { 
      for (EventListener<O> node : children) { 
       node.handle(output); 
      } 
     } 
    } 

    @Override 
    public void handle(final I input) { 

     executor.execute(new Runnable() { 

      @Override 
      public void run() { 
       try{ 
       emit(handler.process(input)); 
       }catch (Exception e){ 
        log.error("Exception occured whilst processing input." ,e); 
        throw e; 
       } 

      } 
     }); 

    } 

}

Answer 1

public class LatestTaskExecutor implements Executor { 
    private final AtomicReference<Runnable> lastTask =new AtomicReference<>(); 
    private final Executor executor; 

    public LatestTaskExecutor(Executor executor) { 
     super(); 
     this.executor = executor; 
    } 

    @Override 
    public void execute(Runnable command) { 
     lastTask.set(command); 
     executor.execute(new Runnable() { 
      @Override 
      public void run() { 
       Runnable task=lastTask.getAndSet(null); 
       if(task!=null){ 
        task.run(); 
       } 
      } 
     }); 

    } 
} 

@RunWith(MockitoJUnitRunner.class) 
public class LatestTaskExecutorTest { 

    @Mock private Executor executor; 
    private LatestTaskExecutor latestExecutor; 
    @Before 
    public void setup(){ 
     latestExecutor=new LatestTaskExecutor(executor); 
    } 
    @Test 
    public void testRunSingleTask() { 
     Runnable run=mock(Runnable.class); 
     latestExecutor.execute(run); 
     ArgumentCaptor<Runnable> captor=ArgumentCaptor.forClass(Runnable.class); 
     verify(executor).execute(captor.capture()); 
     captor.getValue().run(); 
     verify(run).run(); 
    } 

    @Test 
    public void discardsIntermediateUpdates(){ 
     Runnable run=mock(Runnable.class); 
     Runnable run2=mock(Runnable.class); 
     latestExecutor.execute(run); 
     latestExecutor.execute(run2); 
     ArgumentCaptor<Runnable> captor=ArgumentCaptor.forClass(Runnable.class); 
     verify(executor,times(2)).execute(captor.capture()); 
     for (Runnable runnable:captor.getAllValues()){ 
      runnable.run(); 
     } 
     verify(run2).run(); 
     verifyNoMoreInteractions(run); 
    } 
} 

Answer 2

나도하지 않을 것입니다. 내가 처리하고자하는 이벤트에 대한 AtomicReference가 있고 그것을 파괴적인 방식으로 처리하기위한 작업을 추가 할 것입니다. 집행자 키 입력 이벤트를 데 (다른 것들에 사용할 수 있습니다) 집행자 또는 큐

이것은 또한 확장을 정의하지 않고, 무료 때

final AtomicReference<Event> eventRef = 

public void processEvent(Event event) { 
    eventRef.set(event); 
    executor.submit(new Runnable() { 
     public vodi run() { 
      Event e = eventRef.getAndSet(null); 
      if (e == null) return; 
      // process event 
     } 
    } 
} 

이것은 오직 다음 이벤트를 처리 할 원하는 즉, 키의 마지막 이벤트를 처리합니다.

Answer 3

이 답변은 불필요한 작업 제출을 최소화하는 DD에서 수정 된 버전입니다.

atomic reference은 최신 이벤트를 추적하는 데 사용됩니다. 이벤트를 잠재적으로 처리하기 위해 사용자 정의 작업이 대기열에 제출되고, 최신 이벤트를 읽는 작업 만 실제로 진행되고 null에 대한 원자 참조를 지우기 전에 유용한 작업을 수행합니다. 다른 작업을 실행할 기회가 생겨도 처리 할 수있는 이벤트가 없다면 아무 것도하지 않고 조용히 버립니다. 불필요한 작업 제출은 대기열에서 사용 가능한 작업 수를 추적하여 피할 수 있습니다. 대기열에 대기중인 작업이 하나 이상 있으면 이미 대기중인 작업이 대기열에서 제외 될 때 이벤트가 처리되므로 작업 제출을 피할 수 있습니다.

import java.util.concurrent.Executor; 
import java.util.concurrent.atomic.AtomicInteger; 
import java.util.concurrent.atomic.AtomicReference; 

public class EventExecutorService implements Executor { 

    private final Executor executor; 
    // the field which keeps track of the latest available event to process 
    private final AtomicReference<Runnable> latestEventReference = new AtomicReference<>(); 
    private final AtomicInteger activeTaskCount = new AtomicInteger(0); 

    public EventExecutorService(final Executor executor) { 
     this.executor = executor; 
    } 

    @Override 
    public void execute(final Runnable eventTask) { 
     // update the latest event 
     latestEventReference.set(eventTask); 
     // read count _after_ updating event 
     final int activeTasks = activeTaskCount.get(); 

     if (activeTasks == 0) { 
      // there is definitely no other task to process this event, create a new task 
      final Runnable customTask = new Runnable() { 
       @Override 
       public void run() { 
        // decrement the count for available tasks _before_ reading event 
        activeTaskCount.decrementAndGet(); 
        // find the latest available event to process 
        final Runnable currentTask = latestEventReference.getAndSet(null); 
        if (currentTask != null) { 
         // if such an event exists, process it 
         currentTask.run(); 
        } else { 
         // somebody stole away the latest event. Do nothing. 
        } 
       } 
      }; 
      // increment tasks count _before_ submitting task 
      activeTaskCount.incrementAndGet(); 
      // submit the new task to the queue for processing 
      executor.execute(customTask); 
     } 
    } 
}