스프링 XD : 애그리 게이터 작성 중. 비동기 적으로 출력하는 방법, 적어도 1 : 1로 출력하지 않는 방법?

Question

제목별로 수집기 변환기 모듈을 작성하는 모범 사례, 예제 등이 있습니까? 우리는 집계가 데이터의 타임 스탬프를 기반으로하며 데이터가 순서가 뒤늦거나 늦을 수도 있고 너무 늦게 고려 될 수도 있습니다.

"너무 늦은"경우에는 filter을 사용하여 트랜스포머를보기 전에 제거 할 계획입니다.

그러나 이것은 기본적인 질문으로 이어집니다. - 어 그리 게이터의 출력이 입력을 기반으로하지 않을 때 어떻게 변압기를 작성하여 집계 할 수 있습니까? 비동기 적으로 동일한 XD 스트림에 출력하도록 작성할 수 있습니까? 또는 입력의 일부 측면을 기반으로 출력에 변압기를 작성할 수 있지만 특정 입력에만 적용하고 다른 입력에는 아무 것도 출력하지 않을 수 있습니까?

구체적인 예로, 집계자가 (String, Int)의 Tuple을 취해서 Int를 기반으로 Int를 합계 한 다음 String이 특정 값 일 때만 집계를 출력하도록합니다. ; "멈추다"라고 말하십시오.

는 변압기가 작성되는 방법

(foo, 3) 
(bar, 2) 

이 작업을 수행 할 : 마지막 튜플이 변압기를 입력하면

(foo, 1) 
(bar, 2) 
(foo, 2) 
(stop, 0) 

, 나는 튜플의 흐름을 방출해야 보여 쉬운만큼?

Answer 1

transform이 아닌 aggregator 프로세서 모듈을 사용하십시오. standard aggregator module에는 없지만 Spring Integration aggregator은 discard-channel (expire-groups-on-completion이 false 인 경우)을 통해 지연 메시지를 삭제할 수 있습니다.

릴리스 전략에 따라 그룹을 릴리스 할 수 있다고 결정할 때 (또는 group-timeout을 통해) 집계자는 그룹에서 원하는 모든 작업을 수행 할 수 있으며 방출되는 모든 종류의 결과를 반환하고 다음 모듈로 이동합니다. 스트림. 따라서 일반적으로 튜플 축적을 수행하거나 기본 수집기를 사용하여 컬렉션을 릴리스하고 transform 모듈을 다운 스트림으로 사용할 수 있습니다.

기본적으로 수집기는 수동 구성 요소입니다. MessageGroupStoreReaper을 구성하거나 group-timeout을 사용하지 않는 한 그룹 릴리스는 해당 그룹에 대해 메시지가 도착할 때만 트리거됩니다 (상관 관계가 동일 함).