8.5. 수집기

그림 8.5. “수집기 패턴” 에 표시된 수집기 패턴을 사용하면 관련 메시지 배치를 단일 메시지에 결합할 수 있습니다.

그림 8.5. 수집기 패턴

수집기 동작을 제어하기 위해 Apache Camel을 사용하면 다음과 같이 엔터프라이즈 통합 패턴에 설명된 속성을 지정할 수 있습니다.

예를 들어 초당 Cryostat 메시지를 수신하는 주식 시장 데이터 시스템을 고려해 보십시오. GUI 툴이 이러한 대규모 업데이트 속도로 대처할 수 없는 경우 메시지 흐름을 제한해야 할 수 있습니다. 들어오는 주식 견적은 최신 견적을 선택하고 이전 가격을 폐기하여 간단히 집계 할 수 있습니다. (히스토리 중 일부를 캡처하려는 경우 delta 처리 알고리즘을 적용할 수 있습니다.)

그림 8.6. “수집기 구현” 집계기가 작동하는 방식에 대한 개요를 보여줍니다. A, B, C 또는 D와 같은 상관관계 키를 갖는 교환 스트림이 있다고 가정하면 됩니다.

그림 8.6. 수집기 구현

그림 8.6. “수집기 구현” 에 표시된 교환의 수신 스트림은 다음과 같이 처리됩니다.

다음 예제에서는 UseLatestAggregationStrategy 집계 전략을 사용하여 동일한 Cryostat Symbol 헤더 값으로 교환을 집계합니다. 지정된 symbol 값에 대해 해당 상관 관계 키와 마지막 교환이 수신된 이후 3초 이상 경과하면 집계된 교환은 완료된 것으로 간주되고 mock 엔드포인트로 전송됩니다.

from("direct:start")
    .aggregate(header("id"), new UseLatestAggregationStrategy())
        .completionTimeout(3000)
    .to("mock:aggregated");

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

<camelContext xmlns="http://camel.apache.org/schema/spring">
    <route>
        <from uri="direct:start"/>
        <aggregate strategyRef="aggregatorStrategy"
                   completionTimeout="3000">
            <correlationExpression>
                <simple>header.StockSymbol</simple>
            </correlationExpression>
            <to uri="mock:aggregated"/>
        </aggregate>
    </route>
</camelContext>

<bean id="aggregatorStrategy"
      class="org.apache.camel.processor.aggregate.UseLatestAggregationStrategy"/>

Java DSL에서 상관 관계 표현식은 항상 aggregate() DSL 명령에 첫 번째 인수로 전달됩니다. 여기에서 Simple Expression Language를 사용하는 것은 아닙니다. expressions 언어 또는 스크립팅 언어(예: Cryostat, XQuery, SQL 등)를 사용하여 상관 관계 표현식을 지정할 수 있습니다.

Exampe의 경우 Cryostat 표현식을 사용하여 서로 상호 작용하기 위해 다음 Java DSL 경로를 사용할 수 있습니다.

from("direct:start")
    .aggregate(xpath("/stockQuote/@symbol"), new UseLatestAggregationStrategy())
        .completionTimeout(3000)
    .to("mock:aggregated");

특정 들어오는 교환에서 상관 관계 표현식을 평가할 수 없는 경우 집계기가 기본적으로 CamelExchangeException 을 throw합니다. ignoreInvalidCorrelationKeys 옵션을 설정하여 이 예외를 비활성화할 수 있습니다. 예를 들어 Java DSL에서는 다음과 같습니다.

from(...).aggregate(...).ignoreInvalidCorrelationKeys()

XML DSL에서 ignoreInvalidCorrelationKeys 옵션은 다음과 같이 속성으로 설정할 수 있습니다.

<aggregate strategyRef="aggregatorStrategy"
           ignoreInvalidCorrelationKeys="true"
           ...>
    ...
</aggregate>

Java DSL에서는 집계 전략을 두 번째 인수로 aggregate() DSL 명령에 전달하거나 aggregationStrategy() 절을 사용하여 지정할 수 있습니다. 예를 들어 aggregationStrategy() 절을 다음과 같이 사용할 수 있습니다.

from("direct:start")
    .aggregate(header("id"))
        .aggregationStrategy(new UseLatestAggregationStrategy())
        .completionTimeout(3000)
    .to("mock:aggregated");

Apache Camel은 다음과 같은 기본 집계 전략을 제공합니다(클래스가 org.apache.camel.processor.aggregate Java 패키지에 속하는 경우).

다른 집계 전략을 적용하려면 다음 집계 전략 기본 인터페이스 중 하나를 구현할 수 있습니다.

예를 들어 다음 코드는 StringAggregationStrategy 및 Cryostat AggregationStrategy: 두 가지 다른 사용자 정의 집계 전략입니다.

 //simply combines Exchange String body values using '' as a delimiter
 class StringAggregationStrategy implements AggregationStrategy {

     public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
         if (oldExchange == null) {
             return newExchange;
         }

         String oldBody = oldExchange.getIn().getBody(String.class);
         String newBody = newExchange.getIn().getBody(String.class);
         oldExchange.getIn().setBody(oldBody + "" + newBody);
         return oldExchange;
     }
 }

 //simply combines Exchange body values into an ArrayList<Object>
 class ArrayListAggregationStrategy implements AggregationStrategy {

     public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
 	    Object newBody = newExchange.getIn().getBody();
     	ArrayList<Object> list = null;
         if (oldExchange == null) {
 		    list = new ArrayList<Object>();
 		    list.add(newBody);
 		    newExchange.getIn().setBody(list);
 		    return newExchange;
         } else {
 	        list = oldExchange.getIn().getBody(ArrayList.class);
 	    	list.add(newBody);
 		    return oldExchange;
 	    }
     }
 }

사용자 정의 전략 클래스를 사용하여 메시지를 집계하려면 Cryostat AggregationStrategy 에서는 다음과 같은 경로를 정의합니다.

from("direct:start")
    .aggregate(header("StockSymbol"), new ArrayListAggregationStrategy())
    .completionTimeout(3000)
    .to("mock:result");

다음과 같이 XML에서 사용자 정의 집계 전략을 사용하여 경로를 구성할 수도 있습니다.

<camelContext xmlns="http://camel.apache.org/schema/spring">
  <route>
    <from uri="direct:start"/>
    <aggregate strategyRef="aggregatorStrategy"
               completionTimeout="3000">
      <correlationExpression>
        <simple>header.StockSymbol</simple>
      </correlationExpression>
      <to uri="mock:aggregated"/>
    </aggregate>
  </route>
</camelContext>

<bean id="aggregatorStrategy" class="com.my_package_name.ArrayListAggregationStrategy"/>

라이프사이클이 이를 제어하는 엔터프라이즈 통합 패턴의 라이프사이클과 일치하도록 사용자 정의 집계 전략을 구현할 수 있습니다. 이는 집계 전략이 정상적으로 종료될 수 있는지 확인하는 데 유용할 수 있습니다.

라이프사이클 지원을 통해 집계 전략을 구현하려면 org.apache.camel.Service 인터페이스( AggregationStrategy 인터페이스 외에도)를 구현하고 start() 및 stop() 라이프사이클 메서드 구현을 제공해야 합니다. 예를 들어 다음 코드 예제에서는 라이프사이클 지원이 포함된 집계 전략에 대한 개요를 보여줍니다.

// Java
import org.apache.camel.processor.aggregate.AggregationStrategy;
import org.apache.camel.Service;
import java.lang.Exception;
...
class MyAggStrategyWithLifecycleControl
       implements AggregationStrategy, Service {

    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        // Implementation not shown...
        ...
    }

    public void start() throws Exception {
        // Actions to perform when the enclosing EIP starts up
        ...
    }

    public void stop() throws Exception {
        // Actions to perform when the enclosing EIP is stopping
        ...
    }
}

다음 속성은 집계된 각 교환에 대해 설정됩니다.

다음 속성은 SQL 구성 요소 집계 리포지터리에 의해 재전송되는 교환에 설정되어 있습니다( “영구 집계 리포지토리”참조).

집계 교환이 수집기를 떠나 경로의 다음 노드로 진행하는 시기를 결정하는 완료 조건을 하나 이상 지정해야 합니다. 다음 완료 조건을 지정할 수 있습니다.

이전 완료 조건은 동시에 활성화할 수 없는 completionTimeout 및 completionInterval 조건을 제외하고 임의로 결합할 수 있습니다. 조건이 함께 사용되는 경우 일반적인 규칙은 트리거하는 첫 번째 완료 조건이 효과적인 완료 조건이라는 것입니다.

집계된 교환이 완료된 시기를 결정하는 임의의 서술자 표현식을 지정할 수 있습니다. 서술자 표현식을 평가하는 방법은 다음 두 가지가 있습니다.

예를 들어 ALERT 메시지를 받을 때마다 주식 따옴표의 스트림을 종료하려는 경우(최신 들어오는 교환에서 MsgType 헤더 값으로 표시됨) 다음과 같은 경로를 정의할 수 있습니다.

from("direct:start")
    .aggregate(
      header("id"),
      new UseLatestAggregationStrategy()
    )
        .completionPredicate(
          header("MsgType").isEqualTo("ALERT")
         )
        .eagerCheckCompletion()
    .to("mock:result");

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

<camelContext xmlns="http://camel.apache.org/schema/spring">
  <route>
    <from uri="direct:start"/>
    <aggregate strategyRef="aggregatorStrategy"
               eagerCheckCompletion="true">
      <correlationExpression>
          <simple>header.StockSymbol</simple>
      </correlationExpression>
      <completionPredicate>
          <simple>$MsgType = 'ALERT'</simple>
      </completionPredicate>
      <to uri="mock:result"/>
    </aggregate>
  </route>
</camelContext>

<bean id="aggregatorStrategy"
      class="org.apache.camel.processor.aggregate.UseLatestAggregationStrategy"/>

들어오는 모든 교환에 대해 시간 초과 값이 다시 계산되는 동적 완료 시간 초과를 지정할 수 있습니다. 예를 들어 들어오는 각 교환에서 시간 초과 값을 설정하려면 다음과 같이 경로를 정의할 수 있습니다.

from("direct:start")
    .aggregate(header("StockSymbol"), new UseLatestAggregationStrategy())
        .completionTimeout(header("timeout"))
    .to("mock:aggregated");

다음과 같이 XML DSL에서 동일한 경로를 구성할 수 있습니다.

<camelContext xmlns="http://camel.apache.org/schema/spring">
    <route>
        <from uri="direct:start"/>
        <aggregate strategyRef="aggregatorStrategy">
            <correlationExpression>
                <simple>header.StockSymbol</simple>
            </correlationExpression>
            <completionTimeout>
                <header>timeout</header>
            </completionTimeout>
            <to uri="mock:aggregated"/>
        </aggregate>
    </route>
</camelContext>

<bean id="aggregatorStrategy"
      class="org.apache.camel.processor.UseLatestAggregationStrategy"/>

들어오는 모든 교환에 대해 완료 크기가 다시 계산되는 동적 완료 크기를 지정할 수 있습니다. 예를 들어 들어오는 각 교환에서 mySize 헤더에서 완료 크기를 설정하려면 다음과 같이 경로를 정의할 수 있습니다.

from("direct:start")
    .aggregate(header("StockSymbol"), new UseLatestAggregationStrategy())
        .completionSize(header("mySize"))
    .to("mock:aggregated");

Spring XML을 사용하는 것과 동일한 예:

<camelContext xmlns="http://camel.apache.org/schema/spring">
    <route>
        <from uri="direct:start"/>
        <aggregate strategyRef="aggregatorStrategy">
            <correlationExpression>
                <simple>header.StockSymbol</simple>
            </correlationExpression>
            <completionSize>
                <header>mySize</header>
            </completionSize>
            <to uri="mock:aggregated"/>
        </aggregate>
    </route>
</camelContext>

<bean id="aggregatorStrategy"
      class="org.apache.camel.processor.UseLatestAggregationStrategy"/>

사용자 정의 집계Strategy 클래스를 구현하는 경우 AggregationStrategy. aggregate() 메서드에서 반환된 교환 시 Exchange.AGGREGATION_COMPLETE_CURRENT_GROUP 교환 속성을 true 로 설정하여 현재 메시지 그룹을 강제로 완료할 수 있는 메커니즘이 있습니다. 이 메커니즘은 현재 그룹에 만 영향을 미칩니다. 다른 메시지 그룹(다른 상관관계 ID 포함)은 완료 하지 않습니다. 이 메커니즘은 서술자, 크기, 시간 초과 등과 같은 기타 완료 메커니즘을 재정의합니다.

예를 들어 다음 샘플 AggregationStrategy 클래스는 메시지 본문 크기가 5보다 큰 경우 현재 그룹을 완료합니다.

// Java
public final class MyCompletionStrategy implements AggregationStrategy {
    @Override
    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        if (oldExchange == null) {
            return newExchange;
        }
        String body = oldExchange.getIn().getBody(String.class) + "+"
            + newExchange.getIn().getBody(String.class);
        oldExchange.getIn().setBody(body);
        if (body.length() >= 5) {
            oldExchange.setProperty(Exchange.AGGREGATION_COMPLETE_CURRENT_GROUP, true);
        }
        return oldExchange;
    }
}

특수 헤더가 있는 메시지를 경로로 전송하여 모든 미해결 집계 메시지를 강제로 완료할 수 있습니다. 강제 완료에 사용할 수 있는 두 가지 대체 헤더 설정이 있습니다.

org.apache.camel.processor.aggregate.AggregateController 를 사용하면 Java 또는 Cryostat API를 사용하여 런타임 시 집계를 제어할 수 있습니다. 이는 교환 그룹을 강제 수행하거나 현재 런타임 통계를 쿼리하는 데 사용할 수 있습니다.

사용자 지정이 구성되지 않은 경우 집계기는 getAggregateController() 메서드를 사용하여 액세스할 수 있는 기본 구현을 제공합니다. 그러나 aggregateController를 사용하여 경로에서 컨트롤러를 쉽게 구성할 수 있습니다.

private AggregateController controller = new DefaultAggregateController();

from("direct:start")
   .aggregate(header("id"), new MyAggregationStrategy()).completionSize(10).id("myAggregator")
      .aggregateController(controller)
      .to("mock:aggregated");

또한 AggregateController에서 API를 사용하여 강제로 완료할 수 있습니다. 예를 들어, foo 키가 있는 그룹을 완료하려면 다음을 수행합니다.

int groups = controller.forceCompletionOfGroup("foo");

반환되는 수는 완료된 그룹 수입니다. 다음은 모든 그룹을 완료하는 API입니다.

 int groups = controller.forceCompletionOfAllGroups();

일부 집계 시나리오에서는 상관 관계 키가 교환 배치마다 고유한 조건을 적용해야 할 수 있습니다. 즉, 특정 상관 키의 집계 교환이 완료되면 해당 상관 키와의 집계 교환이 더 이상 진행되지 않도록 해야 합니다. 예를 들어 경로의 후자 부분이 고유한 상관관계 키 값으로 교환을 처리할 것으로 예상되는 경우 이 조건을 시행할 수 있습니다.

완료 조건을 구성하는 방법에 따라 특정 상관 관계를 사용하여 여러 집계 교환이 생성될 위험이 있을 수 있습니다. 예를 들어 특정 상관 키와의 모든 교환이 수신될 때까지 대기하도록 설계된 완료 서술자를 정의할 수 있지만 해당 키로의 모든 교환이 도달하기 전에 실행될 수 있는 완료 시간 초과를 정의할 수도 있습니다. 이 경우 늦은 교환으로 인해 동일한 상관 관계 키 값을 사용하여 두 번째 집계 교환이 발생할 수 있습니다.

이러한 시나리오의 경우 closeCorrelationKeyOnCompletion 옵션을 설정하여 이전 상관 키 값을 복제하는 집계 교환을 억제하도록 집계기를 구성할 수 있습니다. 중복된 상관관계 키 값을 억제하려면 집계자가 캐시에 이전 상관관계 키 값을 기록해야 합니다. 이 캐시의 크기(캐시된 상관 키 수)는 closeCorrelationKeyOnCompletion() DSL 명령에 대한 인수로 지정됩니다. 무제한 크기의 캐시를 지정하려면 0 또는 음수 정수 값을 전달할 수 있습니다. 예를 들어 캐시 크기를 10000 키 값으로 지정하려면 다음을 수행합니다.

from("direct:start")
    .aggregate(header("UniqueBatchID"), new MyConcatenateStrategy())
        .completionSize(header("mySize"))
        .closeCorrelationKeyOnCompletion(10000)
    .to("mock:aggregated");

집계 교환이 중복된 상관관계 키 값으로 완료되면 수집기에서 ClosedCorrelationKeyException 예외를 throw합니다.

스트리밍 모드에서 tokenize XML 하위 명령을 사용하여 간단한 언어 표현식을 토큰으로 사용할 수 있습니다. 간단한 언어 표현식을 사용하면 동적 토큰을 지원할 수 있습니다.

예를 들어 Java를 사용하여 태그 사람이 구분하는 일련의 이름을 분할하기 위해 tokenize XML files 및 Simple 언어 토큰을 사용하여 파일을 name 요소로 분할할 수 있습니다.

public void testTokenizeXMLPairSimple() throws Exception {
        Expression exp = TokenizeLanguage.tokenizeXML("${header.foo}", null);

< person>로 구분되는 이름의 입력 문자열을 가져오고 토큰으로 < person >를 설정합니다.

        exchange.getIn().setHeader("foo", "<person>");
        exchange.getIn().setBody("<persons><person>James</person><person>Claus</person><person>Jonathan</person><person>Hadrian</person></persons>");

입력에서 분리된 이름을 나열합니다.

        List<?> names = exp.evaluate(exchange, List.class);
        assertEquals(4, names.size());

        assertEquals("<person>James</person>", names.get(0));
        assertEquals("<person>Claus</person>", names.get(1));
        assertEquals("<person>Jonathan</person>", names.get(2));
        assertEquals("<person>Hadrian</person>", names.get(3));
    }

발신 배치에서 집계된 모든 교환을 단일 org.apache.camel.impl.GroupedExchange 홀더 클래스로 결합할 수 있습니다. 그룹화된 교환을 활성화하려면 다음 Java DSL 경로에 표시된 것처럼 group exchanges() 옵션을 지정합니다.

from("direct:start")
    .aggregate(header("StockSymbol"))
        .completionTimeout(3000)
        .groupExchanges()
    .to("mock:result");

mock:result 로 전송된 그룹화된 교환에는 메시지 본문에 집계된 교환 목록이 포함되어 있습니다. 다음 코드 줄은 후속 프로세서가 목록 형태로 그룹화된 교환의 콘텐츠에 액세스하는 방법을 보여줍니다.

// Java
List<Exchange> grouped = ex.getIn().getBody(List.class);

집계기를 배치 소비자 패턴과 함께 작동하여 배치 소비자가 보고한 총 메시지 수를 집계할 수 있습니다(배치 소비자 엔드포인트는 들어오는 교환에 CamelBatchSize,CamelBatchIndex , CamelBatchComplete 속성 설정). 예를 들어 File 소비자 엔드포인트에서 찾은 모든 파일을 집계하려면 다음과 같은 경로를 사용할 수 있습니다.

from("file://inbox")
    .aggregate(xpath("//order/@customerId"), new AggregateCustomerOrderStrategy())
    .completionFromBatchConsumer()
    .to("bean:processOrder");

현재 다음 엔드포인트는 파일, FTP, 메일, iBatis 및 JPA와 같은 배치 소비자 메커니즘을 지원합니다.

기본 집계자는 메모리 내 전용 AggregationRepository 를 사용합니다. 보류 중인 집계된 리테일링을 영구적으로 저장하려면 SQL 구성 요소를 영구 집계 리포지토리로 사용할 수 있습니다. SQL 구성 요소에는 집계된 메시지를 on-the-fly로 유지하고 메시지를 손실하지 않는 JdbcAggregationRepository 가 포함되어 있습니다.

교환이 성공적으로 처리되면 리포지토리에서 확인 메서드를 호출할 때 완료로 표시됩니다. 즉, 동일한 교환이 다시 실패하면 성공할 때까지 다시 시도됩니다.

<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-sql</artifactId>
    <version>x.x.x</version>
    <!-- use the same version as your Camel core version -->
</dependency>

CREATE TABLE my_aggregation_repo (
 id varchar(255) NOT NULL,
 exchange blob NOT NULL,
 constraint aggregation_pk PRIMARY KEY (id)
);

CREATE TABLE my_aggregation_repo_completed (
 id varchar(255) NOT NULL,
 exchange blob NOT NULL,
 constraint aggregation_completed_pk PRIMARY KEY (id)
);
}

<bean id="my_repo"
    class="org.apache.camel.processor.aggregate.jdbc.JdbcAggregationRepository">
    <property name="repositoryName" value="my_aggregation_repo"/>
    <property name="transactionManager" ref="my_tx_manager"/>
    <property name="dataSource" ref="my_data_source"/>
    ...
</bean>

repositoryName,transactionManager 및 dataSource 속성이 필요합니다. 영구 집계 리포지토리에 대한 자세한 내용은 Apache Camel 구성 요소 참조 가이드의 SQL 구성 요소를 참조하십시오.

그림 8.6. “수집기 구현” 에 표시된 대로 집계기는 경로의 후자 부분과 분리되며, 경로의 후자 부분으로 전송된 교환은 전용 스레드 풀에 의해 처리됩니다. 기본적으로 이 풀에는 단일 스레드만 포함됩니다. 여러 스레드가 있는 풀을 지정하려면 다음과 같이 parallelProcessing 옵션을 활성화합니다.

from("direct:start")
    .aggregate(header("id"), new UseLatestAggregationStrategy())
        .completionTimeout(3000)
        .parallelProcessing()
    .to("mock:aggregated");

기본적으로 10개의 작업자 스레드가 있는 풀이 생성됩니다.

생성된 스레드 풀을 더 많이 제어하려면 executorService 옵션을 사용하여 사용자 지정 java.util.concurrent.ExecutorService 인스턴스를 지정합니다(이 경우 병렬Processing 옵션을 활성화할 필요가 없음).

일반적인 집계 시나리오에서는 일련의 들어오는 메시지 본문을 List 오브젝트에 집계해야 합니다. 이 시나리오를 용이하게 하기 위해 Apache Camel은 AbstractListAggregationStrategy 추상 클래스를 제공합니다. 이 클래스를 빠르게 확장하여 이 사례에 대한 집계 전략을 생성할 수 있습니다. 유형 T 의 들어오는 메시지 본문은 List<T> 유형의 메시지 본문을 사용하여 완료된 교환으로 집계됩니다.

예를 들어 일련의 Integer 메시지 본문을 List <Integer> 오브젝트에 집계하려면 다음과 같이 정의된 집계 전략을 사용할 수 있습니다.

import org.apache.camel.processor.aggregate.AbstractListAggregationStrategy;
...
/**
 * Strategy to aggregate integers into a List<Integer>.
 */
public final class MyListOfNumbersStrategy extends AbstractListAggregationStrategy<Integer> {
 
    @Override
    public Integer getValue(Exchange exchange) {
        // the message body contains a number, so just return that as-is
        return exchange.getIn().getBody(Integer.class);
    }
}

수집기는 다음 옵션을 지원합니다.