Repository navigation
Java_Pub Sub_Api
dev edited this page Jul 28, 2025
·
2 revisions
서로 연관된 인터페이스들과 정적 메서드들은 Publisher가 생성한 항목을 하나 이상의 Subscriber가 소비하도록 설계된 흐름 제어형 구성 요소(flow-controlled components) 를 정의합니다.
각 Subscriber는 Subscription을 통해 관리됩니다.
이 인터페이스들은 Reactive Streams 명세에 해당하며, 동시성 및 분산 비동기 환경 모두에서 사용될 수 있습니다.
모든 7개의 메서드는 void 반환형을 가지며, 일방향 메시지(one-way message) 스타일로 동작합니다.
통신은 간단한 형태의 흐름 제어(flow control)에 의존하는데, 이는 Subscription#request 메서드를 통해 이루어지며,
"푸시 기반 시스템(push-based systems)"에서 발생할 수 있는 리소스 관리 문제를 방지할 수 있습니다.
일반적으로 Publisher는 자신만의 Subscription 구현체를 정의합니다.
이 구현체는 subscribe 메서드에서 생성되어, 호출한 Subscriber에게 전달됩니다.
항목은 일반적으로 비동기적으로 발행되며, 이를 위해 Executor를 사용하는 경우가 많습니다.
다음은 매우 단순한 Publisher의 예시입니다:
- 요청이 있을 때에만
- 단 하나의
TRUE항목을 - 단 하나의 구독자에게 발행(publish)합니다.
이 예제에서는 구독자가 오직 하나의 항목만 받기 때문에,
버퍼링이나 순서 제어가 필요 없습니다.
(복잡한 구현에서는 SubmissionPublisher와 같은 클래스가 이러한 기능을 제공합니다.)
class OneShotPublisher implements Publisher<Boolean> {
private final ExecutorService executor = ForkJoinPool.commonPool(); // daemon-based
private boolean subscribed; // true after first subscribe
public synchronized void subscribe(Subscriber<? super Boolean> subscriber) {
if (subscribed)
subscriber.onError(new IllegalStateException()); // only one allowed
else {
subscribed = true;
subscriber.onSubscribe(new OneShotSubscription(subscriber, executor));
}
}
static class OneShotSubscription implements Subscription {
private final Subscriber<? super Boolean> subscriber;
private final ExecutorService executor;
private Future<?> future; // to allow cancellation
private boolean completed;
OneShotSubscription(Subscriber<? super Boolean> subscriber,
ExecutorService executor) {
this.subscriber = subscriber;
this.executor = executor;
}
public synchronized void request(long n) {
if (!completed) {
completed = true;
if (n <= 0) {
IllegalArgumentException ex = new IllegalArgumentException();
executor.execute(() -> subscriber.onError(ex));
} else {
future = executor.submit(() -> {
subscriber.onNext(Boolean.TRUE);
subscriber.onComplete();
});
}
}
}
public synchronized void cancel() {
completed = true;
if (future != null) future.cancel(false);
}
}
}}Subscriber는 항목(item)을 요청하고 처리하는 역할을 담당합니다.
Subscriber#onNext 메서드를 통해 항목이 전달되며, 요청하지 않으면 항목은 전달되지 않습니다.
하지만 한 번에 여러 항목을 요청할 수 있습니다.
많은 Subscriber 구현체는 다음 예시 스타일처럼 동작할 수 있습니다:
- 버퍼 크기(buffer size)를 1로 설정하면 항목을 하나씩 단계적으로 처리(single-step)하게 됩니다.
- 더 큰 버퍼 크기는 처리와 통신을 병렬적으로 효율적으로 수행할 수 있게 도와줍니다.
예를 들어 버퍼 크기를 64로 설정하면, 총 요청 중 처리 대기 중인 항목 수는 32~64개 사이를 유지하게 됩니다.
특정 Subscription에 대한 Subscriber의 메서드 호출 순서는 엄격하게 보장되므로,
동기화(synchronized), 락(lock), volatile 변수 등을 사용할 필요가 없습니다.
단, Subscriber가 여러 개의 Subscription을 동시에 유지한다면,
이보다는 각각의 Subscription에 대해 별도의 Subscriber를 정의하는 것이 더 좋습니다.
class SampleSubscriber<T> implements Subscriber<T> {
final Consumer<? super T> consumer;
Subscription subscription;
final long bufferSize;
long count;
SampleSubscriber(long bufferSize, Consumer<? super T> consumer) {
this.bufferSize = bufferSize;
this.consumer = consumer;
}
public void onSubscribe(Subscription subscription) {
long initialRequestSize = bufferSize;
count = bufferSize - bufferSize / 2; // re-request when half consumed
(this.subscription = subscription).request(initialRequestSize);
}
public void onNext(T item) {
if (--count <= 0)
subscription.request(count = bufferSize - bufferSize / 2);
consumer.accept(item);
}
public void onError(Throwable ex) { ex.printStackTrace(); }
public void onComplete() {}
}
}defaultBufferSize의 기본값은 다음과 같은 경우에 유용한 시작점이 될 수 있습니다:
- 예상 처리 속도(rate)
- 사용 가능한 리소스(resources)
- 사용 형태(usages)
이 값은 Flow 구성 요소에서 요청 크기(request size) 및 버퍼 용량(capacity) 을 설정할 때 기준이 됩니다.
흐름 제어(flow control)가 전혀 필요하지 않은 경우,
Subscriber는 처음부터 사실상 무제한 개수의 항목을 요청할 수도 있습니다.
예를 들어 다음과 같이 사용할 수 있습니다:
class UnboundedSubscriber<T> implements Subscriber<T> {
public void onSubscribe(Subscription subscription) {
subscription.request(Long.MAX_VALUE); // effectively unbounded
}
public void onNext(T item) { use(item); }
public void onError(Throwable ex) { ex.printStackTrace(); }
public void onComplete() {}
void use(T item) { ... }
public final class Flow {
private Flow() {} // uninstantiable
}
/**
* A producer of items (and related control messages) received by
* Subscribers. Each current {@link Subscriber} receives the same
* items (via method {@code onNext}) in the same order, unless
* drops or errors are encountered. If a Publisher encounters an
* error that does not allow items to be issued to a Subscriber,
* that Subscriber receives {@code onError}, and then receives no
* further messages. Otherwise, when it is known that no further
* messages will be issued to it, a subscriber receives {@code
* onComplete}. Publishers ensure that Subscriber method
* invocations for each subscription are strictly ordered in <a
* href="package-summary.html#MemoryVisibility"><i>happens-before</i></a>
* order.
*
* <p>Publishers may vary in policy about whether drops (failures
* to issue an item because of resource limitations) are treated
* as unrecoverable errors. Publishers may also vary about
* whether Subscribers receive items that were produced or
* available before they subscribed.
*
* @param <T> the published item type
*/
@FunctionalInterface
public static interface Publisher<T> {
/**
* Adds the given Subscriber if possible. If already
* subscribed, or the attempt to subscribe fails due to policy
* violations or errors, the Subscriber's {@code onError}
* method is invoked with an {@link IllegalStateException}.
* Otherwise, the Subscriber's {@code onSubscribe} method is
* invoked with a new {@link Subscription}. Subscribers may
* enable receiving items by invoking the {@code request}
* method of this Subscription, and may unsubscribe by
* invoking its {@code cancel} method.
*
* @param subscriber the subscriber
* @throws NullPointerException if subscriber is null
*/
public void subscribe(Subscriber<? super T> subscriber);
}
/**
* A receiver of messages. The methods in this interface are
* invoked in strict sequential order for each {@link
* Subscription}.
*
* @param <T> the subscribed item type
*/
public static interface Subscriber<T> {
/**
* Method invoked prior to invoking any other Subscriber
* methods for the given Subscription. If this method throws
* an exception, resulting behavior is not guaranteed, but may
* cause the Subscription not to be established or to be cancelled.
*
* <p>Typically, implementations of this method invoke {@code
* subscription.request} to enable receiving items.
*
* @param subscription a new subscription
*/
public void onSubscribe(Subscription subscription);
/**
* Method invoked with a Subscription's next item. If this
* method throws an exception, resulting behavior is not
* guaranteed, but may cause the Subscription to be cancelled.
*
* @param item the item
*/
public void onNext(T item);
/**
* Method invoked upon an unrecoverable error encountered by a
* Publisher or Subscription, after which no other Subscriber
* methods are invoked by the Subscription. If this method
* itself throws an exception, resulting behavior is
* undefined.
*
* @param throwable the exception
*/
public void onError(Throwable throwable);
/**
* Method invoked when it is known that no additional
* Subscriber method invocations will occur for a Subscription
* that is not already terminated by error, after which no
* other Subscriber methods are invoked by the Subscription.
* If this method throws an exception, resulting behavior is
* undefined.
*/
public void onComplete();
}
/**
* Message control linking a {@link Publisher} and {@link
* Subscriber}. Subscribers receive items only when requested,
* and may cancel at any time. The methods in this interface are
* intended to be invoked only by their Subscribers; usages in
* other contexts have undefined effects.
*/
public static interface Subscription {
/**
* Adds the given number {@code n} of items to the current
* unfulfilled demand for this subscription. If {@code n} is
* less than or equal to zero, the Subscriber will receive an
* {@code onError} signal with an {@link
* IllegalArgumentException} argument. Otherwise, the
* Subscriber will receive up to {@code n} additional {@code
* onNext} invocations (or fewer if terminated).
*
* @param n the increment of demand; a value of {@code
* Long.MAX_VALUE} may be considered as effectively unbounded
*/
public void request(long n);
/**
* Causes the Subscriber to (eventually) stop receiving
* messages. Implementation is best-effort -- additional
* messages may be received after invoking this method.
* A cancelled subscription need not ever receive an
* {@code onComplete} or {@code onError} signal.
*/
public void cancel();
}
/**
* A component that acts as both a Subscriber and Publisher.
*
* @param <T> the subscribed item type
* @param <R> the published item type
*/
public static interface Processor<T,R> extends Subscriber<T>, Publisher<R> {
}
static final int DEFAULT_BUFFER_SIZE = 256;
/**
* Returns a default value for Publisher or Subscriber buffering,
* that may be used in the absence of other constraints.
*
* @implNote
* The current value returned is 256.
*
* @return the buffer size value
*/
public static int defaultBufferSize() {
return DEFAULT_BUFFER_SIZE;
}
}