230.7. 소비자

커서를 tailable 커서로 전환하려면 먼저 커서를 생성할 때 몇 가지 특수 플래그를 MongoDB로 전달해야 합니다. 생성되면 커서가 열려 있고 새 데이터가 도착할 때까지 MongoCursor.next() 메서드를 호출할 때 차단됩니다. 그러나 새 데이터가 결정되지 않은 기간 후에 새 데이터가 나타나지 않으면 MongoDB 서버는 자동으로 커서를 종료할 수 있는 권리를 보유합니다. 새 데이터를 계속 사용하려면 커서를 다시 생성해야 합니다. 그리고 그렇게하기 위해, 당신은 당신이 떠났거나 다른 곳에서 다시 사용하기 시작할 위치를 기억해야합니다.

Camel MongoDB tailable 커서 소비자는 이러한 모든 작업을 처리합니다. 증가된 성격의 데이터에 있는 일부 필드에 키를 제공해야 합니다. 이 키는 다시 생성될 때마다 커서를 배치하기 위한 마커 역할을 합니다(예: 타임스탬프, 순차적 ID 등). MongoDB에서 지원하는 모든 데이터 유형일 수 있습니다. date, Strings 및 Integers가 제대로 작동하는 것을 확인할 수 있습니다. 이 구성 요소의 컨텍스트에서 이 메커니즘을 "보통 추적"이라고 합니다.

소비자는 이 필드의 마지막 값을 기억하고 커서가 다시 생성될 때마다, 읽기되지 않은 데이터만 사용되도록: increasingField > lastValue 와 같은 필터를 사용하여 쿼리를 실행합니다.

increasing 필드 설정: 끝점 URI tailtrackingIncreasingField 옵션에 대해 증가 필드의 키를 설정합니다. Camel 2.10에서는 이 필드에 대한 중첩된 탐색은 아직 지원되지 않으므로 데이터의 최상위 필드여야 합니다. 즉, "timestamp" 필드가 괜찮지만 "nested.timestamp"는 작동하지 않습니다. 중첩 필드 지원이 필요한 경우 Camel JIRA에서 티켓을 엽니다.

커서 재생 지연: 한 가지 사항에 유의해야 할 때 새 데이터를 아직 사용할 수 없는 경우 MongoDB는 커서를 즉시 종료한다는 것입니다. 이 경우 서버를 덮어 쓰려고 하지 않기 때문에 커서RegenerationDelay 옵션이 도입되었습니다(기본값 1000ms 사용). 필요에 맞게 수정할 수 있습니다.

예를 들면 다음과 같습니다.

from("mongodb3:myDb?database=flights&collection=cancellations&tailTrackIncreasingField=departureTime")
    .id("tailableCursorConsumer1")
    .autoStartup(false)
    .to("mock:test");

위의 경로는 "flights.cancellations" capped 컬렉션에서 사용하며, "departureTime"을 increasing 필드로 사용하며 기본 재생 커서 지연은 1000ms입니다.

표준 tail 추적은 휘발성이며 마지막 값은 메모리에만 유지됩니다. 그러나 실제로 Camel 컨테이너를 이제 다시 시작해야 하지만 마지막 값이 손실되고, 비상 가능한 커서 소비자가 다시 시작되므로 중복 레코드를 경로로 보낼 가능성이 높습니다.

이러한 상황을 극복하기 위해 영구 꼬리 추적 기능을 활성화하여 MongoDB 데이터베이스 내부의 특수 수집에서 마지막으로 소비되는 값을 추적할 수 있습니다. 소비자가 다시 초기화하면 마지막으로 추적된 값을 복원하고 아무 일도 일어나지 않은 것처럼 계속됩니다.

마지막 읽기 값은 두 번 유지 됩니다 : 커서가 다시 생성될 때마다 그리고 소비자가 종료될 때. 요구가 있는 경우 견고성을 추가하기 위해 앞으로도 5초마다 일정한 간격으로 지속하는 것을 고려할 수 있습니다. 이 기능을 요청하려면 Camel JIRA에서 티켓을 열어야 합니다.

이 기능을 활성화하려면 끝점 URI에 다음 옵션을 설정합니다.

또한 tailtrack Db , tail trackCollection 및 tail trackField 옵션을 설정하여 런타임 정보를 저장할 위치를 사용자 지정할 수 있습니다. 각 옵션에 대한 설명은 이 페이지 상단에 있는 끝점 옵션 표를 참조하십시오.

예를 들어 다음 경로는 증가 필드로 "departureTime"을 사용하여 1000ms의 기본 regeneration 커서 지연을 사용하여 1000ms의 "flights.cancellllations"에서 소비하며, "flights.camelTailing"의 "cancell boundaryer" ID를 유지합니다. 마지막 처리된 값을 "lasttrackingValue" 필드 아래에 저장합니다 (camelTail tracking 및 lastTrackingValue 는 기본값).

from("mongodb3:myDb?database=flights&collection=cancellations&tailTrackIncreasingField=departureTime&persistentTailTracking=true" +
     "&persistentId=cancellationsTracker")
    .id("tailableCursorConsumer2")
    .autoStartup(false)
    .to("mock:test");

다음은 위의 것과 동일한 또 다른 예이지만, 영구 꼬리 추적 런타임 정보는 "lastProcessedDepartureTime" 필드의 "trackers.camelTrackers" 컬렉션에 저장됩니다.

from("mongodb3:myDb?database=flights&collection=cancellations&tailTrackIncreasingField=departureTime&persistentTailTracking=true" +
     "&persistentId=cancellationsTracker&tailTrackDb=trackers&tailTrackCollection=camelTrackers" +
     "&tailTrackField=lastProcessedDepartureTime")
    .id("tailableCursorConsumer3")
    .autoStartup(false)
    .to("mock:test");