집계 필터링/Spring Integration/

Question

에 대한 필터링 후 재 순서화 방법 Spring Integration에서 프로젝트를 수행하고 있는데 큰 문제가 있습니다. 흐름에는 일부 필터링 구성 요소가 있고 나중에 흐름에는 집계 요소가 있습니다.

문제는 필터링 구성 요소가 "apply-sequence"속성을 지원하지 않는다는 것입니다. 원본 시퀀스 번호를 수정하지 않고 일부 레코드를 필터링하지만 메시지 수가 줄어 듭니다. 나중에 흐름에서 일부 메시지가 필터링되어 있기 때문에 요소를 해제하지 못하는 집계가 필요합니다.

apply-sequence 속성을 가진 특수 라우팅 요소를 사용하고 싶지 않습니다. 이러한 유형의 필터링 문제에 대한 일반적인 해결책을 제안 해 주시겠습니까?

감사합니다,

Answer 1

난 당신이 filter 및 aggregator의 행동을 오해라고 말하고 싶지만.

나는 apply-sequence - 업 그레 이드 구성 요소가 있습니다. 따라서 해당 그룹의 모든 메시지는 여러 헤더 (correlationId)를 받아 들여 의 메시지를 그룹화합니다. 기본값은aggregator입니다. sequenceNumber - 메시지의 index; sequenceSize - 그룹의 메시지 수

Filter은 메시지를 일부 조건으로 확인하고 outpu-channel 또는 discard 로직으로 보냅니다. 메시지를 수정하지 않습니다. 그러나 우리가 그렇게 할 수 있다고하더라도 어쨌든 좋은 소리는 아닙니다.

그룹에 단지 두 개의 메시지 만 있다고 가정합니다. 처음에는 필터링을 위해 OK입니다. 단지 aggregator으로 보내면됩니다. 그러나 두 번째는 버려지고, 네, 집계 원에게 보내지 않을 것입니다. 그리고 마지막 그룹은 sequenceSize에 도달하지 않았기 때문에 해당 그룹을 공개하지 않습니다.

귀하의 요구 사항을 극복하기 위해서는 aggregator에 ReleaseStrategy을 맞춤화해야합니다 (기본값은 SequenceSizeReleaseStrategy). 예를 들어 그룹의 모든 메시지가 이후 true 또는 false 결과와 독립적으로 전송 된 시스템의 상태를 확인합니다. 또는 같은 이유로 인해 fake 메시지가 있고 해당 그룹에서 사용 가능 여부를 확인하십시오.

이 경우 수집기에서 메시지를 그룹화하려면 correlationId에주의해야합니다.

UPDATE

이러한 시나리오에 대한 제안 출시 전략은 무엇입니까? 릴리스 전략으로 타임 아웃을 사용하는 것이 좋은 전략입니까?

가끔은 일부 통합 시나리오에 적합한 솔루션을 찾는 것이 어렵다고 말할 수 있습니다. 메시징은 성격 상 stateless이므로, 결정되지 않은 수의 메시지를 상관시키고 그룹화하는 것은 문제가 될 수 있습니다.

요구 사항 및 환경을 확인해야합니다.

예를 들어 모든 메시지가 단일 스레드에서 처리 될 때 fake 마커 메시지를 aggregator으로 직접 보내고 ReleaseStrategy에서 확인할 수 있습니다. 그룹의 모든 메시지가 삭제 될 때도 작동합니다.

이러한 메시지를 병렬로 처리하거나 다른 스레드에서 수신 한 경우 메시지 순서와 각 프로세스의 시간을 결정할 수 없습니다.

이 경우 TimeoutCountSequenceSizeReleaseStrategy이 실제로 도움이 될 수 있습니다. 물론 시스템 요구 사항에 따라 적절한 시간대를 찾아야 할 필요가 있습니다.