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

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

import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;

public class ConcreteSubject implements Subject<String> {
    /**
     * multi-thread 안정성 유지를 위해 업데이트시마다 새 복사본을 생성하는 Set 구현체 사용
     * 복사 비용은 큼 -> 구독자 목록 변경은 거의 없음.
     */
    private final Set<Observer<String>> observers = new CopyOnWriteArraySet<>();

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

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

    @Override
    public void notifyObservers(String event) {
        observers.forEach(observers -> observers.observe(event)); // notify observer
    }
}

public class ConcreteObserverA implements Observer<String> {
    @Override
    public void observe(String event) {
        System.out.println("Observer A:" + event);
    }
}

public class ConcreteObserverB implements Observer<String> {
    @Override
    public void observe(String event) {
        System.out.println("Observer B:" + event);
    }
}

    @Test
    public void observerHandleEventsFromSubject() {
        // given
        Subject<String> subject = new ConcreteSubject();
        Observer<String> observerA = new ConcreteObserverA();
        Observer<String> observerB = new ConcreteObserverB();

        // when
        subject.notifyObservers("no subscriber!");
        subject.registerObserver(observerA);
        subject.registerObserver(observerB);

        subject.notifyObservers("event !");

        subject.unregisterObserver(observerB);

        subject.notifyObservers("event for A");
        subject.unregisterObserver(observerA);

        subject.notifyObservers("no subscriber");
    }

    @Test
    public void subjectLeverageLambdas() {
        Subject<String> subject = new ConcreteSubject();
        subject.registerObserver(new Observer<String>() {
            @Override
            public void observe(String event) {
                System.out.println("A : " + event);
            }
        });
        subject.registerObserver(event -> System.out.println("B : " + event));
        subject.notifyObservers("event start for A & B");
    }

    @Test
    public void subjectThreadTest() {
        Subject<String> subject = new ConcreteSubject();
        subject.registerObserver(event -> { // A observer 는 이벤트 처리시간이 길다. 
            try {
                Thread.sleep(3000);
            } catch (Exception e) {

            }

            System.out.println(Thread.currentThread().getName());
            System.out.println("A: " + event);
        });

        subject.registerObserver(event -> {
            System.out.println(Thread.currentThread().getName());
            System.out.println("B: " + event);
        });

        subject.notifyObservers("event!");

        try {
            Thread.sleep(3100);
        } catch (Exception e) {

        }
    }

public class ConcreteParallelSubject implements Subject<String> {
    private final ExecutorService ex;

    public ConcreteParallelSubject(ExecutorService ex){
        this.ex = ex;
    }
    /**
     * multi-thread 안정성 유지를 위해 업데이트시마다 새 복사본을 생성하는 Set 구현체 사용
     * 복사 비용은 큼 -> 구독자 목록 변경은 거의 없음.
     */
    private final Set<Observer<String>> observers = new CopyOnWriteArraySet<>();


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

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

    @Override
    public void notifyObservers(String event) {
        observers.forEach(observers -> {
            ex.submit(() -> observers.observe(event));
        });
    }
}

    @Test
    public void subjectParallelTest() throws InterruptedException {
        final ExecutorService ex = Executors.newFixedThreadPool(10);

        Subject<String> parallelSubject = new ConcreteParallelSubject(ex);
        parallelSubject.registerObserver(event -> { // 이벤트 처리 시간이 길다. 
            try {
                Thread.sleep(3000);
            } catch (Exception e) {

            }

            System.out.println(Thread.currentThread().getName()); // thread name 
            System.out.println("A : " + event); 
        });

        parallelSubject.registerObserver(event -> {
            System.out.println(Thread.currentThread().getName());
            System.out.println("B : " + event);
        });

        parallelSubject.notifyObservers("event!");

        ex.awaitTermination(3100, TimeUnit.MILLISECONDS); // 모든 이벤트가 종료될때까지 기다린다. 
    }

public ParallelFlux<Map<String, Object>> parallelAttributes(OneInfo info) {
    Mono<ComponentRequest> request = fetcher.createRequest(info);
    return Flux.from(request) // Flux<ComponentRequest>
               .flatMap(fetcher::receiveFlux) // Flux<ComponentReponse>
               .parallel(core * 2) // ParallelFlux<ComponentResponse>
               .runOn(Schedulers.parallel())
               .map(received -> handler.handleResponse(received, info)));
    }

public Mono<Map<String, Object>> attributes(OneInfo info) { 
    Mono<ComponentRequest> request = fetcher.createRequest(info);
    return Mono.from(request)
               .flatMap(fetcher::receiveMono) // Mono<List<ComponentResponse>>
               .map(received -> handler.handleResponse(received, info));
    }

/**
 * Prepare this {@link Flux} by dividing data on a number of 'rails' matching the
 * provided {@code parallelism} parameter, in a round-robin fashion. Note that to
 * actually perform the work in parallel, you should call {@link ParallelFlux#runOn(Scheduler)}
 * afterward.
 *
 * <p>
 * <img class="marble" src="doc-files/marbles/parallel.svg" alt="">
 *
 * @param parallelism the number of parallel rails
 *
 * @return a new {@link ParallelFlux} instance
 */
public final ParallelFlux<T> parallel(int parallelism) {
	return parallel(parallelism, Queues.SMALL_BUFFER_SIZE);
}

/**
 * Specifies where each 'rail' will observe its incoming values with no work-stealing
 * and default prefetch amount.
 * <p>
 * This operator uses the default prefetch size returned by {@code
 * Queues.SMALL_BUFFER_SIZE}.
 * ----- 중략 ------
 * @param scheduler the scheduler to use
 *
 * @return the new {@link ParallelFlux} instance
 */
public final ParallelFlux<T> runOn(Scheduler scheduler) {
	return runOn(scheduler, Queues.SMALL_BUFFER_SIZE);
}

/**
 * A small default of available slots in a given container, compromise between intensive pipelines, small
 * subscribers numbers and memory use.
 */
public static final int SMALL_BUFFER_SIZE = Math.max(16,
			Integer.parseInt(System.getProperty("reactor.bufferSize.small", "256")));

return Flux.from(request) // Flux<ComponentRequest>
           .flatMap(fetcher::receiveFlux) // Flux<ComponentReponse>
           .parallel(core * 2) // ParallelFlux<ComponentResponse>
           .runOn(Schedulers.parallel())
           .map(received -> handler.handleResponse(received, info)));

return Flux.from(request) // Flux<ComponentRequest>
           .flatMap(fetcher::receiveFlux) // Flux<ComponentReponse>
           .parallel(core * 2) // ParallelFlux<ComponentResponse>
           .runOn(Schedulers.parallel())
           .log()
           .map(received -> handler.handleResponse(received, info))
           .doOnComplete(() -> log.info("complete"));

    public ParallelFlux<Map<String, Object>> parallelAttributes(OneInfo info) {
        Mono<ComponentRequest> request = fetcher.createRequest(info);
        return Flux.from(request)
                   .flatMap(fetcher::receiveFlux)
                   .parallel(3)
                   .runOn(Schedulers.parallel())
                   .map(received -> handler.handleResponse(received, info));
    }

  flux.subscribeOn(Schedulers.single()).subscribe() 

public Flux<Map<String, Object>> fluxAttributes(OneInfo info) {
        Mono<ComponentRequest> request = fetcher.createRequest(info);
        return Flux.from(request)
                   .flatMap(fetcher::receiveFlux)
                   .subscribeOn(Schedulers.parallel())
                   .map(received -> handler.handleResponse(received, info));
    }

public enum OneBody {

    /** ARTICLE **/
    ARTICLE("ARTICLE", Arrays.asList(OneTail.ARTICLE, OneTail.OFFICE_INFO)),
    ARTICLE_COMMENT("ARTICLE_COMMENT", Arrays.asList(OneTail.ARTICLE_COMMENT)),

    /** OFFICE **/
    OFFICE_INFO("OFFICE_INFO", Arrays.asList(OneTail.OFFICE_INFO)),
    OFFICE_HEADLINE("OFFICE_HEADLINE", Arrays.asList(OneTail.OFFICE_HEADLINE)),
    OFFICE_RANKING("OFFICE_RANKING", OneTail.getAllDependencies("OFFICE_RANKING"));
    ... 생략 ...
}

public enum OneTail {
    /** ARTICLE **/
    ARTICLE("ARTICLE"),
    ARTICLE_NEWS("ARTICLE_NEWS"),
    ARTICLE_COMMENT("ARTICLE_COMMENT"),

    /** OFFICE **/
    OFFICE_INFO("OFFICE_INFO"),
    OFFICE_HEADLINE("OFFICE_HEADLINE"),
    OFFICE_RANKING_VIEW("OFFICE_RANKING_VIEW"),
    OFFICE_RANKING_ALL_AGE10("OFFICE_RANKING_ALL_AGE10"),
    OFFICE_RANKING_ALL_AGE20("OFFICE_RANKING_ALL_AGE20"),
    OFFICE_RANKING_ALL_AGE30("OFFICE_RANKING_ALL_AGE30"),
    OFFICE_RANKING_ALL_AGE40("OFFICE_RANKING_ALL_AGE40"),
    OFFICE_RANKING_ALL_AGE50("OFFICE_RANKING_ALL_AGE50"),
    OFFICE_RANKING_ALL_AGE60("OFFICE_RANKING_ALL_AGE60");
    ... 생략 ...
}

    public Mono<Map<String, Object>> attributes(OneInfo info) {
        Mono<ComponentRequest> request = fetcher.createRequest(info); 
        return Mono.from(request)
                   .flatMap(fetcher::receiveMono) // Mono<List<ComponentResponse>>
                   .map(received -> handler.handleResponse(received, info)); // Mono<Map<String, Object>>
    }

    public ParallelFlux<Map<String, Object>> parallelAttributes(OneInfo info) {
        Mono<ComponentRequest> request = fetcher.createRequest(info);
        return Flux.from(request) // Flux<ComponentRequest>
                   .flatMap(fetcher::receiveFlux) // Flux<ComponentReponse>
                   .parallel(3) // ParallelFlux<ComponentResponse>
                   .runOn(Schedulers.parallel()) // fixed thread pool에서 parallel 실행 
                   .map(received -> handler.handleResponse(received, info)); // ParallelFlux<Map<String, Object>>
    }

    public Flux<Map<String, Object>> fluxAttributes(OneInfo info) {
        Mono<ComponentRequest> request = fetcher.createRequest(info);
        return Flux.from(request) // Flux<ComponentRequest>
                   .flatMap(fetcher::receiveFlux)  // Flux<ComponentReponse>
                   .subscribeOn(Schedulers.parallel())
                   .map(received -> handler.handleResponse(received, info));  
                   // 1 thread(in thread-pool) subscribe
    }