데이터 변경 사항 모니터링

개요

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

스칼라 운전자 사용하는 경우 watch() 메서드를 호출하여 ChangeStreamObservable의 인스턴스 반환할 수 있습니다. 그런 다음 ChangeStreamObservable 인스턴스 를 구독 업데이트, 삽입 및 삭제와 같은 새로운 데이터 변경 사항을 볼 수 있습니다.

val database: MongoDatabase = client.getDatabase("sample_restaurants")
val collection: MongoCollection[Document] = database.getCollection("restaurants")

case class LatchedObserver() extends Observer[ChangeStreamDocument[Document]] {
  val latch = new CountDownLatch(1)
  override def onSubscribe(subscription: Subscription): Unit = subscription.request(Long.MaxValue) // Request data
  override def onNext(changeDocument: ChangeStreamDocument[Document]): Unit = println(changeDocument)
  override def onError(throwable: Throwable): Unit = {
      println(s"Error: '$throwable")
      latch.countDown()
  }
  override def onComplete(): Unit = latch.countDown()
  def await(): Unit = latch.await()
}

변경 스트림 열기

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

• MongoClient: 시스템 컬렉션 adminlocal또는, 및 데이터베이스의 컬렉션을 제외한 배포서버 내 모든 데이터베이스의 모든 컬렉션에 대한 변경 사항을 모니터링합니다.config

• MongoDatabase: 하나의 데이터베이스 에 있는 모든 컬렉션의 변경 사항을 모니터링합니다.

• MongoCollection: 하나의 컬렉션 에 대한 변경 사항을 모니터링합니다.

다음 예시 watch() 메서드를 호출하여 restaurants 컬렉션 에서 변경 스트림 엽니다. 이 코드는 변경 사항이 발생하면 이를 수신하고 출력하는 LatchedObserver 인스턴스 만듭니다.

val observer = LatchedObserver()
collection.watch().subscribe(observer)
observer.await()

변경 사항을 확인하려면 앞의 코드를 실행 . 그런 다음 별도의 셸 에서 다음 코드를 실행 name 필드 값이 "Blarney Castle"인 문서 업데이트 .

val filter = equal("name", "Blarney Castle")
val update = set("cuisine", "American")
collection.updateOne(filter, update)
          .subscribe((res: UpdateResult) => println(res),
                    (e: Throwable) => println(s"There was an error: $e"))

컬렉션 업데이트 위해 앞의 코드를 실행 변경 스트림 애플리케이션 변경 사항이 발생하는 즉시 출력합니다. 인쇄된 변경 이벤트 다음 출력과 유사합니다.

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

변경 스트림 출력 수정

변경 스트림 출력을 수정하려면 파이프라인 단계 목록을 watch() 메서드에 매개 변수로 전달하면 됩니다. 목록에 다음 단계를 포함할 수 있습니다.

• $addFields 또는 $set: 문서에 새 필드를 추가합니다.

• $match: 문서를 필터링합니다.

• $project: 문서 필드의 하위 집합을 프로젝션합니다.

• $replaceWith 또는 $replaceRoot: 입력 문서 를 지정된 문서 로 바꿉니다.

• $redact: 문서의 내용을 제한합니다.

• $unset: 문서에서 필드를 제거합니다.

스칼라 운전자 이전 파이프라인 단계를 빌드하기 위한 헬퍼 메서드가 포함된 Aggregates 클래스를 제공합니다.

다음 예시 Aggregates.filter() 메서드를 사용하여 $match 단계를 빌드 파이프라인 만듭니다. 그런 다음 코드는 이 파이프라인 watch() 메서드에 전달하고 업데이트 작업이 발생할 때만 이벤트를 출력하도록 watch() 에 지시합니다.

val observer = LatchedObserver()
collection.watch(Seq(Aggregates.filter(Filters.in("operationType", "update"))))
observer.await()

watch() 동작 수정

ChangeStreamObservable 클래스에서 제공하는 메서드를 연결하여 watch() 메서드의 동작을 수정할 수 있습니다. 다음 표에서는 이러한 메서드 중 일부에 대해 설명합니다.

watch() 옵션의 전체 목록은 API 문서에서 ChangeStreamObservable을 참조하세요.

val observer = LatchedObserver()
collection.watch()
          .fullDocument(FullDocument.UPDATE_LOOKUP)
          .subscribe(observer)
observer.await()

ChangeStreamDocument{ operationType=update, resumeToken={"_data": "..."},
namespace=sample_restaurants.restaurants, destinationNamespace=null,
fullDocument=Iterable((_id,BsonObjectId{...}), (address,{"building": "202-24",
"coord": [-73.9250442, 40.5595462], "street": "Rockaway Point Boulevard",
"zipcode": "11697"}), (borough,BsonString{value='Queens'}),
(cuisine,BsonString{value='Irish'}), (grades,BsonArray{values=[...]}),
(name,BsonString{value='Blarney Castle'}), (restaurant_id,BsonString{...}),
(blank,BsonString{value='Irish'})), fullDocumentBeforeChange=null,
documentKey={"_id": {"$oid": "..."}}, clusterTime=Timestamp{...}, updateDescription=
UpdateDescription{removedFields=[], updatedFields={"cuisine": "Irish"},
truncatedArrays=[], disambiguatedPaths=null}, txnNumber=null, lsid=null,
splitEvent=null, wallTime=BsonDateTime{...}}

추가 정보

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