[ReactiveProgramming] Iterable, Iterator와 Observer Pattern

 

- 토비의 '스프링 리액티브 프로그래밍' 강의를 듣고 정리한 글입니다.

 

 

- 전체 소스코드) https://github.com/Cl8D/SpringReactive

GitHub - Cl8D/SpringReactive: Spring Reactive Programming 학습 용도 레파지토리입니다.

---

✔️ Iterable / Iterator

package chapter1;

        import java.util.Iterator;

public class Basic {
    public static void main(String[] args) {
        // Iterable을 구현한 구현체 생성
        Iterable<Integer> iter = () ->
        // Iterator라는 도구를 custom 하기
        new Iterator<Integer>() {
            // 1~10까지 생성
            int i = 0;
            final static int MAX = 10;

            @Override
            public boolean hasNext() {
                return i < MAX;
            }

            @Override
            public Integer next() {
                return ++i; // 다음 값 가져오기
            }
        };

        // custom이지만 for-each 구문을 사용할 수 있다는 것. (enhanced for)
        // 즉, 자바에서 for-each는 Iterable의 구현체에서 사용이 가능하다.
        for(Integer i : iter) {
            System.out.println(i);
        }

        // Java 10 버전
        for(Iterator<Integer> it = iter.iterator(); it.hasNext();) {
            // pull의 느낌, 값을 가져오는 것.
            System.out.println(it.next());
        }
    }
}

• 1~10까지 생성하는 Iterator 객체를 만들기.
• 이때 hatNext()와 next()를 오버라이딩해서 재정의할 수 있다.
  • hasNext()는 다음 요소가 있는지, next()는 다음 요소를 가져오는 것.
• for-each 구문은 이렇게 커스텀한 객체에도 사용할 수 있다.
  • 즉, Iterable의 구현체라면 enhanced for를 사용할 수 있는 것이다.

---

✔️ Observer

• Full Code

package chapter1;

        import java.util.Observable;
        import java.util.Observer;
        import java.util.concurrent.ExecutorService;
        import java.util.concurrent.Executors;

@SuppressWarnings("deprecation")
public class ObserverTest {

    /**
     * 1. 옵저버 패턴에게는 complete가 없다.
     * 2. 코드가 동작하다 오류가 발생하게 된다면 예외 전파 방식, 에외 처리 방식에 대해 어려워진다.
     */

    public static void main(String[] args) {
        // iterable 때문에 deprecate된 친구...
        Observer observer1 = new Observer() {
            // 일종의 subscriber
            // notifyObservers 동작 시 실행된다.
            // 즉, observable이 던진 데이터를 받아서 출력해주는 역할!
            @Override
            public void update(Observable o, Object arg) {
                System.out.println("observer1 = " + arg);
            }
        };

        Observer observer2 = new Observer() {
            @Override
            public void update(Observable o, Object arg) {
                System.out.println("observer2 = " + arg);
            }
        };

        Observer observer3 = new Observer() {
            @Override
            public void update(Observable o, Object arg) {
                System.out.println("observer3 = " + arg);
            }
        };

        IntObservable io = new IntObservable();
        // observer를 등록해준다. 이러면 observable이 던진 작업을 등록된 옵저버들이 받는다.
        // 실행 순서의 경우 가장 마지막에 등록해준 옵저버부터 순차적으로 실행되는 것 같다.
        io.addObserver(observer1);
        io.addObserver(observer2);
        io.addObserver(observer3);

        // Executors를 통해서 ExecutorService 생성 가능
        // 스레드를 새롭게 생성해준 다음에 할당해주기
        ExecutorService es = Executors.newSingleThreadExecutor();
        es.execute(io);

        // execute보다 나중에 입력했는데도 더 먼저 실행된다.
        // excute는 스레드를 새롭게 생성해서 (별개의 스레드) 동작하기 때문에 더 나중에 동작하게 되는 것
        System.out.println("Hello!");
        es.shutdown();
    }

    // observable은 외부에서 이벤트를 만들어내야 한다.
    // 비동기로 실행하기 위해서 runnable 구현
    static class IntObservable extends Observable implements Runnable {
        @Override
        public void run() {
            for(int i=1; i<=10; i++) {
                setChanged(); // 변화가 생겼음을 알려주기 위해서, Observable에 등록되어 있음.
                // 일종의 Push 느낌. 사실 iter에서의 it.next()와 비슷한 느낌이다.
                notifyObservers(i); // 옵저버에게 알려주기!, 일종의 publisher
            }
        }
    }
}

• 옵저버 패턴
  • complete()가 없다.
  • 코드가 동작하다 오류가 발생하면 예외가 전파되며, 이 경우 예외 처리에 대해 까다로워진다.
• Observer 객체
  • 일종의 subscriber로서, notifyObservers가 동작하면 실행된다.
  • 즉, observable이 던진 데이터를 받아서 출력해주는 역할을 한다.

  Observer observer1 = new Observer() {
@Override
public void update(Observable o, Object arg) {
        System.out.println("observer1 = " + arg);
        }
        };

        Observer observer2 = new Observer() {
@Override
public void update(Observable o, Object arg) {
        System.out.println("observer2 = " + arg);
        }
        };

        Observer observer3 = new Observer() {
@Override
public void update(Observable o, Object arg) {
        System.out.println("observer3 = " + arg);
        }
        };

• IntObservable
  • IntObservable에 옵저버를 등록해주는 걸 볼 수 있는데, observable이 던진 작업을 등록된 옵저버들이 받게 된다.
  • 실행은 가장 마지막으로 등록한 옵저버부터 역으로 수행된다.
  • observer3 = 1
  • observer2 = 1
  • observer1 = 1
  • observer3 = 2
  • observer2 = 2
  • observer1 = 2 ...

        IntObservable io = new IntObservable();
        io.addObserver(observer1);
        io.addObserver(observer2);
        io.addObserver(observer3);

• 다른 스레드에서 실행되도록 생성
  • main 스레드가 아닌 별개의 스레드에서 실행하였기 때문에 Hello! 실행 뒤에 순차적으로 결과값이 출력된다.

      ExecutorService es = Executors.newSingleThreadExecutor();
        es.execute(io);

        System.out.println("Hello!");
        es.shutdown();

• 외부에서 이벤트 생성해주기
  • Runnable 인터페이스를 구현한 것을 볼 수 있는데, 이는 비동기로 실행하기 위함이다.
  • Observable의 setchanged() 메서드를 호출하면 Observable 객체에 변화가 생겼음을 나타내며 내부적으로 changed 변수가 true로 세팅된다.
  • notifyObservers() 메서드를 호출하면 일종의 publisher로서, 등록된 Observer들에게 변화된 내용을 업데이트해준다.

static class IntObservable extends Observable implements Runnable {
    @Override
    public void run() {
        for(int i=1; i<=10; i++) {
            setChanged(); // 변화가 생겼음을 알려주기 위해서, Observable에 등록되어 있음.
            // 일종의 Push 느낌. 사실 iter에서의 it.next()와 비슷한 느낌이다.
            notifyObservers(i); // 옵저버에게 알려주기!, 일종의 publisher
        }
    }
}