Java 9 Subscriber 인터페이스란?
Subscriber(구독자) 인터페이스는 Publisher(발행자)를 구독하여 데이터를 수신하는 역할을 담당합니다. 구독자는 onNext() 메서드를 통해 발행되는 아이템을 받고, 오류가 발생하면 onError() 메서드로 에러 메시지를 전달받으며, 더 이상 받을 아이템이 없다는 신호는 onComplete() 메서드를 통해 확인할 수 있습니다.
이러한 신호들이 시작되기 전에, Publisher는 반드시 onSubscribe() 메서드를 먼저 호출해 구독 관계를 설정합니다.
public interface Subscriber<T> {
public void onSubscribe(Subscription s);
public void onNext(T t);
public void onError(Throwable t);
public void onComplete();
}Subscriber 인터페이스의 필수 규칙
- 요청 기반 데이터 수신: Subscriber는 Subscription.request(long n) 메서드를 호출해야만 onNext() 신호를 받을 수 있습니다.
- 종료 후 상호작용 금지: Subscriber.onComplete()와 Subscriber.onError(Throwable t) 메서드 내에서는 Subscription 또는 Publisher의 어떤 메서드도 호출해서는 안 됩니다.
- 구독 취소 간주: 위 두 메서드가 신호를 수신한 이후에는 해당 구독이 이미 취소된 것으로 간주해야 합니다.
- 중복 구독 방지: Subscriber가 이미 활성화된 구독(Subscription)을 보유하고 있다면, 새로운 onSubscribe() 신호를 받았을 때 기존 구독에 대해 Subscription.cancel()을 호출해야 합니다.
- 불필요한 구독 해제: 더 이상 구독이 필요하지 않게 되면 반드시 Subscription.cancel() 메서드를 호출해야 합니다.
- 직렬 호출 보장: Subscriber는 자신의 구독에 대한 request()와 cancel() 메서드 호출이 직렬(순차)적으로 수행되도록 보장해야 합니다.
- cancel() 이후의 신호: 아직 요청된 요소가 남아 있는 상태에서 Subscription.cancel()을 호출하더라도, 하나 이상의 onNext() 신호를 추가로 받을 수 있습니다. cancel()은 내부 정리(cleanup) 작업이 즉시 실행됨을 보장하지 않습니다.
- onComplete() 수신 조건: Subscriber는 Subscription.request(long n) 호출 여부와 관계없이 onComplete() 신호를 받을 수 있어야 합니다.
- onError() 수신 조건: 마찬가지로 request 호출 여부와 무관하게 onError() 신호를 받을 수 있어야 합니다.
- 시그널 순서 보장: 모든 시그널 메서드의 호출은 해당 시그널의 처리보다 먼저 발생해야 합니다(happens-before 관계). 즉, Subscriber는 수신한 시그널을 자체 처리 로직에 올바르게 전달할 책임이 있습니다.
- onSubscribe() 단일 호출: Subscriber.onSubscribe()는 동일한 Subscriber에 대해 최대 한 번만 호출되어야 합니다.
- null 파라미터 처리: onSubscribe(), onNext(), onError(), onComplete() 호출은 정상적으로 반환되어야 하며, 단 전달된 파라미터가 null인 경우에는 호출자에게 NullPointerException을 던져야 합니다.
마무리
이러한 규칙들은 Java 9에 도입된 리액티브 스트림즈(Reactive Streams) API의 핵심 계약입니다. Subscriber가 이 규칙들을 준수하면 배압(backpressure) 처리, 비동기 데이터 흐름 제어, 그리고 안정적인 스트림 종료가 가능해집니다. 특히 request()를 통한 명시적 데이터 요청과 cancel()을 통한 적절한 구독 해제는 메모리 누수와 과부하 문제를 예방하는 중요한 장치이므로, 반응형 애플리케이션을 개발할 때 반드시 숙지해야 합니다.