0002 Reactive Programming 3

Reactive Programming 2 편에서 다루었던 개발건에 추가적인 요구사항에 생겼다고 해보자, 

 

 

 요구사항

 

 ( 2편의 요구사항 )

   일정한 시간을 주기로 2000건의 정수 배열을 받은 후 해당 각각의 결과 값을 합하는 요구사항이 발생하였다.

   생성해야할 정수 배열은 난수로 이루어진 값으로 1000개의 아이템을 가지고 있다. 

 

 추가 요구사항 

    난수로 생성되는 2000건의 정수 배열에 대해서 각 배열의 총합이  음수인 값은 제외하고, 양수인 값을 추출하여하 출력한다.  

 

 ( 사실, 자바8의 Stream API를 사용한다면 이런 힘들고 복잡한 작업을 전혀할 필요가 없다. 우리는 Reactive Stream의 기본이 되는 개념을 익히고 이를 실질적으로 활용하기 위한 Akka, RXJava, Java9 Flow API 등등을 쉽게 이해하기 위해 진행 중인 것이다. )

 

---

2편에서 다룬 코드 샘플을 아래와 같이 변경하였다. 

아래의 코드에서 중요한 부분은 Publisher 1, Publisher 2를 연결하면서 처리되는 Data 흐름이다.  

데이터 전달 순서 : CustomPublisher -> FilterPublisher -> CustomSubscriber  

 

1번 : Main 진입점

public static void main(String[] args) {
	// 1000개의 난수 배열을 가진 아이템 생성 - 각각을 합쳐

	BlockingQueue<Command> blockingQueue = new ArrayBlockingQueue<Command>(500);

	for (int i = 0; i < 500; i++) {

	IntStream intStream = new Random().ints();

	List<Integer> integerList = intStream.limit(1000).boxed().collect(Collectors.toList());

	Command command = new IntegerSumCommand(integerList);

	blockingQueue.add(command);
}

// Publisher 1
Publisher<Command> publisher = new CustomPublisher(blockingQueue);

// Publisher 2
Publisher<Command> publisher1 = new FilterPublisher(publisher);

// Subscriber
Subscriber subscriber = new CustomSubscriber();

// Start !
publisher1.subscribe(subscriber);
}

2번 : Publisher 1 : Publisher를 구현한 CustomPublisher 

public class CustomPublisher implements Publisher<Command> {

    private final BlockingQueue<Command> blockingQueue;

    private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();

    public CustomPublisher(BlockingQueue<Command> blockingQueue){
        this.blockingQueue = blockingQueue;
    }

    @Override
    public void subscribe(Subscriber subscriber) {

        subscriber.onSubscribe(new Subscription() {
            @Override
            public void request(long n) {
                while (!blockingQueue.isEmpty()){
                    try {
                        Command command = blockingQueue.take();

                        subscriber.onNext(command);

                    } catch (InterruptedException e) {

                        e.printStackTrace();
                    }
                }

                scheduledExecutorService.schedule(() -> {
                    if(Counter.get() == 500){
                        subscriber.onComplete();

                        if(!scheduledExecutorService.isShutdown()){
                            scheduledExecutorService.shutdown();
                        }
                    }
                }, 1, TimeUnit.SECONDS);

            }

            @Override
            public void cancel() {

            }
        });
    }
}

 

3번 : Publisher 2 : Publisher 1을 전달 받아 값을 필터링하는 FilterPublisher

public class FilterPublisher implements Publisher<Command> {

    private Publisher<Command> publisher;

    public FilterPublisher(Publisher<Command> publisher){
        this.publisher = publisher;
    }

    @Override
    public void subscribe(Subscriber<? super Command> subscriber) {
        this.publisher.subscribe(new Subscriber<Command>() {
            @Override
            public void onSubscribe(Subscription subscription) {
                subscriber.onSubscribe(subscription);
            }

            @Override
            public void onNext(Command command) {
                if(!command.isMinus()){
                    subscriber.onNext(command);
                }
            }

            @Override
            public void onError(Throwable t) {
                subscriber.onError(t);
            }

            @Override
            public void onComplete() {
                subscriber.onComplete();
            }
        });
    }
}

4번 : Subscriber : 최종적으로 Publisher 2의 데이터 처리를 전달받아 최종값을 출력한다.

public class CustomSubscriber implements Subscriber<Command> {

    private ExecutorService executorService = Executors
            .newFixedThreadPool(1000);

    private Subscription subscription;

    @Override
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1);
    }

    @Override
    public void onNext(Command command) {
        executorService.execute(() -> {
            int total = command.getTotal();

            System.out.println(String.format("총 값은 %s", total));
        });
    }

    @Override
    public void onError(Throwable throwable) {
        subscription.cancel();
    }

    @Override
    public void onComplete() {
        if(!executorService.isShutdown()){
            executorService.shutdown();
        }
    }
}

 

5번 : Command Interface 

public interface Command {
    public int getTotal();
    public boolean isMinus();
}

 

6번 : Command Implementation 

public class IntegerHandlerCommand implements Command {

    private Optional<Integer> optionalInteger;

    public IntegerHandlerCommand(List<Integer> integerList){

        Counter.addAndGet();

        this.optionalInteger = integerList.stream().reduce(Integer::sum);
    }

    @Override
    public int getTotal() {
        return this.optionalInteger.orElse(0);
    }

    @Override
    public boolean isMinus() {
        return this.optionalInteger.orElse(0) < 0;
    }
}

 

7번 : Counter Class - Executor Service 종료를 위한 Counter 서비스 

public class Counter {
    private static int count = 0;

    public synchronized static int addAndGet(){
        return count ++;
    }

    public synchronized static int get(){
        return count;
    }
}