외부 라이브러리가 아닌 순수 자바에서 제공하는 Flow 구현체가 있을까?(이장안)

[lja3723]

문제가 무엇인가?

외부 라이브러리가 아닌 Flow 구현체가 있는가?

왜 이런 문제를 선정하였는가?

551페이지에서 Flow API는 자바에서 기본으로 제공하는 구현이 없다고 소개하고 있다. 정말 외부 라이브러리 도움이 없다면 개발자가 직접 구현해야 하는 것일까? 아니면 뭔가가 있을까? 알아보자.

자신이 생각한 답변은 무엇인가?

구현체가 정말 없다면 이런 이슈가 있을 수 없었을 것이다. 책과 달리, 외부 라이브러리가 아닌 자바에서 제공하는 Flow API의 구현체가 존재한다. 아래 코드를 살펴보자.

package modern.ch17;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;

public class FlowExample {
    public static void main(String[] args) throws InterruptedException {
        // Publisher 생성
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        // Subscriber 생성 및 구독
        publisher.subscribe(getSubscriber("1"));
        publisher.subscribe(getSubscriber("2"));

        // 데이터 전송
        System.out.println("Publishing items...");
        publisher.submit("Hello");
        publisher.submit("world!");
        publisher.submit("java");

        //중간에 구독자 추가
        publisher.subscribe(getSubscriber("3"));
        publisher.submit("study!");

        // Publisher 종료
        publisher.close();

        // 잠시 대기하여 비동기 작업 완료 대기
        TimeUnit.SECONDS.sleep(1);
    }

    public static Flow.Subscriber<String> getSubscriber(String name) {
        return new Flow.Subscriber<>() {
            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                subscription.request(1); // 처음 아이템 하나 요청
            }

            @Override
            public void onNext(String item) {
                System.out.println(name + " Received: " + item);
                subscription.request(1); // 다음 아이템 하나 요청
            }

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

            @Override
            public void onComplete() {
                System.out.println(name + " Done");
            }
        };
    }
}

아래는 실행결과이다.

Publishing items...
1 Received: Hello
2 Received: Hello
3 Received: study!
1 Received: world!
1 Received: java
2 Received: world!
1 Received: study!
2 Received: java
2 Received: study!
3 Done
1 Done
2 Done

SubmissionPublisher 는 java.util.concurrent.Flow.Publisher의 구현체이다. subscribe()로 구독자를 등록한 후 submit() 메서드를 호출하면 구독자에게 값을 전달하는 기능을 구현하였다.

SubmissionPublisher 구현체는 Java 표준 라이브러리에서 제공되는 유일한 구현체로, 이외의 다른 구현체는 존재하지 않는다.

Flow를 간단히 사용하고자 할 때 이 구현체를 이용할 수 있다. 하지만 실제로 비동기 스트림 처리를 구현할 때는 외부 라이브러리를 사용하는 것이 일반적이라고 한다. 대표적으로 Project Reactor, RxJava, Akka Streams 등이 있다.