Message Channels

The MessageChannel Interface

public interface MessageChannel {

    boolean send(Message message);

    boolean send(Message message, long timeout);
}

메시지를 보낼 때 메시지가 성공적으로 전송되면 반환 값은 true입니다. 보내기 호출이 시간 초과되거나 중단되면 false를 반환합니다.

PollableChannel

메시지 채널은 메시지를 버퍼링하거나 버퍼링하지 않을 수 있으므로 두 개의 하위 인터페이스가 버퍼링(폴링 가능) 및 비버퍼링(구독 가능) 채널 동작을 정의합니다

public interface PollableChannel extends MessageChannel {

    Message<?> receive();

    Message<?> receive(long timeout);

}

SubscribableChannel

SubscribableChannel 기본 인터페이스는 구독된 MessageHandler 인스턴스에 메시지를 직접 보내는 채널에 의해 구현됩니다. 따라서 폴링을 위한 수신 방법을 제공하지 않습니다. 대신 이러한 구독자를 관리하는 방법을 정의합니다.

public interface SubscribableChannel extends MessageChannel {

    boolean subscribe(MessageHandler handler);

    boolean unsubscribe(MessageHandler handler);

}

Message Channel Implementations

PublishSubscribeChannel

PublishSubscribeChannel 구현은 전송된 모든 메시지를 구독한 모든 핸들러로 브로드캐스트합니다.

QueueChannel

QueueChannel 구현은 queue로 래핑합니다. PublishSubscribeChannel과 달리 QueueChannel에는 point-to-point semantics가 있습니다. 즉, 채널에 여러 소비자가 있더라도 그 중 하나만 해당 채널로 전송된 메시지를 수신해야 합니다.

인수가 없는 기본 생성자(본질적으로 제한되지 않은 Integer.MAX_VALUE 용량 제공)와 다음 목록과 같이 대기열 용량을 허용하는 생성자를 제공합니다.

public QueueChannel(int capacity)

PriorityChannel

QueueChannel은 선입선출(FIFO) 순서를 적용하는 반면, PriorityChannel은 우선순위에 따라 채널 내에서 메시지를 정렬할 수 있는 대체 구현입니다

기본적으로 우선 순위는 각 메시지 내의 우선 순위 헤더에 의해 결정됩니다.

RendezvousChannel

RendezvousChannel은 다른 당사자가 채널의 receive() 메서드를 호출할 때까지 발신자가 차단하는 "직접 핸드오프" 시나리오를 가능하게 합니다

DirectChannel

DirectChannel에는 지점 간 의미 체계가 있지만 그 외에는 앞에서 설명한 대기열 기반 채널 구현보다 PublishSubscribeChannel과 더 유사합니다

PollableChannel 인터페이스 대신 SubscribableChannel 인터페이스를 구현하므로 메시지를 구독자에게 직접 발송합니다. 그러나 점대점 채널로서 각 메시지를 단일 가입 MessageHandler로 보낸다는 점에서 PublishSubscribeChannel과 다릅니다.

ExecutorChannel

ExecutorChannel은 DirectChannel과 동일한 디스패처 구성(부하 균형 전략 및 장애 조치 부울 속성)을 지원하는 지점간 채널입니다.

이 두 디스패치 채널 유형의 주요 차이점은 ExecutorChannel이 디스패치를 ​​수행하기 위해 TaskExecutor의 인스턴스에 위임한다는 것입니다.

즉, send 메서드는 일반적으로 차단하지 않지만 보낸 사람의 스레드에서 처리기 호출이 발생하지 않을 수도 있음을 의미합니다.

PartitionedChannel

버전 6.1부터 PartitionedChannel 구현이 제공됩니다. 이것은 AbstractExecutorChannel의 확장이며 이 채널로 전송된 메시지에서 평가된 파티션 키에 의해 결정되는 특정 스레드에서 실제 소비가 처리되는 지점간 디스패칭 논리를 나타냅니다.

이 채널은 위에서 언급한 ExecutorChannel과 비슷하지만 파티션 키가 같은 메시지는 항상 순서를 유지하면서 동일한 스레드에서 처리된다는 차이점이 있습니다.

외부 TaskExecutor가 필요하지 않지만 사용자 지정 ThreadFactory로 구성할 수 있습니다.

이 팩토리는 단일 스레드 실행기를 파티션별로 MessageDispatcher 대리자로 채우는 데 사용됩니다. 기본적으로 IntegrationMessageHeaderAccessor.CORRELATION_ID 메시지 헤더가 파티션 키로 사용됩니다.

@Bean
PartitionedChannel somePartitionedChannel() {
    return new PartitionedChannel(3, (message) -> message.getHeaders().get("partitionKey"));
}
// 채널에는 전용 스레드인 3개의 파티션이 있습니다. 
// 메시지를 처리할 파티션을 결정하기 위해 partitionKey 헤더를 사용합니다. 

FluxMessageChannel

FluxMessageChannel은 보낸 메시지를 내부 react.core.publisher.Flux로 "싱크"하기 위한 org.reactivestreams.Publisher 구현입니다.

Channel Interceptors

메시징 아키텍처의 장점 중 하나는 일반적인 동작을 제공하고 시스템을 통과하는 메시지에 대한 의미 있는 정보를 비침투적인 방식으로 캡처하는 기능입니다.

Message 인스턴스는 MessageChannel 인스턴스에서 송수신되므로 이러한 채널은 송수신 작업을 가로챌 수 있는 기회를 제공합니다.

public interface ChannelInterceptor {

    Message<?> preSend(Message<?> message, MessageChannel channel);

    void postSend(Message<?> message, MessageChannel channel, boolean sent);

    void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex);

    boolean preReceive(MessageChannel channel);

    Message<?> postReceive(Message<?> message, MessageChannel channel);

    void afterReceiveCompletion(Message<?> message, MessageChannel channel, Exception ex);
}

인터페이스를 구현한 후 인터셉터를 채널에 등록하는 것은 다음 호출을 수행하기만 하면 됩니다.

channel.addInterceptor(someChannelInterceptor);

MessagingTemplate

Spring Integration은 요청 및 응답 시나리오를 포함하여 메시지 채널에서 다양한 작업을 지원하는 MessagingTemplate을 제공합니다.

Example

MessagingTemplate template = new MessagingTemplate();

Message reply = template.sendAndReceive(someChannel, new GenericMessage("test"));

Signatures

public boolean send(final MessageChannel channel, final Message<?> message) { ...
}

public Message<?> sendAndReceive(final MessageChannel channel, final Message<?> request) { ...
}

public Message<?> receive(final PollableChannel<?> channel) { ...
}

Special Channels

ErrorChannel

NullChannel

Reference

Core Messaging