8.3. 수신자 목록

그림 8.3. “수신자 목록 패턴” 에 표시된 수신자 목록은 수신되는 각 메시지를 여러 다른 대상으로 전송하는 라우터 유형입니다. 또한 수신자 목록은 일반적으로 수신자 목록을 런타임에 계산해야 합니다.

그림 8.3. 수신자 목록 패턴

가장 간단한 종류의 수신자 목록은 대상 목록이 수정되고 미리 알려진 위치이며 교환 패턴은 InOnly 입니다. 이 경우 대상 목록을 to() Java DSL 명령으로 연결할 수 있습니다.

다음 예제에서는 소비자 끝점, queue:a 에서 고정 대상 목록으로 InOnly 교환을 라우팅하는 방법을 보여줍니다.

from("seda:a").to("seda:b", "seda:c", "seda:d");

다음 예제에서는 XML로 동일한 경로를 구성하는 방법을 보여줍니다.

<camelContext id="buildStaticRecipientList" xmlns="http://camel.apache.org/schema/spring">
  <route>
    <from uri="seda:a"/>
    <to uri="seda:b"/>
    <to uri="seda:c"/>
    <to uri="seda:d"/>
  </route>
</camelContext>

대부분의 경우 수신자 목록 패턴을 사용하는 경우 수신자 목록을 런타임에 계산해야 합니다. 이렇게 하려면 대상 목록을 유일한 인수로 사용하는 recipientList() 프로세서를 사용합니다. Apache Camel은 목록 인수에 유형 변환기를 적용하므로 대부분의 표준 Java 목록 유형(예: 컬렉션, 목록 또는 배열)을 사용할 수 있어야 합니다. 유형 변환기에 대한 자세한 내용은 34.3절. “붙박임 유형” 을 참조하십시오.

수신자는 동일한 교환 인스턴스의 사본을 수신하고 Apache Camel은 이를 순차적으로 실행합니다.

다음 예제에서는 recipientListHeader 라는 메시지 헤더에서 대상 목록을 추출하는 방법을 보여줍니다. 여기서 헤더 값은 끝점 URI의 쉼표로 구분된 목록입니다.

from("direct:a").recipientList(header("recipientListHeader").tokenize(","));

경우에 따라 헤더 값이 목록 유형인 경우 recipientList() 의 인수로 직접 사용할 수 있습니다. 예를 들면 다음과 같습니다.

from("seda:a").recipientList(header("recipientListHeader"));

그러나 이 예제는 기본 구성 요소가 이 특정 헤더를 구문 분석하는 방법에 전적으로 의존합니다. 구성 요소가 헤더를 간단한 문자열로 구문 분석하면 이 예제가 작동하지 않습니다. 헤더는 특정 유형의 Java 목록으로 구문 분석해야 합니다.

다음 예제에서는 XML에서 이전 경로를 구성하는 방법을 보여줍니다. 여기서 헤더 값은 엔드포인트 URI의 쉼표로 구분된 목록입니다.

<camelContext id="buildDynamicRecipientList" xmlns="http://camel.apache.org/schema/spring">
  <route>
    <from uri="seda:a"/>
    <recipientList delimiter=",">
      <header>recipientListHeader</header>
    </recipientList>
  </route>
</camelContext>

Camel 2.2에서 사용 가능

수신자 목록 패턴은 병렬Processing 을 지원하며 이는 splitter 패턴 의 해당 기능과 유사합니다. 병렬 처리 기능을 사용하여 동시에 여러 수신자에게 교환을 보냅니다. 예를 들면 다음과 같습니다.

from("direct:a").recipientList(header("myHeader")).parallelProcessing();

Spring XML에서 병렬 처리 기능은 recipientList 태그의 속성으로 구현됩니다. 예를 들면 다음과 같습니다.

<route>
  <from uri="direct:a"/>
  <recipientList parallelProcessing="true">
    <header>myHeader</header>
  </recipientList>
</route>

Camel 2.2에서 사용 가능

수신자 목록은 수신자 가 실패하는 경우 추가 수신자에게 전송을 중지하는 데 사용할 수 있는 stopOnException 기능을 지원합니다.

from("direct:a").recipientList(header("myHeader")).stopOnException();

그리고 Spring XML은 수신자 목록 태그의 특성입니다.

Spring XML에서 예외에 대한 중지 기능은 recipientList 태그의 속성으로 구현됩니다. 예를 들면 다음과 같습니다.

<route>
  <from uri="direct:a"/>
  <recipientList stopOnException="true">
    <header>myHeader</header>
  </recipientList>
</route>

Camel 2.3부터 사용 가능

수신자 목록 패턴은 무시InvalidEndpoints 옵션을 지원하므로 수신자 목록이 잘못된 엔드포인트를 건너뛸 수 있습니다(outoutslips 패턴 도 이 옵션을 지원합니다). 예를 들면 다음과 같습니다.

from("direct:a").recipientList(header("myHeader")).ignoreInvalidEndpoints();

Spring XML에서는 다음과 같이 recipientList 태그에서 ignoreInvalidEndpoints 특성을 설정하여 이 옵션을 활성화할 수 있습니다.

<route>
  <from uri="direct:a"/>
  <recipientList ignoreInvalidEndpoints="true">
    <header>myHeader</header>
  </recipientList>
</route>

myHeader 에 두 개의 끝점인 direct:foo,xxx:bar 가 포함된 경우를 고려하십시오. 첫 번째 끝점이 유효하고 작동합니다. 두 번째는 유효하지 않으므로 무시됩니다. 잘못된 엔드포인트가 발생할 때마다 INFO 수준에서 Apache Camel 로그를 기록합니다.

Camel 2.2에서 사용 가능

수신자 목록 패턴에 사용자 정의 집계Strategy 를 사용할 수 있습니다. 이 패턴은 목록 의 수신자에서 응답을 집계하는 데 유용합니다. 기본적으로 Apache Camel은 UseLatestAggregationStrategy 집계 전략을 사용하여 마지막으로 수신된 응답만 유지합니다. 보다 정교한 집계 전략의 경우 AggregationStrategy 인터페이스의 자체 구현을 정의할 수 있습니다. 자세한 내용은 8.5절. “수집기” 에서 참조하십시오. 예를 들어 응답 메시지에 사용자 정의 집계 전략 MyOwnAggregationStrategy 를 적용하려면 다음과 같이 Java DSL 경로를 정의할 수 있습니다.

from("direct:a")
    .recipientList(header("myHeader")).aggregationStrategy(new MyOwnAggregationStrategy())
    .to("direct:b");

Spring XML에서는 다음과 같이 사용자 정의 집계 전략을 recipientList 태그의 속성으로 지정할 수 있습니다.

<route>
  <from uri="direct:a"/>
  <recipientList strategyRef="myStrategy">
    <header>myHeader</header>
  </recipientList>
  <to uri="direct:b"/>
</route>

<bean id="myStrategy" class="com.mycompany.MyOwnAggregationStrategy"/>

Camel 2.2에서 사용 가능

이는 parallelProcessing 을 사용하는 경우에만 필요합니다. 기본적으로 Camel은 10개의 스레드가 있는 스레드 풀을 사용합니다. 이는 나중에 스레드 풀 관리 및 구성을 조정할 때 변경될 수 있습니다( Camel 2.2에서 제외).

사용자 정의 집계 전략과 마찬가지로 이를 구성합니다.

metrics 통합을 사용하여 수신자를 제공할 수 있습니다. 예를 들면 다음과 같습니다.

from("activemq:queue:test").recipientList().method(MessageRouter.class, "routeTo");

MessageRouter 8080은 다음과 같이 정의됩니다.

public class MessageRouter {

    public String routeTo() {
        String queueName = "activemq:queue:test2";
        return queueName;
    }
}

빈이 수신자 목록을 반환하는 메서드에 @RecipientList 주석을 추가하여 수신자 목록으로 작동하도록 할 수 있습니다. 예를 들면 다음과 같습니다.

public class MessageRouter {

    @RecipientList
    public String routeTo() {
        String queueList = "activemq:queue:test1,activemq:queue:test2";
        return queueList;
    }
}

이 경우 경로에 recipientList DSL 명령을 포함하지 마십시오. 경로를 다음과 같이 정의합니다.

from("activemq:queue:test").bean(MessageRouter.class, "routeTo");

Camel 2.5부터 사용 가능

parallelProcessing 을 사용하는 경우 총 시간 초과 값을 밀리초 단위로 구성할 수 있습니다. 그러면 Camel에서 시간 초과에 도달할 때까지 메시지를 병렬로 처리합니다. 이렇게 하면 하나의 메시지가 느린 경우 처리를 계속할 수 있습니다.

아래 예제에서 수신자 목록 헤더의 값은 direct:a,direct:b,direct:c 이므로 메시지가 3개의 수신자로 전송됩니다. 시간 초과는 250밀리초이며, 이는 마지막 두 메시지만 기간 내에 완료할 수 있음을 의미합니다. 따라서 집계는 문자열 결과인 Cryostat를 생성합니다.

from("direct:start")
    .recipientList(header("recipients"), ",")
    .aggregationStrategy(new AggregationStrategy() {
            public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
                if (oldExchange == null) {
                    return newExchange;
                }

                String body = oldExchange.getIn().getBody(String.class);
                oldExchange.getIn().setBody(body + newExchange.getIn().getBody(String.class));
                return oldExchange;
            }
        })
        .parallelProcessing().timeout(250)
    // use end to indicate end of recipientList clause
    .end()
    .to("mock:result");

from("direct:a").delay(500).to("mock:A").setBody(constant("A"));

from("direct:b").to("mock:B").setBody(constant("B"));

from("direct:c").to("mock:C").setBody(constant("C"));

기본적으로 시간 초과가 발생하면 AggregationStrategy 가 호출되지 않습니다. 그러나 전문 버전을 구현할 수 있습니다.

// Java
public interface TimeoutAwareAggregationStrategy extends AggregationStrategy {

    /**
     * A timeout occurred
     *
     * @param oldExchange  the oldest exchange (is <tt>null</tt> on first aggregation as we only have the new exchange)
     * @param index        the index
     * @param total        the total
     * @param timeout      the timeout value in millis
     */
    void timeout(Exchange oldExchange, int index, int total, long timeout);

이를 통해 실제로 필요한 경우 AggregationStrategy 에서 시간 초과를 처리할 수 있습니다.

받는 사람 끝점 중 하나에 메시지를 보내기 전에 원본 메시지의 단순 복사본인 메시지 복제본을 만듭니다.Before recipientList sends a message to one of the recipient endpoints, it creates a message replica, which is a shallow copy of the original message. 단순 복사본에서는 원본 메시지의 헤더와 페이로드가 참조로만 복사됩니다. 각 새 복사본에는 해당 요소의 자체 인스턴스가 포함되어 있지 않습니다. 결과적으로 메시지의 부분 복사본이 연결되며 다른 엔드포인트로 라우팅할 때 사용자 지정 처리를 적용할 수 없습니다.

복제본이 끝점으로 전송되기 전에 각 메시지 복제본에서 일부 사용자 지정 처리를 수행하려면 recipientList 절에서 onPrepare DSL 명령을 호출할 수 있습니다. onPrepare 명령은 메시지가 부분 복사된 직후 그리고 메시지가 끝점으로 디스패치되기 직전에 사용자 지정 프로세서를 삽입합니다. 예를 들어 다음 경로에서 각 수신자 끝점에 대한 메시지 복제본에서 CustomProc 프로세서가 호출됩니다.

from("direct:start")
  .recipientList().onPrepare(new CustomProc());

onPrepare DSL 명령의 일반적인 사용 사례는 메시지의 일부 또는 모든 요소의 깊은 복사본을 수행하는 것입니다. 이를 통해 각 메시지 복제본을 다른 사용자와 독립적으로 수정할 수 있습니다. 예를 들어 다음 CustomProc 프로세서 클래스는 메시지 본문의 깊은 복사본을 수행합니다. 여기서 메시지 본문은 유형인 BodyType.deepCopy()로 추정되고 깊은 복사는 메서드인 BodyType.deepCopy() 에 의해 수행됩니다.

// Java
import org.apache.camel.*;
...
public class CustomProc implements Processor {

    public void process(Exchange exchange) throws Exception {
        BodyType body = exchange.getIn().getBody(BodyType.class);

        // Make a _deep_ copy of of the body object
        BodyType clone =  BodyType.deepCopy();
        exchange.getIn().setBody(clone);

        // Headers and attachments have already been
        // shallow-copied. If you need deep copies,
        // add some more code here.
    }
}

recipientList DSL 명령은 다음 옵션을 지원합니다.

기본적으로 수신자 목록은 현재 교환 패턴을 사용합니다. 그러나 다른 교환 패턴을 사용하여 받는 사람에게 메시지를 보낼 수 있는 경우는 몇 가지 있을 수 있습니다.

예를 들어 InOnly 경로로 시작하는 경로가 있을 수 있습니다. 이제 받는 사람 목록과 함께 InOut 교환 패턴을 사용하려면 수신자 끝점에서 직접 교환 패턴을 구성해야 합니다.

다음 예제에서는 새 파일이 InOnly로 시작되고 수신자 목록으로 라우팅되는 경로를 보여줍니다. InOut을 ActiveMQ(JMS) 끝점과 함께 사용하려면 exchangePattern을 InOut 옵션과 동일하게 사용하여 지정해야 합니다. 그러나 응답 형식은 JMS 요청 또는 응답을 계속 라우팅하므로 응답은 outbox 디렉터리에 파일로 저장됩니다.

from("file:inbox")
  // the exchange pattern is InOnly initially when using a file route
  .recipientList().constant("activemq:queue:inbox?exchangePattern=InOut")
  .to("file:outbox");