MongoDB 컴포넌트

원문 보기 위키 갱신 2026.10.11

MongoDB 컴포넌트 (camel-mongodb)

camel-mongodb 컴포넌트는 Camel 라우트에서 MongoDB 컬렉션과 상호작용할 수 있게 해 주는 컴포넌트예요. Camel 2.19부터 제공됐고 producer와 consumer를 모두 지원해요. findById·findAll·insert·save·update·remove·bulkWrite·aggregate 같은 연산과, tailable 커서·Change Streams 두 종류의 consumer, tail tracking 옵션들을 제공해요.

출처: 문서

본문

Camel 2.19부터 지원

producer와 consumer를 모두 지원해요

위키피디아에 따르면 "NoSQL은 오랜 관계형 데이터베이스와 ACID 보장의 역사에서 벗어나려는, 느슨하게 정의된 비관계형 데이터 스토어 부류를 지지하는 흐름이에요." NoSQL 솔루션은 지난 몇 년간 인기를 얻었고, Facebook, LinkedIn, Twitter 같이 극단적으로 많이 쓰이는 사이트와 서비스들이 확장성과 민첩성을 얻으려고 NoSQL을 폭넓게 활용하는 것으로 알려져 있어요.

기본적으로 NoSQL 솔루션은 전통적인 RDBMS(Relational Database Management System)와 다른데, SQL을 쿼리 언어로 쓰지 않고 일반적으로 ACID 같은 트랜잭션 동작이나 관계형 데이터를 제공하지 않아요. 대신 유연한 데이터 구조와 스키마(고정 스키마를 가진 전통적인 데이터베이스 테이블 개념은 버린다는 뜻이에요), 범용 하드웨어에서의 극한 확장성, 매우 빠른 처리를 중심으로 설계됐어요.

MongoDB는 아주 인기 있는 NoSQL 솔루션이에요. camel-mongodb 컴포넌트는 Camel과 MongoDB를 통합해서, producer(컬렉션에 연산 수행)와 consumer(MongoDB 컬렉션에서 문서 소비) 양쪽으로 MongoDB 컬렉션과 상호작용할 수 있게 해 줘요.

MongoDB는 documents(사무용 문서가 아니라 JSON/BSON으로 정의한 계층형 데이터예요)와 collections 개념을 중심으로 돌아가요. 이 컴포넌트 페이지는 여러분이 이 개념에 익숙하다고 가정해요. 그렇지 않다면 http://www.mongodb.org/를 방문하세요.

참고 MongoDB Camel 컴포넌트는 Mongo Java Driver 4.x를 사용해요.

Maven 사용자라면 이 컴포넌트를 위해 pom.xml에 다음 의존성을 추가해야 해요.

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

URI 형식 (URI Format)

mongodb:connectionBean?database=databaseName&collection=collectionName&operation=operationName[&moreOptions...] mongodb:dummy?hosts=hostnames&database=databaseName&collection=collectionName&operation=operationName[&moreOptions...]

옵션 설정 방법 (Configuring Options)

Camel 컴포넌트는 서로 다른 두 단계로 설정해요:

컴포넌트 레벨
엔드포인트 레벨

컴포넌트 옵션 설정 (Configuring Component Options)

컴포넌트 레벨에서는 엔드포인트들이 이어서 물려받는 일반적이고 공유되는 설정을 해요. 가장 상위의 설정 레벨이죠.

예를 들어 컴포넌트에는 보안 설정, 인증용 자격 증명, 네트워크 연결 URL 같은 것들이 있을 수 있어요.

컴포넌트에 따라 옵션이 몇 개뿐인 것도 있고 아주 많은 것도 있어요. 컴포넌트는 보통 널리 쓰이는 기본값이 미리 설정되어 있어서, 몇 가지 옵션만 설정하거나 아예 하나도 건드리지 않아도 되는 경우가 많아요.

컴포넌트는 다음 방법으로 설정할 수 있어요:

the Component DSL.
in a configuration file (application.properties, *.yaml files, etc).
directly in the Java code.

엔드포인트 옵션 설정 (Configuring Endpoint Options)

엔드포인트는 옵션이 많아서 설정에 시간을 더 쓰게 될 거예요. 이 옵션들이 엔드포인트가 원하는 일을 하도록 커스터마이즈해 줘요. 옵션은 엔드포인트가 컨슈머(from)로 쓰이는지, 프로듀서(to)로 쓰이는지, 둘 다인지에 따라 분류되기도 해요.

엔드포인트 설정은 대부분 엔드포인트 URI에 _path_와 query 파라미터로 바로 하는 경우가 많아요. Java에서 엔드포인트와 데이터 포맷을 _타입 세이프_하게 설정하려면 Endpoint DSL과 DataFormat DSL을 쓸 수도 있어요.

옵션을 설정할 때 좋은 습관은 Property Placeholders를 사용하는 거예요.

프로퍼티 플레이스홀더는 이런 장점이 있어요:

URL, 포트 번호, 민감한 정보 같은 설정을 하드코딩하는 걸 막아 줘요.
설정을 코드 밖으로 분리할 수 있어요.
코드가 더 유연해지고 재사용하기 좋아져요.

다음 두 섹션에서 모든 옵션을 나열해요. 먼저 컴포넌트 옵션, 그다음 엔드포인트 옵션 순서예요.

컴포넌트 옵션 (Component Options)

MongoDB 컴포넌트가 지원하는 옵션은 아래와 같아요.

Name	Description	Default	Type
mongoConnection (common)	Autowired 연결에 사용하는 공유 클라이언트예요. 이 컴포넌트에서 생성된 모든 엔드포인트가 이 연결 클라이언트를 공유해요.		MongoClient
bridgeErrorHandler (consumer)	컨슈머를 Camel 라우팅 에러 핸들러(Error Handler)에 브리지할 수 있게 해 줘요. 즉, Camel 컨슈머가 들어오는 메시지를 집어 올리는 동안 등에서 예외가 발생하면(가능한 경우), 그 예외를 메시지로 처리해서 라우팅 에러 핸들러가 다루게 돼요. 중요: 이것은 서드파티 컴포넌트가 예외가 던져졌을 때 Camel에 알릴 수 있도록 허용하는 경우에만 가능해요. 일부 컴포넌트는 내부에서만 처리하기 때문에 bridgeErrorHandler를 쓸 수 없어요. 다른 상황에서는 Camel 컴포넌트를 개선해서 서드파티 컴포넌트에 연결(hook)해 이를 가능하게 만들 수도 있어요. 기본으로 컨슈머는 org.apache.camel.spi.ExceptionHandler로 예외를 처리하는데, WARN 또는 ERROR 레벨로 로그를 남기고 무시해요.	false	boolean
lazyStartProducer (producer)	producer를 lazy(첫 메시지 시점)로 시작할지 여부예요. lazy 시작을 쓰면 producer가 시작 중 실패해서 라우트 시작이 실패하는 상황에서도 CamelContext와 라우트를 시작할 수 있고, 시작 실패 처리를 Camel의 라우팅 에러 핸들러로 넘길 수 있어요. 다만 첫 메시지를 처리하는 시점에 producer를 만들고 시작하는 데 시간이 조금 걸려서 전체 처리 시간이 길어질 수 있다는 점은 유의하세요.	false	boolean
autowiredEnabled (advanced)	오토와이어링(autowiring)을 활성화할지 여부예요. 레지스트리에서 일치하는 타입의 인스턴스가 딱 하나 있는지 찾아서, 그 인스턴스로 오토와이어링 옵션(옵션이 autowired로 표시되어 있어야 함)을 자동 설정하는 데 쓰여요. JDBC 데이터 소스, JMS 커넥션 팩토리, AWS 클라이언트 등을 자동 설정할 때 사용할 수 있어요.	true	boolean
useGlobalSslContextParameters (security)	전역 SSL 컨텍스트 파라미터 사용을 활성화해요.	false	boolean

엔드포인트 옵션 (Endpoint Options)

MongoDB 엔드포인트는 다음 URI 문법으로 설정해요:

mongodb:connectionBean

다음 path 및 query 파라미터를 사용해요:

경로 파라미터 (Path Parameters)

Name	Description	Default	Type
connectionBean (common)	필수 hosts 파라미터가 없을 때 데이터베이스 연결용 클라이언트를 찾는 데 사용할 connection bean 참조를 설정해요.		String

쿼리 파라미터 (Query Parameters)

Name	Description	Default	Type
collection (common)	이 엔드포인트에 바인딩할 MongoDB 컬렉션 이름을 설정해요.		String
collectionIndex (common)	컬렉션 인덱스를 설정해요 (JSON FORMAT : \{ field1 : order1, field2 : order2}).		String
connectionUriString (common)	mongodb 엔드포인트의 전체 Connection String/Uri를 설정해요.		String
createCollection (common)	초기화 시점에 컬렉션이 없으면 만들어요. 기본값은 true예요.	true	boolean
database (common)	대상 MongoDB 데이터베이스 이름을 설정해요.		String
hosts (common)	host:port 형식의 mongodb 서버 호스트 주소예요. 주소를 여러 개 쓸 수도 있는데, host1:port1,host2:port2처럼 콤마로 구분된 호스트 목록을 쓰면 돼요. 이 파라미터를 지정하면 제공된 connectionBean은 무시돼요.		String
mongoConnection (common)	데이터베이스 연결용 클라이언트로 쓸 connection bean을 설정해요.		MongoClient
operation (common)
이 엔드포인트가 MongoDB에 대해 실행할 연산을 설정해요.

Enum values:

findById
findOneAndDelete
findOneAndReplace
findOneAndUpdate
findOneByQuery
findAll
findDistinct
insert
save
update
remove
bulkWrite
aggregate
getDbStats
getColStats
count
command

producer의 출력을 선택한 타입으로 변환해요: DocumentList, Document 또는 MongoIterable. DocumentList와 MongoIterable은 findAll과 aggregate에 적용되고, Document는 그 외 모든 연산에 적용돼요.

Enum values:

DocumentList
Document
MongoIterable

컨슈머 타입이에요.

Enum values:

tailable
changeStreams

update 연산으로 문서가 수정됐을 때 changeStream 컨슈머가 전체 문서의 복사본을 포함할지 지정해요. 가능한 값은 default, updateLookup, required, whenAvailable이에요.

Enum values:

default
updateLookup
required
whenAvailable

| default | FullDocument | | persistentId (consumer) | tail tracking 컬렉션 하나가 여러 tailable 컨슈머의 여러 트래커를 수용할 수 있어요. 서로 분리하려면 각 트래커가 고유한 persistentId를 가져야 해요. | | String | | persistentTailTracking (consumer) | persistent tail tracking을 활성화해요. 시스템 재시작을 넘어 마지막으로 소비한 메시지를 추적하는 메커니즘이에요. 시스템이 다시 올라오면 엔드포인트가 마지막으로 레코드를 읽던 지점부터 커서를 복원해요. | false | boolean | | tailTrackCollection (consumer) | tail tracking 정보가 영속화될 컬렉션이에요. 지정하지 않으면 MongoDbTailTrackingConfig#DEFAULT_COLLECTION이 기본으로 사용돼요. | | String | | tailTrackDb (consumer) | tail tracking 메커니즘이 영속화될 데이터베이스를 지시해요. 지정하지 않으면 기본으로 현재 데이터베이스가 선택돼요. dynamicity가 켜져 있어도 고려되지 않아요. 즉, tail tracking 데이터베이스는 엔드포인트 초기화 이후로는 바뀌지 않아요. | | String | | tailTrackField (consumer) | 마지막 추적 값이 담길 필드예요. 지정하지 않으면 MongoDbTailTrackingConfig#DEFAULT_FIELD가 기본으로 사용돼요. | | String | | tailTrackIncreasingField (consumer) | 들어오는 레코드에서 증가하는 성격의 상관 필드예요. tailable 커서를 생성할 때마다 위치를 잡는 데 사용돼요. 커서는 tailTrackIncreasingField greater than lastValue(영구 tail tracking에서 복원한 값일 수도 있어요) 형태의 쿼리로 (재)생성돼요. Integer, Date, String 등 어떤 타입이든 될 수 있어요. 참고: 현재는 dot 표기법을 지원하지 않으니 필드는 문서의 최상위 레벨에 있어야 해요. | | String | | bridgeErrorHandler (consumer (advanced)) | 컨슈머를 Camel 라우팅 에러 핸들러(Error Handler)에 브리지할 수 있게 해 줘요. 즉, Camel 컨슈머가 들어오는 메시지를 집어 올리는 동안 등에서 예외가 발생하면(가능한 경우), 그 예외를 메시지로 처리해서 라우팅 에러 핸들러가 다루게 돼요. 중요: 이것은 서드파티 컴포넌트가 예외가 던져졌을 때 Camel에 알릴 수 있도록 허용하는 경우에만 가능해요. 일부 컴포넌트는 내부에서만 처리하기 때문에 bridgeErrorHandler를 쓸 수 없어요. 다른 상황에서는 Camel 컴포넌트를 개선해서 서드파티 컴포넌트에 연결(hook)해 이를 가능하게 만들 수도 있어요. 기본으로 컨슈머는 org.apache.camel.spi.ExceptionHandler로 예외를 처리하는데, WARN 또는 ERROR 레벨로 로그를 남기고 무시해요. | false | boolean | | exceptionHandler (consumer (advanced)) | 컨슈머가 커스텀 ExceptionHandler를 쓰게 할 수 있어요. bridgeErrorHandler 옵션이 활성화되어 있으면 이 옵션은 사용되지 않는다는 점에 주의하세요. 기본으로 컨슈머는 예외를 처리하는데, WARN 또는 ERROR 레벨로 로그를 남기고 무시해요. | | ExceptionHandler | | exchangePattern (consumer (advanced)) |

컨슈머가 exchange를 만들 때 exchange 패턴을 설정해요.

Enum values:

InOnly
InOut

MongoDB 클라이언트가 레플리카 셋의 멤버들에게 읽기 연산을 라우팅하는 방식을 설정해요. 가능한 값은 PRIMARY, PRIMARY_PREFERRED, SECONDARY, SECONDARY_PREFERRED 또는 NEAREST예요.

Enum values:

PRIMARY
PRIMARY_PREFERRED
SECONDARY
SECONDARY_PREFERRED
NEAREST

connection bean에, standalone mongod·replicaset·cluster에 대한 쓰기 연산에서 MongoDB에 요청하는 승인 수준을 설정해요. 가능한 값은 ACKNOWLEDGED, W1, W2, W3, UNACKNOWLEDGED, JOURNALED 또는 MAJORITY예요.

Enum values:

ACKNOWLEDGED
W1
W2
W3
UNACKNOWLEDGED
JOURNALED
MAJORITY

메시지 헤더 (Message Headers)

MongoDB 컴포넌트는 다음 메시지 헤더(들)를 지원해요:

Name	Description	Default	Type
CamelMongoDbOperation (producer) Constant: `OPERATION_HEADER`	이 엔드포인트가 MongoDB에 대해 실행할 연산이에요.		MongoDbOperation or String
CamelMongoDbResultTotalSize (producer findAll) Constant: `RESULT_TOTAL_SIZE`	쿼리에 매칭되는 객체 수예요. limit/skip은 고려하지 않아요.		Integer
CamelMongoDbResultPageSize (producer findAll) Constant: `RESULT_PAGE_SIZE`	쿼리에 매칭되는 객체 수예요. limit/skip은 고려하지 않아요.		Integer
CamelMongoDbCriteria (producer) Constant: `CRITERIA`	MongoDB에 대해 실행할 쿼리예요.		Bson
CamelMongoDbFieldsProjection (producer) Constant: `FIELDS_PROJECTION`	프로젝션 문서예요.		Bson
CamelMongoDbBatchSize (producer findAll aggregate) Constant: `BATCH_SIZE`	배치당 문서 수예요.		Integer
CamelMongoDbNumToSkip (producer findAll) Constant: `NUM_TO_SKIP`	커서 앞부분에서 주어진 개수만큼 요소를 버려요.		Integer
CamelMongoDbMultiUpdate (producer update) Constant: `MULTIUPDATE`	업데이트를 매칭되는 모든 객체에 적용할지 여부예요. See http://www.mongodb.org/display/DOCS/AtomicOperationsAtomic Operations.		Boolean
CamelMongoDbUpsert (producer update) Constant: `UPSERT`	요소가 없을 때 데이터베이스가 만들어야 하는지 여부예요.		Boolean
CamelMongoDbRecordsAffected (producer) Constant: `RECORDS_AFFECTED`	수정 또는 삭제된 레코드 수예요.		long
CamelMongoDbRecordsMatched (producer) Constant: `RECORDS_MATCHED`	쿼리에 매칭된 문서 수예요.		long
CamelMongoDbSortBy (producer) Constant: `SORT_BY`	정렬 criteria예요.		Bson or Document
CamelMongoDbDatabase (common) Constant: `DATABASE`	대상 MongoDB 데이터베이스 이름이에요.		String
CamelMongoDbCollection (common) Constant: `COLLECTION`	이 엔드포인트에 바인딩할 MongoDB 컬렉션 이름이에요.		String
CamelMongoDbCollectionIndex (producer) Constant: `COLLECTION_INDEX`	즉석에서 만들 동적 인덱스 목록이에요.		List
CamelMongoDbLimit (producer findAll) Constant: `LIMIT`	반환되는 요소 수를 제한해요.		Integer
CamelMongoDbTailable (consumer) Constant: `FROM_TAILABLE`	tailable에서 왔는지 여부예요.		Boolean
CamelMongoWriteResult (producer) Constant: `WRITERESULT`	쓰기 연산의 결과예요.		Object
CamelMongoOid (producer) Constant: `OID`	insert된 레코드(들)의 OID(들)예요.		Object or List
CamelMongoDbDistinctQueryField (producer) Constant: `DISTINCT_QUERY_FIELD`	distinct 값을 얻고 싶은 지정 필드 이름이에요.		String
CamelMongoDbAllowDiskUse (producer findAll aggregate) Constant: `ALLOW_DISK_USE`	allowDiskUse MongoDB 플래그를 설정해요. MongoDB Server 4.3.1부터 지원해요. 이보다 오래된 MongoDB Server 버전에서 이 헤더를 쓰면 쿼리가 실패할 수 있어요.		Boolean
CamelMongoDbBulkOrdered (producer bulkWrite) Constant: `BULK_ORDERED`	순차 또는 비순차 연산 실행을 수행해요.	TRUE	Boolean
_id (consumer changeStreams) Constant: `MONGO_ID`	insert, replace, delete 또는 update 연산(즉 CRUD 연산)으로 생성되거나 수정된 문서의 _id예요. MongoDB가 id를 생성했다면 org.bson.types.ObjectId이고, 그렇지 않으면 자연스러운 Java 타입의 id예요: 복합 키의 경우 String, 숫자 또는 Document. invalidate, drop, rename, dropDatabase처럼 단일 문서에 속하지 않는 이벤트에서는 이 헤더가 없어요.		Object
CamelMongoDbStreamOperationType (consumer changeStreams) Constant: `STREAM_OPERATION_TYPE`	발생한 연산의 타입이에요. insert, delete, replace, update, drop, rename, dropDatabase, invalidate 중 하나일 수 있어요.		String
CamelMongoDbReturnDocumentType (producer update one and return) Constant: `RETURN_DOCUMENT`
어떤 문서를 반환할지 지시해요. update and return 원자적 연산에서 업데이트 전의 문서일 수도, 후의 문서일 수도 있어요.

Enum values:

BEFORE
AFTER

사용법 (Usage)

Spring XML에서 데이터베이스 설정 (Configuration of a database in Spring XML)

다음 Spring XML은 MongoDB 인스턴스로의 연결을 정의하는 bean을 만들어요.

mongo java driver 3부터 WriteConcern과 readPreference 옵션은 동적으로 바꿀 수 없게 됐어요. 두 옵션은 mongoClient 객체에 정의해요:

<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:mongo="http://www.springframework.org/schema/data/mongo"
xsi:schemaLocation="http://www.springframework.org/schema/context
      http://www.springframework.org/schema/context/spring-context.xsd
      http://www.springframework.org/schema/data/mongo
      http://www.springframework.org/schema/data/mongo/spring-mongo.xsd
      http://www.springframework.org/schema/beans
      http://www.springframework.org/schema/beans/spring-beans.xsd">

  <mongo:mongo-client id="mongoBean" host="${mongo.url}" port="${mongo.port}" credentials="${mongo.user}:${mongo.pass}@${mongo.dbname}">
    <mongo:client-options write-concern="NORMAL" />
  </mongo:mongo-client>
</beans>

MongoDB 연산 - producer 엔드포인트 (MongoDB operations - producer endpoints)

쿼리 연산 (Query operations)

findById

이 연산은 _id 필드가 IN 메시지 바디 내용과 일치하는 요소를 컬렉션에서 딱 하나만 가져와요. 들어오는 객체는 Bson 타입과 동등한 것이면 무엇이든 돼요. http://bsonspec.org/spec.html와 http://www.mongodb.org/display/DOCS/Java+Types를 참고하세요.

Java
XML
YAML

from("direct:findById")
    .to("mongodb:myDb?database=flights&collection=tickets&operation=findById")
    .to("mock:resultFindById");

<route>
  <from uri="direct:findById"/>
  <to uri="mongodb:myDb?database=flights&amp;collection=tickets&amp;operation=findById"/>
  <to uri="mock:resultFindById"/>
</route>

- route:
    from:
      uri: direct:findById
      steps:
        - to:
            uri: mongodb:myDb
            parameters:
              database: flights
              collection: tickets
              operation: findById
        - to:
            uri: mock:resultFindById

기본 _id는 Mongo가 ObjectId 타입으로 다루니까, 필요하면 적절히 변환해야 할 수 있어요.

Java
XML
YAML

from("direct:findById")
    .convertBodyTo(ObjectId.class)
    .to("mongodb:myDb?database=flights&collection=tickets&operation=findById")
    .to("mock:resultFindById");

<route>
  <from uri="direct:findById"/>
  <convertBodyTo type="org.bson.types.ObjectId"/>
  <to uri="mongodb:myDb?database=flights&amp;collection=tickets&amp;operation=findById"/>
  <to uri="mock:resultFindById"/>
</route>

- route:
    from:
      uri: direct:findById
      steps:
        - convertBodyTo:
            type: org.bson.types.ObjectId
        - to:
            uri: mongodb:myDb
            parameters:
              database: flights
              collection: tickets
              operation: findById
        - to:
            uri: mock:resultFindById

팁 선택 파라미터를 지원해요

이 연산은 프로젝션 연산자를 지원해요. Specifying a fields filter (projection)를 참고하세요.

findOneByQuery

MongoDB 쿼리 셀렉터에 매칭되는 첫 번째 요소를 컬렉션에서 가져와요. CamelMongoDbCriteria 헤더가 설정되어 있으면 그 값이 쿼리 셀렉터로 쓰여요. CamelMongoDbCriteria 헤더가 _null_이면 IN 메시지 바디가 쿼리 셀렉터로 쓰여요. 어느 쪽이든 쿼리 셀렉터는 Bson 타입이거나 Bson으로 변환 가능한 것(예: JSON 문자열이나 HashMap)이어야 해요. 자세한 내용은 Type conversions를 참고하세요.

MongoDB Driver가 제공하는 Filters로 쿼리 셀렉터를 만드세요.

쿼리 셀렉터가 없는 예제 (Example without a query selector (returns the first document in a collection))

Java
XML
YAML

from("direct:findOneByQuery")
    .to("mongodb:myDb?database=flights&collection=tickets&operation=findOneByQuery")
    .to("mock:resultFindOneByQuery");

<route>
  <from uri="direct:findOneByQuery"/>
  <to uri="mongodb:myDb?database=flights&amp;collection=tickets&amp;operation=findOneByQuery"/>
  <to uri="mock:resultFindOneByQuery"/>
</route>

- route:
    from:
      uri: direct:findOneByQuery
      steps:
        - to:
            uri: mongodb:myDb
            parameters:
              database: flights
              collection: tickets
              operation: findOneByQuery
        - to:
            uri: mock:resultFindOneByQuery

쿼리 셀렉터가 있는 예제 (Example with a query selector (returns the first matching document in a collection):)

Java 전용: Bson Filters 객체가 필요해요

from("direct:findOneByQuery")
    .setHeader("CamelMongoDbCriteria", constant(Filters.eq("name", "Raul Kripalani")))
    .to("mongodb:myDb?database=flights&collection=tickets&operation=findOneByQuery")
    .to("mock:resultFindOneByQuery");

팁 선택 파라미터를 지원해요

이 연산은 프로젝션 연산자와 sort 절을 지원해요. Specifying a fields filter (projection), Specifying a sort clause를 참고하세요.

findAll

findAll 연산은 쿼리에 매칭되는 모든 문서를 반환하고, 쿼리가 없으면 컬렉션에 담긴 모든 문서를 반환해요. 쿼리 객체는 CamelMongoDbCriteria 헤더에서 추출해요. CamelMongoDbCriteria 헤더가 null이면 쿼리 객체를 메시지 바디에서 추출해요. 즉, Bson 타입이거나 Bson으로 변환 가능한 것이어야 해요. JSON 문자열이나 Hashmap도 돼요. 자세한 내용은 Type conversions를 참고하세요.

쿼리 셀렉터가 없는 예제 (Example without a query selector (returns all documents in a collection))

Java
XML
YAML

from("direct:findAll")
    .to("mongodb:myDb?database=flights&collection=tickets&operation=findAll")
    .to("mock:resultFindAll");

<route>
  <from uri="direct:findAll"/>
  <to uri="mongodb:myDb?database=flights&amp;collection=tickets&amp;operation=findAll"/>
  <to uri="mock:resultFindAll"/>
</route>

- route:
    from:
      uri: direct:findAll
      steps:
        - to:
            uri: mongodb:myDb
            parameters:
              database: flights
              collection: tickets
              operation: findAll
        - to:
            uri: mock:resultFindAll

쿼리 셀렉터가 있는 예제 (Example with a query selector (returns all matching documents in a collection))

Java 전용: Bson Filters 객체가 필요해요

from("direct:findAll")
    .setHeader("CamelMongoDbCriteria", constant(Filters.eq("name", "Raul Kripalani")))
    .to("mongodb:myDb?database=flights&collection=tickets&operation=findAll")
    .to("mock:resultFindAll");

outputType=MongoIterable 옵션과 배치 크기 예제 (Example with option outputType=MongoIterable and batch size)

Java 전용: Bson Filters 객체가 필요해요

from("direct:findAll")
    .setHeader("CamelMongoDbBatchSize", constant(10))
    .setHeader("CamelMongoDbCriteria", constant(Filters.eq("name", "Raul Kripalani")))
    .to("mongodb:myDb?database=flights&collection=tickets&operation=findAll&outputType=MongoIterable")
    .to("mock:resultFindAll");

팁 선택 파라미터를 지원해요

이 연산은 프로젝션 연산자와 sort 절을 지원해요. Specifying a fields filter (projection), Specifying a sort clause를 참고하세요.

count

컬렉션에 담긴 객체의 총 개수를 반환하고, OUT 메시지 바디로 Long을 돌려줘요.
다음 예제는 "dynamicCollectionName" 컬렉션의 레코드 수를 세요. dynamicity가 켜져 있다는 점에 주목하세요. 그 결과 이 연산은 "notableScientists" 컬렉션이 아니라 "dynamicCollectionName" 컬렉션에 대해 실행돼요.

Java
XML
YAML

from("direct:count")
    .to("mongodb:myDb?database=tickets&collection=flights&operation=count&dynamicity=true");

<route>
  <from uri="direct:count"/>
  <to uri="mongodb:myDb?database=tickets&amp;collection=flights&amp;operation=count&amp;dynamicity=true"/>
</route>

- route:
    from:
      uri: direct:count
      steps:
        - to:
            uri: mongodb:myDb
            parameters:
              database: tickets
              collection: flights
              operation: count
              dynamicity: true

다른 컬렉션을 동적으로 대상으로 삼으려면 CamelMongoDbCollection 헤더를 설정하세요:

Java 전용: Java 테스트 API (ProducerTemplate)

Long result = template.requestBodyAndHeader("direct:count", "irrelevantBody",
        "CamelMongoDbCollection", "dynamicCollectionName");

쿼리를 제공할 수도 있어요. 쿼리 객체는 CamelMongoDbCriteria 헤더에서 추출해요. CamelMongoDbCriteria 헤더가 null이면 쿼리 객체를 메시지 바디에서 추출해요. 즉, Bson 타입이거나 Bson으로 변환 가능한 것이어야 해요. 그러면 연산은 criteria에 매칭되는 문서 수를 반환해요.

Java 전용: Java 테스트 API (ProducerTemplate)

Document query = ...
Long count = template.requestBodyAndHeader("direct:count", query,
        "CamelMongoDbCollection", "dynamicCollectionName");

`fields` 필터(프로젝션) 지정 (Specifying a `fields` filter (projection))

쿼리 연산은 기본적으로 매칭되는 객체를 그대로(모든 필드 포함) 반환해요. 문서가 크고 필드의 일부만 가져오고 싶다면, 모든 쿼리 연산에서 필드 필터를 지정할 수 있어요. CamelMongoDbFieldsProjection 헤더에 해당하는 Bson(Bson으로 변환 가능한 타입도 돼요. JSON 문자열, Map 등)을 설정하기만 하면 돼요.

MongoDB의 Projections를 사용해 Bson 생성을 단순화한 예제예요. _id와 boringField를 제외한 모든 필드를 가져와요:

Java 전용: Bson Projection을 쓰는 Java 테스트 API

// route: from("direct:findAll").to("mongodb:myDb?database=flights&collection=tickets&operation=findAll")
Bson fieldProjection = Projection.exclude("_id", "boringField");
Object result = template.requestBodyAndHeader("direct:findAll", ObjectUtils.NULL,
        "CamelMongoDbFieldsProjection", fieldProjection);

정렬 절 지정 (Specifying a sort clause)

특정 필드로 정렬해서 컬렉션에서 min/max 레코드를 가져와야 하는 요구가 자주 있어요. MongoDB의 Sorts를 사용하면 Bson 생성을 단순화할 수 있어요. 이 예제는 _id와 boringField를 제외한 모든 필드를 가져와요:

Java 전용: Bson Sorts를 쓰는 Java 테스트 API

// route: from("direct:findAll").to("mongodb:myDb?database=flights&collection=tickets&operation=findAll")
Bson sorts = Sorts.descending("_id");
Object result = template.requestBodyAndHeader("direct:findAll", ObjectUtils.NULL,
        "CamelMongoDbSortBy", sorts);

Camel 라우트에서는 SORT_BY 헤더를 findOneByQuery 연산과 함께 써서 같은 결과를 낼 수 있어요. FIELDS_PROJECTION 헤더도 함께 지정하면, 이 연산은 다른 컴포넌트(예: 파라미터화된 MyBatis SELECT 쿼리)로 바로 넘길 수 있는 단일 field/value 쌍을 반환해요. 이 예제는 컬렉션에서 시간상 가장 최신 문서를 가져와 documentTimestamp 필드를 기준으로 결과를 단일 필드로 줄여요:

Java 전용: Bson Sorts와 Projection 객체가 필요해요

from("direct:someTriggeringEvent")
    .setHeader("CamelMongoDbSortBy", constant(Sorts.descending("documentTimestamp")))
    .setHeader("CamelMongoDbFieldsProjection", constant(Projection.include("documentTimestamp")))
    .setBody(constant("{}"))
    .to("mongodb:myDb?database=local&collection=myDemoCollection&operation=findOneByQuery")
    .to("direct:aMyBatisParameterizedSelect");

생성/업데이트 연산 (Create/update operations)

insert

IN 메시지 바디에서 가져온 새 객체를 MongoDB 컬렉션에 insert해요. Document나 List로 바꾸려고 타입 변환을 시도해요. 두 가지 모드를 지원해요: 단일 insert와 다중 insert. 다중 insert의 경우 엔드포인트는 객체의 List, Array 또는 Collections를 기대해요. 어떤 타입이든 Document이거나 변환 가능하면 돼요. 예제:

Java
XML
YAML

from("direct:insert")
    .to("mongodb:myDb?database=flights&collection=tickets&operation=insert");

<route>
  <from uri="direct:insert"/>
  <to uri="mongodb:myDb?database=flights&amp;collection=tickets&amp;operation=insert"/>
</route>

- route:
    from:
      uri: direct:insert
      steps:
        - to:
            uri: mongodb:myDb
            parameters:
              database: flights
              collection: tickets
              operation: insert

이 연산은 WriteResult를 반환하고, WriteConcern이나 invokeGetLastError 옵션 값에 따라 getLastError()가 이미 호출됐을 수도, 아닐 수도 있어요. 쓰기 연산의 최종 결과에 접근하려면 WriteResult에서 getLastError()나 getCachedLastError()를 호출해 CommandResult를 꺼내야 해요. 그런 다음 CommandResult.ok(), CommandResult.getErrorMessage() 또는 CommandResult.getException()을 호출해서 결과를 검증할 수 있어요.

새 객체의 _id는 컬렉션에서 유일해야 한다는 점에 주의하세요. 값을 지정하지 않으면 MongoDB가 자동으로 만들어 줘요. 하지만 값을 지정했는데 유일하지 않으면 insert 연산은 실패해요 (Camel이 이를 알아차리려면 invokeGetLastError를 켜거나 쓰기 결과를 기다리는 WriteConcern을 설정해야 해요).

이건 컴포넌트의 한계가 아니라, 높은 처리량을 위해 MongoDB가 동작하는 방식이에요. 커스텀 _id를 쓴다면 애플리케이션 레벨에서 유일함을 보장해야 해요 (그리고 이건 좋은 습관이기도 해요).

insert된 레코드(들)의 OID(들)는 CamelMongoOid 키로 메시지 헤더에 저장돼요. 단일 insert면 저장된 값은 org.bson.types.ObjectId이고, 여러 레코드가 insert됐다면 java.util.List<org.bson.types.ObjectId>예요.

MongoDB Java Driver 3.x에서 insertOne과 insertMany 연산은 void를 반환해요. Camel의 insert 연산은 insert된 문서 또는 문서 목록을 반환해요. 필요하면 각 문서가 새 OID로 갱신된다는 점에 주의하세요.

save

save 연산은 upsert(UPdate, inSERT) 연산과 같아요. 레코드를 업데이트하고, 없으면 insert하는데, 모두 하나의 원자적 연산으로 처리돼요. MongoDB는 _id 필드를 기준으로 매칭을 수행해요.

업데이트의 경우 객체가 통째로 교체되고 MongoDB의 $modifiers 사용이 허용되지 않는다는 점에 주의하세요. 따라서 이미 존재하는 객체를 조작하고 싶다면 두 가지 선택지가 있어요:

전체 객체를 모든 필드와 함께 먼저 조회하는 쿼리를 수행하고(비효율적일 수 있어요), Camel 내부에서 객체를 고친 뒤 다시 저장해요.
$modifiers와 함께 update 연산을 사용해요. 그러면 업데이트를 서버 쪽에서 대신 실행해요. upsert 플래그를 켤 수도 있는데, 이러면 insert가 필요할 때 MongoDB가 필터 쿼리 객체에 $modifiers를 적용하고 그 결과를 insert해요.

저장하려는 문서에 _id 속성이 없으면 이 연산은 insert가 되고, 새로 만들어진 _id가 CamelMongoOid 헤더에 담겨요.

예를 들면:

Java
XML
YAML

from("direct:insert")
    .to("mongodb:myDb?database=flights&collection=tickets&operation=save");

<route>
  <from uri="direct:insert"/>
  <to uri="mongodb:myDb?database=flights&amp;collection=tickets&amp;operation=save"/>
</route>

- route:
    from:
      uri: direct:insert
      steps:
        - to:
            uri: mongodb:myDb
            parameters:
              database: flights
              collection: tickets
              operation: save

Java 전용: Java 테스트 API (ProducerTemplate)

// route: from("direct:insert").to("mongodb:myDb?database=flights&collection=tickets&operation=save");
Document docForSave = new Document();
docForSave.put("key", "value");
Object result = template.requestBody("direct:insert", docForSave);

update

컬렉션에서 하나 이상의 레코드를 업데이트해요. 필터 쿼리와 업데이트 규칙이 필요해요.

CamelMongoDbCriteria 헤더를 Bson으로 필터를 정의하고, 바디에 Bson으로 업데이트 규칙을 정의할 수 있어요.

참고 enrich 후 업데이트 (Update after enrich)

업데이트 전에 CamelMongoDbCriteria 헤더를 Bson으로 써서 mongodb를 조회하는 경우, aggregation strategy와 함께 enrich 패턴을 쓴 뒤 mongodb update를 적용한다면 aggregation 중에 그 헤더를 결과 Camel exchange에서 제거해야 한다는 점에 유의하세요. aggregation 중에 이 헤더를 제거하지 않거나, mongodb producer 엔드포인트로 Camel exchange를 보내기 전에 CamelMongoDbCriteria 헤더를 재정의하지 않으면, mongodb를 업데이트할 때 잘못된 Camel exchange 페이로드가 만들어질 수 있어요.

두 번째 방법은 정확히 두 개의 요소를 담은 List<Bson>을 IN 메시지 바디로 요구해요:

요소 1 (인덱스 0) ⇒ 필터 쿼리 ⇒ 어떤 객체가 영향을 받을지 결정해요. 일반적인 쿼리 객체와 같아요
요소 2 (인덱스 1) ⇒ 업데이트 규칙 ⇒ 매칭된 객체를 어떻게 업데이트할지 정해요. MongoDB의 모든 modifier operations를 지원해요.

참고 다중 업데이트 (Multiupdates)

기본으로 MongoDB는 필터 쿼리에 여러 객체가 매칭돼도 1개 객체만 업데이트해요. 매칭되는 모든 레코드를 업데이트하라고 지시하려면 CamelMongoDbMultiUpdate IN 메시지 헤더를 true로 설정하세요.

CamelMongoDbRecordsAffected 키의 헤더가 업데이트된 레코드 수와 함께 반환돼요 (WriteResult.getN()에서 복사해요).

예를 들어 다음은 filterField 필드가 true인 모든 레코드를, "scientist" 필드 값을 "Darwin"으로 설정해서 업데이트해요:

Java 전용: Bson 필터와 업데이트를 쓰는 Java 테스트 API

// route: from("direct:update").to("mongodb:myDb?database=science&collection=notableScientists&operation=update");
List<Bson> body = new ArrayList<>();
Bson filterField = Filters.eq("filterField", true);
body.add(filterField);
BsonDocument updateObj = new BsonDocument().append("$set", new BsonDocument("scientist", new BsonString("Darwin")));
body.add(updateObj);
Object result = template.requestBodyAndHeader("direct:update", body, "CamelMongoDbMultiUpdate", true);

Java 전용: criteria 헤더를 쓰는 Java 테스트 API

// route: from("direct:update").to("mongodb:myDb?database=science&collection=notableScientists&operation=update");
Map<String, Object> headers = new HashMap<>(2);
headers.put("CamelMongoDbMultiUpdate", true);
headers.put("CamelMongoDbCriteria", Filters.eq("filterField", true));
Bson updateObj = Updates.set("scientist", "Darwin");
Object result = template.requestBodyAndHeaders("direct:update", updateObj, headers);

Java 전용: JSON 문자열을 쓰는 Java 테스트 API

// route: from("direct:update").to("mongodb:myDb?database=science&collection=notableScientists&operation=update");
String updateObj = "[{\"filterField\": true}, {\"$set\", {\"scientist\", \"Darwin\"}}]";
Object result = template.requestBodyAndHeader("direct:update", updateObj, "CamelMongoDbMultiUpdate", true);

삭제 연산 (Delete operations)

remove

컬렉션에서 매칭되는 레코드를 제거해요. IN 메시지 바디가 제거 필터 쿼리 역할을 하고, DBObject 타입이거나 변환 가능한 타입이어야 해요.
다음 예제는 science 데이터베이스의 notableScientists 컬렉션에서 'conditionField' 필드가 true인 객체를 모두 제거해요:

Java 전용: Bson 필터를 쓰는 Java 테스트 API

// route: from("direct:remove").to("mongodb:myDb?database=science&collection=notableScientists&operation=remove");
Bson conditionField = Filters.eq("conditionField", true);
Object result = template.requestBody("direct:remove", conditionField);

CamelMongoDbRecordsAffected 키의 헤더가 int 타입으로 반환되고, 삭제된 레코드 수를 담고 있어요 (WriteResult.getN()에서 복사해요).

벌크 쓰기 연산 (Bulk Write Operations)

bulkWrite

실행 순서를 제어하면서 쓰기 연산을 벌크로 수행해요. insert, update, delete 연산용 커맨드를 담은 List<WriteModel<Document>>가 IN 메시지 바디로 필요해요.

다음 예제는 새 과학자 "Pierre Curie"를 insert하고, id가 "5"인 레코드의 "scientist" 필드 값을 "Marie Curie"로 설정해 update하고, id가 "3"인 레코드를 delete해요:

Java 전용: WriteModel 객체를 쓰는 Java 테스트 API

// route: from("direct:bulkWrite").to("mongodb:myDb?database=science&collection=notableScientists&operation=bulkWrite");
List<WriteModel<Document>> bulkOperations = Arrays.asList(
        new InsertOneModel<>(new Document("scientist", "Pierre Curie")),
        new UpdateOneModel<>(new Document("_id", "5"),
                new Document("$set", new Document("scientist", "Marie Curie"))),
        new DeleteOneModel<>(new Document("_id", "3")));

BulkWriteResult result = template.requestBody("direct:bulkWrite", bulkOperations, BulkWriteResult.class);

기본으로 연산은 순서대로 실행되고, 목록에 남은 쓰기 연산을 처리하지 않은 채 첫 번째 쓰기 에서 멈춰요. 목록의 나머지 쓰기 연산을 계속 처리하라고 MongoDB에 지시하려면 CamelMongoDbBulkOrdered IN 메시지 헤더를 false로 설정하세요. 비순차 연산은 병렬로 실행되고 이 동작은 보장되지 않아요.

기타 연산 (Other operations)

aggregate

바디에 담긴 주어진 파이프라인으로 애그리게이션을 수행해요. 애그리게이션은 길고 무거운 연산일 수 있으니 신중히 사용하세요.

Java 전용: Bson 애그리게이션 파이프라인이 필요해요

// route: from("direct:aggregate").to("mongodb:myDb?database=science&collection=notableScientists&operation=aggregate");
List<Bson> aggregate = Arrays.asList(match(or(eq("scientist", "Darwin"), eq("scientist", "Einstein"))),
        group("$scientist", sum("count", 1)));
from("direct:aggregate")
    .setBody(constant(aggregate))
    .to("mongodb:myDb?database=science&collection=notableScientists&operation=aggregate")
    .to("mock:resultAggregate");

기본으로 모든 결과의 List가 반환돼요. 결과 크기에 따라 메모리를 많이 쓸 수 있어요. 더 안전한 대안은 outputType=MongoIterable로 설정하는 거예요. 다음 Processor는 메시지 바디에서 iterable을 보게 되고, 결과를 하나씩 넘나들 수 있어요. 따라서 배치 크기를 정하고 iterable을 반환하면 결과를 효율적으로 가져오고 처리할 수 있어요.

예를 들면 다음과 같아요:

Java 전용: Bson 애그리게이션 파이프라인이 필요해요

List<Bson> aggregate = Arrays.asList(match(or(eq("scientist", "Darwin"), eq("scientist", "Einstein"))),
        group("$scientist", sum("count", 1)));
from("direct:aggregate")
    .setHeader("CamelMongoDbBatchSize", constant(10))
    .setBody(constant(aggregate))
    .to("mongodb:myDb?database=science&collection=notableScientists&operation=aggregate&outputType=MongoIterable")
    .split(body())
    .streaming()
    .to("mock:resultAggregate");

.split(body())를 호출하는 것만으로 항목을 라우트로 하나씩 보낼 수 있지만, 그래도 모든 항목을 먼저 메모리에 로드해요. 배치 단위로 메모리에 데이터를 로드하려면 .streaming() 호출이 필요해요.

getDbStats

MongoDB 셸에서 db.stats() 커맨드를 실행하는 것과 같아요. 데이터베이스에 대한 유용한 통계 수치를 보여 줘요.
예를 들면:

> db.stats(); { "db" : "test", "collections" : 7, "objects" : 719, "avgObjSize" : 59.73296244784423, "dataSize" : 42948, "storageSize" : 1000058880, "numExtents" : 9, "indexes" : 4, "indexSize" : 32704, "fileSize" : 1275068416, "nsSizeMB" : 16, "ok" : 1 }

Usage example:

Java 전용: Java 테스트 API (ProducerTemplate)

// from("direct:getDbStats").to("mongodb:myDb?database=flights&collection=tickets&operation=getDbStats");
Object result = template.requestBody("direct:getDbStats", "irrelevantBody");
assertTrue("Result is not of type Document", result instanceof Document);

이 연산은 셸에 표시된 것과 비슷한 데이터 구조를 OUT 메시지 바디에 Document 형태로 반환해요.

getColStats

MongoDB 셸에서 db.collection.stats() 커맨드를 실행하는 것과 같아요. 컬렉션에 대한 유용한 통계 수치를 보여 줘요.
예를 들면:

> db.camelTest.stats(); { "ns" : "test.camelTest", "count" : 100, "size" : 5792, "avgObjSize" : 57.92, "storageSize" : 20480, "numExtents" : 2, "nindexes" : 1, "lastExtentSize" : 16384, "paddingFactor" : 1, "flags" : 1, "totalIndexSize" : 8176, "indexSizes" : { "_id_" : 8176 }, "ok" : 1 }

Usage example:

Java 전용: Java 테스트 API (ProducerTemplate)

// from("direct:getColStats").to("mongodb:myDb?database=flights&collection=tickets&operation=getColStats");
Object result = template.requestBody("direct:getColStats", "irrelevantBody");
assertTrue("Result is not of type Document", result instanceof Document);

이 연산은 셸에 표시된 것과 비슷한 데이터 구조를 OUT 메시지 바디에 Document 형태로 반환해요.

command

바디를 데이터베이스에서 커맨드로 실행해요. 호스트 정보, replication 또는 sharding 상태를 얻는 것 같은 관리 작업에 유용해요.

이 연산에는 collection 파라미터가 사용되지 않아요.

Java 전용: Java 테스트 API (ProducerTemplate)

// route: from("command").to("mongodb:myDb?database=science&operation=command");
DBObject commandBody = new BasicDBObject("hostInfo", "1");
Object result = template.requestBody("direct:command", commandBody);

동적 연산 (Dynamic operations)

Exchange는 CamelMongoDbOperation 헤더를 설정해서 엔드포인트의 고정 연산을 재정의할 수 있어요. 지원되는 값은 MongoDbOperation 열거형이 결정하고, 엔드포인트 URI의 operation 파라미터에 허용되는 값과 일치해요.

예를 들면:

Java 전용: Java 테스트 API (ProducerTemplate)

// from("direct:insert").to("mongodb:myDb?database=flights&collection=tickets&operation=insert");
Object result = template.requestBodyAndHeader("direct:insert", "irrelevantBody", "CamelMongoDbOperation", "count");
assertTrue("Result is not of type Long", result instanceof Long);

컨슈머 (Consumers)

컨슈머에는 여러 타입이 있어요:

Tailable Cursor Consumer
Change Streams Consumer

Tailable Cursor Consumer

MongoDB는 커서를 *nix 시스템의 tail -f 커맨드처럼 열어 두는 방식으로, 컬렉션에서 흘러 들어오는 데이터를 즉시 소비하는 메커니즘을 제공해요. 새 데이터가 생기는 대로 서버가 클라이언트로 밀어 주기 때문에, 클라이언트가 정해진 간격으로 핑을 보내 새 데이터를 가져오는 스케줄 폴링보다 이 메커니즘이 훨씬 효율적이에요. 불필요한 네트워크 트래픽도 줄여 줘요.

tailable 커서를 쓰기 위한 조건은 딱 하나예요: 컬렉션이 "capped collection"이어야 해요. capped collection은 N개의 객체만 담고, 한도에 도달하면 MongoDB가 원래 insert된 순서대로 오래된 객체를 비워요. 자세한 내용은 http://www.mongodb.org/display/DOCS/Tailable+Cursors를 참고하세요.

Camel MongoDB 컴포넌트는 tailable cursor consumer를 구현해서 이 기능을 Camel 라우트에서 쓸 수 있게 해 줘요. 새 객체가 insert되면 MongoDB가 자연 순서대로 Document로 tailable cursor consumer에 밀어 주고, consumer는 이를 Exchange로 변환해 라우트 로직을 트리거해요.

tailable cursor consumer의 동작 방식 (How the tailable cursor consumer works)

커서를 tailable 커서로 만들려면 커서를 처음 생성할 때 몇 가지 특수 플래그를 MongoDB에 알려 줘야 해요. 일단 만들어지면 커서는 열린 상태로 유지되고, MongoCursor.next() 메서드를 호출하면 새 데이터가 도착할 때까지 블록해요. 하지만 MongoDB 서버는 정해지지 않은 기간 동안 새 데이터가 나타나지 않으면 커서를 죽일 권리를 자기 자신에게 reserved해요. 새 데이터를 계속 소비하고 싶다면 커서를 다시 생성(regenerate)해야 해요. 그러려면 어디까지 읽었는지 위치를 기억해야 하고, 그렇지 않으면 다시 처음부터 소비하게 돼요.

Camel MongoDB tailable cursor consumer가 이 모든 작업을 대신 처리해 줘요. 증가하는 성격의 데이터 필드 키를 제공해야 하는데, 이 필드가 커서를 다시 생성할 때마다 위치를 잡는 마커 역할을 해요. 예를 들면 timestamp, 순차 ID 같은 것이에요. MongoDB가 지원하는 어떤 데이터 타입도 될 수 있어요. Date, String, Integer가 잘 동작하는 걸로 알려져 있어요. 이 컴포넌트의 맥락에서 이 메커니즘을 "tail tracking"이라고 불러요.

consumer는 이 필드의 마지막 값을 기억하고, 커서를 다시 생성해야 할 때마다 increasingField > lastValue 같은 필터로 쿼리를 실행해서 아직 읽지 않은 데이터만 소비해요.

증가 필드 설정: 엔드포인트 URI의 tailTrackingIncreasingField 옵션에 증가 필드의 키를 설정하세요. Camel 2.10에서는 이 필드가 데이터의 최상위 필드여야 해요. 이 필드에 대한 중첩 탐색은 아직 지원되지 않아요. 즉, "timestamp" 필드는 괜찮지만 nested.timestamp는 동작하지 않아요. 중첩된 증가 필드 지원이 꼭 필요하면 Camel JIRA에 티켓을 열어 주세요.

커서 재생성 지연: 초기화 시점에 새 데이터가 아직 없으면 MongoDB가 커서를 즉시 죽인다는 점에 유의하세요. 이 경우 서버에 무리를 주지 않으려고 cursorRegenerationDelay 옵션(기본값 1000ms.)이 도입됐으니, 필요에 맞게 수정하면 돼요.

예제:

Java
XML
YAML

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

<route id="tailableCursorConsumer1" autoStartup="false">
  <from uri="mongodb:myDb?database=flights&amp;collection=cancellations&amp;tailTrackIncreasingField=departureTime"/>
  <to uri="mock:test"/>
</route>

- route:
    id: tailableCursorConsumer1
    autoStartup: false
    from:
      uri: mongodb:myDb
      parameters:
        database: flights
        collection: cancellations
        tailTrackIncreasingField: departureTime
      steps:
        - to:
            uri: mock:test

위 라우트는 flights.cancellations capped collection에서 소비하고, departureTime을 증가 필드로 쓰며, 커서 재생성 지연 기본값 1000ms를 가져요.

영구 tail tracking (Persistent tail tracking)

표준 tail tracking은 휘발성이라 마지막 값이 메모리에만 유지돼요. 하지만 실제로는 Camel 컨테이너를 재시작해야 할 때가 있죠. 그러면 마지막 값이 사라지고 tailable cursor consumer가 다시 처음부터 소비를 시작해서, 라우트에 중복 레코드를 보낼 가능성이 높아져요.

이 상황을 극복하려면 persistent tail tracking 기능을 켜서 마지막으로 소비한 증가 값을 MongoDB 데이터베이스 안의 특수 컬렉션에 함께 기록하면 돼요. consumer가 다시 초기화되면 마지막으로 기록된 값을 복원하고 아무 일도 없었던 것처럼 계속해요.

마지막 읽기 값은 두 경우에 영속화돼요: 커서가 재생성될 때마다, 그리고 consumer가 종료될 때. 수요가 있다면 추후에 정기 간격(5초마다 플러시) 영속화도 고려할 수 있어요. 이 기능을 요청하려면 Camel JIRA에 티켓을 열어 주세요.

영구 tail tracking 켜기 (Enabling persistent tail tracking)

이 기능을 켜려면 엔드포인트 URI에 최소한 다음 옵션들을 설정하세요:

persistentTailTracking 옵션을 true로
persistentId 옵션을 이 consumer의 고유 식별자로. 그러면 같은 컬렉션을 여러 consumer가 재사용할 수 있어요

추가로 tailTrackDb, tailTrackCollection, tailTrackField 옵션을 설정해서 런타임 정보가 저장될 위치를 커스터마이즈할 수 있어요. 각 옵션의 설명은 이 페이지 위쪽의 엔드포인트 옵션 표를 참고하세요.

예를 들어 다음 라우트는 "flights.cancellations" capped collection에서 소비하고, "departureTime"을 증가 필드로 쓰며, 커서 재생성 지연 기본값 1000ms를 가지고, persistent tail tracking을 켜고, "cancellationsTracker" id로 "flights.camelTailTracking"에 영속화하며, 마지막 처리 값을 "lastTrackingValue" 필드에 저장해요 (camelTailTracking과 lastTrackingValue는 기본값이에요).

Java
XML
YAML

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

<route id="tailableCursorConsumer2" autoStartup="false">
  <from uri="mongodb:myDb?database=flights&amp;collection=cancellations&amp;tailTrackIncreasingField=departureTime&amp;persistentTailTracking=true&amp;persistentId=cancellationsTracker"/>
  <to uri="mock:test"/>
</route>

- route:
    id: tailableCursorConsumer2
    autoStartup: false
    from:
      uri: mongodb:myDb
      parameters:
        database: flights
        collection: cancellations
        tailTrackIncreasingField: departureTime
        persistentTailTracking: true
        persistentId: cancellationsTracker
      steps:
        - to:
            uri: mock:test

아래는 위와 동일한 또 다른 예제인데, persistent tail tracking 런타임 정보를 "trackers.camelTrackers" 컬렉션의 "lastProcessedDepartureTime" 필드에 저장해요:

Java
XML
YAML

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

<route id="tailableCursorConsumer3" autoStartup="false">
  <from uri="mongodb:myDb?database=flights&amp;collection=cancellations&amp;tailTrackIncreasingField=departureTime&amp;persistentTailTracking=true&amp;persistentId=cancellationsTracker&amp;tailTrackDb=trackers&amp;tailTrackCollection=camelTrackers&amp;tailTrackField=lastProcessedDepartureTime"/>
  <to uri="mock:test"/>
</route>

- route:
    id: tailableCursorConsumer3
    autoStartup: false
    from:
      uri: mongodb:myDb
      parameters:
        database: flights
        collection: cancellations
        tailTrackIncreasingField: departureTime
        persistentTailTracking: true
        persistentId: cancellationsTracker
        tailTrackDb: trackers
        tailTrackCollection: camelTrackers
        tailTrackField: lastProcessedDepartureTime
      steps:
        - to:
            uri: mock:test

Change Streams Consumer

Change Streams는 MongoDB oplog를 tailing하는 복잡함과 위험 없이 애플리케이션이 실시간 데이터 변경에 접근하게 해 줘요. 애플리케이션은 change streams로 컬렉션의 모든 데이터 변경을 구독하고 즉시 반응할 수 있어요. change streams는 aggregation framework를 사용하기 때문에, 특정 변경을 필터링하거나 알림을 마음대로 변환할 수도 있어요. exchange 바디에는 어떤 변경이든 전체 문서가 담겨요.

Change Streams Consumer를 설정하려면 consumerType, database, collection과 이벤트를 필터링할 선택적 JSON 프로퍼티 streamFilter를 지정해야 해요. 그 JSON 프로퍼티는 표준 MongoDB $match 애그리게이션이에요. DSL 설정으로 쉽게 지정할 수 있어요:

Java
XML
YAML

from("mongodb:myDb?consumerType=changeStreams&database=flights&collection=tickets&streamFilter={ '$match':{'$or':[{'fullDocument.stringValue': 'specificValue'}]} }")
    .to("mock:test");

<route id="filterConsumer">
  <from uri="mongodb:myDb?consumerType=changeStreams&amp;database=flights&amp;collection=tickets&amp;streamFilter={ '$match':{'$or':[{'fullDocument.stringValue': 'specificValue'}]} }"/>
  <to uri="mock:test"/>
</route>

- route:
    id: filterConsumer
    from:
      uri: mongodb:myDb
      parameters:
        consumerType: changeStreams
        database: flights
        collection: tickets
        streamFilter: "{ '$match':{'$or':[{'fullDocument.stringValue': 'specificValue'}]} }"
      steps:
        - to:
            uri: mock:test

팁 streamFilter 값을 프로퍼티 플레이스홀더로 외부화하면 엔드포인트 URI 파라미터가 더 깔끔해지고 읽기 쉬워져요.

Change Streams 이어서 처리 (Change Streams Resume)

Change Streams consumer는 상호 보완적인 두 가지 방식으로 resume을 지원해요:

resume 토큰을 복원하고 영속화하는 엔드포인트 옵션 (changeStreamTokenRepository, changeStreamToken).
라우트 레벨 resume 처리를 위한 Resume Strategies EIP (resumable()).

route id가 설정되어 있으면 토큰 키는 <routeId>/<collection>으로 계산돼요. route id가 설정되어 있지 않으면 Camel은 엔드포인트 URI로 폴백해요.

resume 토큰은 BSON 문서예요. 직렬화된 JSON 문자열로 영속화되고, 복원될 때 BsonDocument로 파싱돼요.

다음 예제는 엔드포인트 레벨 토큰 영속화를 설정해요:

from("mongodb:myDb?consumerType=changeStreams"
   + "&database=flights"
   + "&collection=tickets"
   + "&changeStreamTokenRepository=#changeStreamTokenRepo")
  .id("resumeChangeStreamConsumer")
  .to("mock:test");

다음 예제는 라우트 레벨 resume strategy 지원을 설정해요:

MongoDbResumeStrategyConfigurationBuilder configurationBuilder =
    MongoDbResumeStrategyConfigurationBuilder.newBuilder()
        .withStateRepository(memoryStateRepository)
        .withResumeCache(new CaffeineCache<>(100));

from("mongodb:myDb?consumerType=changeStreams&database=flights&collection=tickets")
  .id("resumeChangeStreamConsumer")
  .resumable().configuration(configurationBuilder)
  .to("mock:test");

엔드포인트 옵션만 쓰거나, 라우트 레벨 resumable 설정만 쓰거나, 배포 요구사항에 따라 둘 다 쓸 수도 있어요.

타입 변환 (Type conversions)

camel-mongodb 컴포넌트에 포함된 MongoDbBasicConverters 타입 컨버터는 다음 변환을 제공해요:

Name	From type	To type	How?
fromMapToDocument	`Map`	`Document`	`new Document(Map m)` 생성자로 새 `Document`를 만들어요.
fromDocumentToMap	`Document`	`Map`	`Document`는 이미 `Map`을 구현하고 있어요.
fromStringToDocument	`String`	`Document`	`com.mongodb.Document.parse(String s)`를 사용해요.
fromStringToObjectId	`String`	`ObjectId`	`new ObjectId(s)`로 새 `ObjectId`를 만들어요.
fromFileToDocument	`File`	`Document`	내부적으로 `fromInputStreamToDocument`를 사용해요.
fromInputStreamToDocument	`InputStream`	`Document`	inputstream 바이트를 `Document`로 변환해요.
fromStringToList	`String`	`List<Bson>`	`org.bson.codecs.configuration.CodecRegistries`를 사용해 BsonArray로 변환한 뒤 List으로 만들어요.

이 타입 컨버터는 자동으로 발견(auto-discovered)되니, 뭔가를 수동으로 설정할 필요가 없어요.

예제 (Examples)

예제 라우트 (Example route)

다음 Spring XML에 정의된 라우트는 컬렉션에 대해 getDbStats 연산을 실행해요.

지정한 컬렉션의 DB 통계 얻기 (Get DB stats for specified collection)

Java
XML
YAML

from("direct:start")
    .to("mongodb:mongoBean?database={{mongodb.database}}&collection={{mongodb.collection}}&operation=getDbStats")
    .to("direct:result");

<route>
  <from uri="direct:start" />
  <!-- using bean 'mongoBean' defined above -->
  <to uri="mongodb:mongoBean?database={{mongodb.database}}&amp;collection={{mongodb.collection}}&amp;operation=getDbStats" />
  <to uri="direct:result" />
</route>

- route:
    from:
      uri: direct:start
      steps:
        - to:
            uri: mongodb:mongoBean
            parameters:
              database: "{{mongodb.database}}"
              collection: "{{mongodb.collection}}"
              operation: getDbStats
        - to:
            uri: direct:result

더 알아보기 (Learn more)

Apache Camel MongoDB 컴포넌트 문서 — 이 번역문의 출처(옵션·헤더·연산 전체 레퍼런스)
MongoDB 공식 문서 — documents·collections 개념과 capped collection, tailable 커서 배경 지식
MongoDB Java Driver 문서 — Connection String, WriteConcern, Change Streams 동작의 근거

MongoDB 컴포넌트