[java] RxJava에서의 백프레셔 제어

28 Nov 2023

java

RxJava는 비동기적인 이벤트 처리를 위한 리액티브 프로그래밍 라이브러리로, 네트워크 요청이나 데이터베이스 쿼리 등의 작업을 쉽게 처리할 수 있도록 도와줍니다. 하지만 이런 비동기 작업에서 발생하는 데이터의 양이 많아지면, 이를 처리하기 위해 시스템의 자원이 과도하게 사용될 수 있습니다. 이를 방지하기 위해 RxJava는 백프레셔라는 메커니즘을 제공합니다.

백프레셔란 소비자가 생산자로부터 받을 데이터의 양을 제어하는 메커니즘입니다. 생산자는 소비자로부터 받은 요청에 따라 데이터를 생성하고 전송하는 방식으로 동작하며, 소비자는 생산자로부터 받은 데이터를 처리한 후 추가 데이터를 요청합니다. 이런 방식으로 데이터의 흐름을 조절하면서, 소비자가 처리할 수 있는 범위를 넘어서는 데이터의 생산을 방지할 수 있습니다.

RxJava에서 백프레셔를 사용하려면, Flowable이라는 클래스를 사용해야 합니다. Flowable은 소비자와 생산자 사이의 데이터의 흐름을 효과적으로 제어하기 위해 설계된 클래스입니다.

Flowable<Integer> flowable = Flowable.range(1, 1000000)
    .onBackpressureBuffer()
    .observeOn(Schedulers.computation());

flowable.subscribe(new Subscriber<Integer>() {
    @Override
    public void onSubscribe(Subscription s) {
        s.request(10); // 소비자가 처음에 받을 데이터의 양을 지정합니다.
    }

    @Override
    public void onNext(Integer integer) {
        System.out.println(integer);
    }

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

    @Override
    public void onComplete() {
        System.out.println("Completed");
    }
});

위의 예제에서는 Flowable 클래스의 range 메서드를 통해 1부터 1000000까지의 숫자를 발행하는 생산자를 생성합니다. 그리고 onBackpressureBuffer 메서드를 호출하여 백프레셔를 설정합니다. 마지막으로 observeOn 메서드를 호출하여 데이터를 받을 스레드를 지정합니다.

Subscriber 인터페이스를 구현한 구독자에서는 onSubscribe 메서드에서 처음에 받을 데이터의 양을 지정할 수 있습니다. request 메서드를 호출하여 데이터의 양을 지정하면, 해당 양만큼의 데이터가 소비자에게 전달됩니다. 이후에는 onNext 메서드에서 데이터를 처리하고 onComplete 메서드에서 작업이 완료되었음을 알립니다.

RxJava의 백프레셔는 데이터의 흐름을 효과적으로 제어하여 메모리나 CPU 자원 등을 효율적으로 사용할 수 있도록 도와줍니다. 따라서 RxJava를 사용할 때는 백프레셔의 개념을 이해하고 적절하게 활용하는 것이 중요합니다.

참고 자료

RxJava Official Documentation