Change Stream 열기

개요

이 가이드에서는 change stream을 사용하여 데이터베이스의 실시간 변경 사항을 모니터링하는 방법을 배울 수 있습니다. change stream은 애플리케이션이 단일 collection, 데이터베이스 또는 배포에서 데이터 변경 사항을 구독할 수 있도록 하는 MongoDB Server 기능입니다.

애그리게이션 연산자 세트를 지정하여 애플리케이션이 수신하는 데이터를 필터링하고 변환할 수 있습니다. MongoDB deployment v6.0 이상에 연결할 때 변경 전후의 문서 데이터를 포함하도록 이벤트를 구성할 수도 있습니다.

다음 섹션에서 change stream을 열고 구성하는 방법을 알아보세요.

• Change Stream 열기

• change stream에 애그리게이션 연산자 적용

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

Change Stream 열기

change stream을 열어 특정 유형의 데이터 변경 사항을 구독하고 애플리케이션에서 변경 이벤트를 생성할 수 있습니다.

change stream을 열려면 watch() MongoCollectionMongoDatabaseinstance, 또는 인스턴스에서 메서드를 MongoClient 호출합니다.

watch() 메서드를 호출하는 객체에 따라 change stream이 수신하는 이벤트의 범위가 결정됩니다.

MongoCollection 에서 watch() 를 호출하면 change stream이 collection을 모니터링합니다.

MongoDatabase 에서 watch() 를 호출하면 change stream이 해당 데이터베이스의 모든 collection을 모니터링합니다.

MongoClient에서 watch() 를 호출하면 change stream은 연결된 MongoDB의 모든 변경 사항을 모니터링합니다.

MongoCollection<Document> collection = database.getCollection("myColl");
ChangeStreamIterable<Document> changeStream = collection.watch();
changeStream.forEach(event ->
        System.out.println("Received a change: " + event));

Received a change: ChangeStreamDocument{
   operationType=insert,
   resumeToken={"_data": "..."},
   namespace=myDb.myColl,
   ...
}

change stream에 애그리게이션 연산자 적용

집계 파이프라인을 watch() 메서드에 매개변수로 전달하여 change stream이 수신할 이벤트를 지정할 수 있습니다.

사용 중인 MongoDB Server 버전이 지원하는 애그리게이션 연산자를 알아보려면 변경 스트림 출력 수정을 참조하세요.

MongoCollection<Document> collection = database.getCollection("myColl");
List<Bson> pipeline = Arrays.asList(
        Aggregates.match(Filters.in("operationType", Arrays.asList("insert", "update"))));
ChangeStreamIterable<Document> changeStream = collection.watch(pipeline);
changeStream.forEach(event ->
        System.out.println("Received a change to the collection: " + event));

Received a change: ChangeStreamDocument{
   operationType=update,
   resumeToken={"_data": "..."},
   namespace=myDb.myColl,
   ...
}

대규모 change stream 이벤트 분할

MongoDB 7.0부터는 $changeStreamSplitLargeEvent 애그리게이션 단계를 사용하여 16MB를 초과하는 이벤트를 더 작은 조각으로 분할할 수 있습니다.

꼭 필요한 경우에만 $changeStreamSplitLargeEvent 를 사용합니다. 예를 들어, 애플리케이션에 전체 문서 사전 또는 사후 이미지가 필요하고 16MB를 초과하는 이벤트를 생성하는 경우 $changeStreamSplitLargeEvent 를 사용합니다.

$changeStreamSplitLargeEvent 단계는 프래그먼트를 순차적으로 반환합니다. change stream 커서를 사용하여 프래그먼트에 액세스할 수 있습니다. 각 프래그먼트에는 다음 필드를 포함하는 SplitEvent 객체가 포함되어 있습니다.

다음 예에서는 $changeStreamSplitLargeEvent 애그리게이션 단계를 사용하여 대규모 이벤트를 split하여 change stream을 수정합니다.

ChangeStreamIterable<Document> changeStream = collection.watch(
      Arrays.asList(Document.parse("{ $changeStreamSplitLargeEvent: {} }")));

다음 예와 같이 change stream 커서에서 getSplitEvent() 메서드를 호출하여 SplitEvent 에 액세스할 수 있습니다.

MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = changeStream.cursor();
SplitEvent event = cursor.tryNext().getSplitEvent();

$changeStreamSplitLargeEvent 애그리게이션 단계에 대한 자세한 내용은 $changeStreamSplitLargeEvent 서버 설명서를 참조하세요.

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

다음 데이터를 포함하거나 생략하도록 변경 이벤트를 구성할 수 있습니다.

• 사전 이미지 작업 전의 문서의 버전을 나타내는 문서(있는 경우)

• 사후 이미지 작업 후의 문서의 버전을 나타내는 문서(있는 경우)

사전 이미지 또는 사후 이미지가 포함된 변경 스트림 이벤트를 수신하려면 다음 작업을 수행해야 합니다.

• MongoDB deployment에서 collection에 대한 사전 이미지 및 사후 이미지를 활성화합니다.

  팁

  배포에서 사전 및 사후 이미지를 활성화하는 방법에 대한 자세한 내용은 서버 설명서의 Change Streams with Document Pre- and Post-Images(이미지 사전 및 사후 문서화의 Change Streams)를 참조하세요.

  사전 이미지 및 사후 이미지가 활성화된 컬렉션을 생성하도록 드라이버에 지시하는 방법을 알아보려면 사전 이미지 및 사후 이미지가 활성화 된 컬렉션 생성 섹션을 참조하세요.

• 사전 이미지와 사후 이미지 중 하나 또는 둘 다를 검색하도록 change stream을 구성합니다.

  팁

  변경 이벤트에 사전 이미지를 기록하도록 변경 스트림을 구성하려면 사전 이미지 구성 예시를 참조하세요.

  변경 이벤트에 사후 이미지를 기록하도록 변경 스트림을 구성하려면 사후 이미지 구성 예시를 참조하세요.

CreateCollectionOptions collectionOptions = new CreateCollectionOptions();
collectionOptions.changeStreamPreAndPostImagesOptions(new ChangeStreamPreAndPostImagesOptions(true));
database.createCollection("myColl", collectionOptions);

MongoCollection<Document> collection = database.getCollection("myColl");
ChangeStreamIterable<Document> changeStream = collection.watch()
        .fullDocumentBeforeChange(FullDocumentBeforeChange.REQUIRED);
changeStream.forEach(event ->
        System.out.println("Received a change: " + event));

Received a change: ChangeStreamDocument{
   operationType=update,
   resumeToken={"_data": "..."},
   namespace=myDb.myColl,
   destinationNamespace=null,
   fullDocument=null,
   fullDocumentBeforeChange=Document{{_id=..., amount=150, ...}},
   ...
}

MongoCollection<Document> collection = database.getCollection("myColl");
ChangeStreamIterable<Document> changeStream = collection.watch()
        .fullDocument(FullDocument.WHEN_AVAILABLE);
changeStream.forEach(event ->
        System.out.println("Received a change: " + event));

Received a change: ChangeStreamDocument{
   operationType=update,
   resumeToken={"_data": "..."},
   namespace=myDb.myColl,
   destinationNamespace=null,
   fullDocument=Document{{_id=..., color=purple, ...}},
   updatedFields={"color": purple},
   ...
}