/ /

변경 스트림

개요

이 가이드 에서는 변경 스트림 을 사용하여 데이터의 실시간 변경 사항을 모니터 하는 방법을 학습 수 있습니다. 변경 스트림 은 애플리케이션 이 컬렉션, 데이터베이스 또는 배포서버 서버의 데이터 변경 사항을 구독 할 수 있도록 하는 MongoDB Server 기능 입니다.

샘플 데이터

이 가이드 의 예제에서는 Atlas 샘플 데이터 세트 의 sample_restaurants.restaurants 컬렉션 을 사용합니다. 무료 MongoDB Atlas cluster 를 생성하고 샘플 데이터 세트를 로드하는 방법을 학습 보려면 시작하기 가이드 를 참조하세요.

중요

프로젝트 리액터 라이브러리

이 가이드 에서는 Project Reactor 라이브러리를 사용하여 Java Reactive Streams 운전자 메서드에서 반환된 Publisher 인스턴스를 사용합니다. 프로젝트 Reactor 라이브러리와 사용 방법에 대해 자세히 학습 Reactor 문서에서 시작하기 참조하세요. 이 가이드에서 Project Reactor 라이브러리 메서드를 사용하는 방법에 대해 자세히 학습하려면 MongoDB에 데이터 쓰기 가이드를 참조하세요.

변경 스트림 열기

변경 스트림 을 열려면 watch() 메서드를 호출합니다. 메서드를 호출하는 인스턴스 에 따라 변경 스트림 이 수신 대기하는 이벤트 범위가 결정됩니다. 다음 클래스의 인스턴스에서 watch() 메서드를 호출할 수 있습니다.

MongoClient: MongoDB deployment의 모든 변경 사항을 모니터링합니다.
MongoDatabase: 데이터베이스에 있는 모든 컬렉션의 변경 사항을 모니터링합니다.
MongoCollection: 컬렉션의 변경 사항을 모니터링합니다.

다음 예시 에서는 restaurants 컬렉션 에서 변경 스트림 을 열고 변경 사항이 발생할 때 출력합니다.

// Opens a change stream and prints the changes as they're received
ChangeStreamPublisher<Document> changeStreamPublisher = restaurants.watch();
Flux.from(changeStreamPublisher)
        .doOnNext(change -> System.out.println("Received change: " + change))
        .blockLast();

변경 사항을 확인하려면 애플리케이션 을 실행 합니다. 그런 다음 별도의 애플리케이션 또는 shell 에서 restaurants 컬렉션 에 대한 쓰기 (write) 작업을 수행합니다. "name" 필드 값이 "Blarney Castle" 인 문서 를 업데이트하면 다음과 같은 변경 스트림 출력이 생성됩니다.

Received change: ChangeStreamDocument{ operationType=update, resumeToken={"_data": "..."},
namespace=sample_restaurants.restaurants, destinationNamespace=null, fullDocument=null,
fullDocumentBeforeChange=null, documentKey={"_id": {"$oid": "..."}}, clusterTime=Timestamp{...},
updateDescription=UpdateDescription{removedFields=[], updatedFields={"cuisine": "Traditional Irish"},
truncatedArrays=[], disambiguatedPaths=null}, txnNumber=null, lsid=null,
splitEvent=null, wallTime=BsonDateTime{value=...}}

변경 스트림 출력 수정

집계 파이프라인 을 watch() 메서드에 매개 변수로 전달하여 변경 스트림 출력을 수정할 수 있습니다. 이 매개변수를 사용하면 지정된 변경 이벤트만 감시할 수 있습니다.

pipeline 매개변수에 다음과 같은 집계 단계를 지정할 수 있습니다.

$addFields
$match
$project
$replaceRoot
$replaceWith
$redact
$set
$unset

다음 예시 에서는 업데이트 작업만 기록 하기 위해 집계 파이프라인 을 변경 스트림 에 전달합니다.

// Creates a change stream pipeline
List<Bson> pipeline = Arrays.asList(
        Aggregates.match(Filters.eq("operationType", "update"))
);
// Opens a change stream and prints the changes as they're received
ChangeStreamPublisher<Document> changeStreamPublisher = restaurants.watch(pipeline);
Flux.from(changeStreamPublisher)
        .doOnNext(change -> System.out.println("Received change: " + change))
        .blockLast();

변경 스트림 출력 수정에 대해 자세히 알아보려면 MongoDB Server 매뉴얼의 변경 스트림 출력 수정 섹션을 참조하세요.

`watch()` 동작 수정

변경 스트림 작업을 구성하는 데 사용할 수 있는 옵션을 나타내는 watch() 메서드에 메서드를 연결할 수 있습니다. 옵션을 지정하지 않으면 운전자 는 작업을 사용자 지정하지 않습니다.

다음 표에서는 동작을 사용자 지정하기 위해 watch() 에 연결할 수 있는 메서드에 대해 설명합니다.

옵션	설명
`fullDocument()`	문서의 변경 사항만 표시하지 않고 변경 후 전체 문서를 표시할지 여부를 지정합니다. 이 옵션에 대해 자세히 알아보려면 사전 이미지 및 사후 이미지 포함 을 참조하세요.
`fullDocumentBeforeChange()`	문서의 변경 사항만 표시하는 대신 변경 전의 전체 문서를 표시할지 여부를 지정합니다. 이 옵션에 대해 자세히 알아보려면 사전 이미지 및 사후 이미지 포함 을 참조하세요.
`resumeAfter()`	`watch()` 재개 토큰에 지정된 작업 후 변경 사항 반환을 재개하도록 에 지시합니다. 각 변경 스트림 이벤트 문서 필드 로 재개 토큰이 포함되어 있습니다.`_id` `_id` 이후에 재개하려는 작업을 나타내는 변경 이벤트 문서 의 전체 필드 전달합니다. `resumeAfter()` 은 및 와 상호 `startAfter()` `startAtOperationTime()`배타적입니다.
`startAfter()`	`watch()` `_id` 재개 `_id` 토큰에 지정된 작업 후 새 변경 스트림 시작하도록 에 지시합니다. 무효화 이벤트 후 알림 다시 시작할 `startAfter()` 수 있습니다. 각 변경 스트림 이벤트 문서 필드 로 재개 토큰이 포함되어 있습니다. 이후에 재개하려는 작업을 나타내는 변경 이벤트 문서 의 전체 필드 전달합니다.`resumeAfter()` 은 및 와 상호 `startAtOperationTime()` 배타적입니다.
`startAtOperationTime()`	`watch()` 지정된 타임스탬프 이후에 발생하는 이벤트만 반환하도록 `startAtOperationTime()` 에 지시합니다.`resumeAfter()` 은 및 와 상호 `startAfter()` 배타적입니다.
`maxAwaitTime()`	서버 빈 배치 를 반환하기 전에 변경 스트림 커서 에 보고하기 위해 새 데이터 변경 사항을 기다리는 최대 시간(밀리초)을 지정합니다. 기본값은 1000 밀리초입니다.
`showExpandedEvents()`	MongoDB Server v6.0부터는 변경 스트림은 `createIndexes` 및 `dropIndexes` 이벤트와 같은 데이터 정의 언어(DDL) 이벤트에 대한 변경 알림 지원 . 변경 스트림 에 확장 이벤트를 포함하려면 이 메서드를 호출하고 `true` 값을 전달합니다.
`batchSize()`	MongoDB cluster 의 각 응답 배치 에서 반환할 최대 변경 이벤트 수를 지정합니다. 기본값 으로 운전자 이 값을 `Long.MAX_VALUE`로 설정합니다. `0` 의 `batchSize` 은 커서 설정되지만 첫 번째 배치 에서는 문서가 반환되지 않음을 의미합니다.
`collation()`	변경 스트림 커서 에 사용할 데이터 정렬을 지정합니다.
`comment()`	작업에 주석을 첨부합니다.

사전 이미지 및 사후 이미지 포함하기

중요

배포에서 MongoDB v6.0 이상을 사용하는 경우에만 컬렉션에서 사전 이미지 및 사후 이미지를 활성화할 수 있습니다.

기본값 으로 컬렉션 에서 작업을 수행할 때 해당 변경 이벤트 에는 해당 작업에 의해 수정된 필드의 델타만 포함됩니다. 변경 전후의 전체 문서 를 보려면 fullDocumentBeforeChange() 또는 fullDocument() 메서드를 watch() 메서드에 연결합니다.

사전 이미지 는 변경 전의 문서 전체 버전입니다. 변경 스트림 이벤트 에 사전 이미지를 포함하려면 다음 값 중 하나를 fullDocumentBeforeChange() 메서드에 전달합니다.

FullDocumentBeforeChange.WHEN_AVAILABLE: 변경 이벤트에는 사전 이미지를 사용할 수 있는 경우에만 변경 이벤트에 대해 수정된 문서의 사전 이미지가 포함됩니다.
FullDocumentBeforeChange.REQUIRED: 변경 이벤트에는 변경 이벤트에 대한 수정된 문서의 사전 이미지가 포함됩니다. 사전 이미지를 사용할 수 없는 경우 드라이버에서 오류가 발생합니다.

사후 이미지 는 변경 후 문서 의 전체 버전입니다. 변경 스트림 이벤트 에 사후 이미지를 포함하려면 다음 값 중 하나를 fullDocument() 메서드에 전달합니다.

FullDocument.UPDATE_LOOKUP: 변경 이벤트에는 변경 후 일정 시간 이후의 변경된 문서 전체의 복사본이 포함됩니다.
FullDocument.WHEN_AVAILABLE: 변경 이벤트에는 사후 이미지를 사용할 수 있는 경우에만 변경 이벤트에 대해 수정된 문서의 사후 이미지가 포함됩니다.
FullDocument.REQUIRED: 변경 이벤트에는 변경 이벤트에 대한 수정된 문서의 사후 이미지가 포함됩니다. 사후 이미지를 사용할 수 없는 경우 드라이버에서 오류가 발생합니다.

다음 예시 에서는 컬렉션 에서 변경 스트림 을 열고 fullDocument() 메서드를 watch() 메서드에 연결하여 업데이트된 문서의 사후 이미지를 포함합니다.

// Creates a change stream pipeline
List<Bson> pipeline = Arrays.asList(
        Aggregates.match(Filters.eq("operationType", "update"))
);
// Opens a change stream and prints the changes as they're received including the full
// document after the update
ChangeStreamPublisher<Document> changeStreamPublisher = restaurants.watch(pipeline)
        .fullDocument(FullDocument.UPDATE_LOOKUP);
Flux.from(changeStreamPublisher)
        .doOnNext(change -> System.out.println("Received change: " + change))
        .blockLast();

사전 이미지 및 사후 이미지에 대해 자세히 알아보려면 Change Streams 매뉴얼에서 문서 사전 및 사후 이미지로 MongoDB Server 을 참조하세요.

추가 정보

변경 스트림에 대해 자세히 알아보려면 Change Streams 매뉴얼의 MongoDB Server 을 참조하세요.

API 문서

이 가이드에서 사용되는 메서드 또는 유형에 대해 자세히 알아보려면 다음 API 설명서를 참조하세요.

돌아가기

모니터링

인증