리액티브 프로그래밍의 관찰자패턴

리액티브 프로그래밍의 관찰자 패턴

관찰자 패턴은 리액티브 프로그래밍의 기초이다. 관찰자 패턴은 관찰자라고 불리는 자손의 리스트를 가지고 있는 주체(subject)를 필요로한다. 주체는 자신의 메서드 중 하나를 호출하여 관찰자에게 상태 변경을 알린다. 관찰자 패턴은 이벤트 처리를 기반으로 시스템을 구현할 때 필수적이다. MVC(Model-View_Controller) 패턴의 중요한 부분으로, 거의 모든 UI 라이브러리가 내부적으로 이 패턴을 사용한다.

 

 

 

관찰자 패턴의 구현

Observer(관찰자)는 Subject(주체)에 등록되고 Subject 로부터 알림을 받는다. 위에서 설명한 상태변경을 알린다. 라는 말이 이에 해당한다.

Observer (관찰자) 2개
Subject (주체) 1개

 

Subject 인터페이스를 구현해보자.

 

public interface Subject<T> {
  void registerObserver(Observer<T> observer);
}

 

제네릭 인터페이스는 이벤트 타입 T를 사용하여 프로그램의 타입 안정성을 향상시킨다.

Observer 인터페이스를 구현해보자.

 

public interface Observer<T> {
  void observe(T event);
}

 

Oberver는 T 타입으로 매개변수화한 일반 인터페이스이다. 이벤트를 처리하는 observe 메서드를 가지고 있고, 이 Observer(관찰자)와 Subject(주체) 모두 인터페이스에 기술된 것 이상은 서로 알지 못한다.

 

위 Observer<T> 인터페이스를 구현하는 Observer을 만들어보자.

 

public class ConcreateObserverA implements Observer<String> {
  @Override
  public void observe(Stirng event) {
    log.info("ovserve A : " + event)
  }
}

public class ConcreateObserverB implements Observer<String> {
  @Override
  public void observe(Stirng event) {
    log.info("ovserve B : " + event)
  }
}

 

위 Subject<T> 인터페이스를 구현하는 Observer을 만들어보자.

 

public class ConcreateSubject implements Subject<String> {
  // notify를 받는데 관심이 있는 객체
  // CopyOnWriteArraySet : 스레드 안정성을 유지하기위해 업데이트 작업이 발생할 때마다 새 복사본을 생성하는 Set 구현체
  // CopyOnWriteArraySet의 내용을 업데이트하는 것은 상대적으로 비용이 많이들지만, 자주 변경되지 않는다면 스레드 세이프한 구현을 위해 사용해도 좋다.
  private final Set<Observer<String>> observers = new CopyOnWriteArraySet<>();

  public void registerObserver(Observer<String> observer) {
    observers.add(observer);
  }

  public void unregisterObserver(Observer<String> observer) {
    observers.remove(observer);
  }

  // 이벤트를 브로드캐스트 하기 위해 각 Observer에 대해 반복적으로 observe() 메서드를 호출한다.
  // 
  public void notifyObservers(String event) {
    observers.forEach(observer -> observer.observe(event));
  }
}

 

위 notifyObservers 메서드를 병렬 메서드로 구현해보자.

 

private final ExcecuteService executorService = Executor.newCachedThreadPool();

public void notifyObservers(String event) {
    observers.forEach(observer -> executorService.submit(
      () -> observer.observe(event)
    )
  );
}

 

병렬 메서드로의 구현은 직접 개발한 솔루션에서 자주 발생하는 비효율성 및 내재된 버그를 포함하는 파악하기 어려운 코드를 만들 수 있다. 스레드 풀 크기를 제한하는 것을 깜박한다면 OutOfMemoryError를 발생시킬 것이다. 계속해서 스레드를 생성하게 되는데, JVM 응용 프로그램은 단 몇 천개의 스레드만으로도 사용 가능한 메모리를 모두 소모할 수 있다.

 

 

 

위 코드들의 테스트 코드 작성

package com.example.chapter02;

import com.example.redo.chapter02.*;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;

import static org.mockito.Mockito.times;

public class ObserverTest {
    @Test
    public void observersHandleEventsFromSubjectWithAssertions() {
        // given
        Subject<String> subject = new ConcreteSubject();
        Observer<String> observerA = Mockito.spy(new ConcreteObserverA());
        Observer<String> observerB = Mockito.spy(new ConcreteObserverB());

        // when
        subject.notifyObservers("No listeners"); /* 출력이 안되겠지? 관찰자가 현재 시점에 없으므로 */

        subject.registerObserver(observerA); /* 구독 등록 */
        subject.notifyObservers("Message for A"); /* 출력이 되겠지 */

        /** observer : observerA */

        subject.registerObserver(observerB); /* 구독 등록 */

        /** observer : observerA, observerB */
        subject.notifyObservers("Message for A & B"); /* 출력 O (2번 출력, 구독자가 2명임) */

        subject.unregisterObserver(observerA); /* 구독 취소 */
        subject.notifyObservers("Message for B"); /* 출력 O */

        subject.unregisterObserver(observerB); /* 구독 취소 */

        /** observer : X */
        subject.notifyObservers("No listeners");  /* 출력 X */

        // then
        Mockito.verify(observerA, times(1)).observe("Message for A");
        Mockito.verify(observerA, times(1)).observe("Message for A & B");
        Mockito.verifyNoMoreInteractions(observerA);

        Mockito.verify(observerB, times(1)).observe("Message for A & B");
        Mockito.verify(observerB, times(1)).observe("Message for B");
        Mockito.verifyNoMoreInteractions(observerB);
    }
}