db.collection.watch()

정의

db.collection.watch( pipeline, options )

중요

Mongo쉬 방법

이 페이지에서는 mongosh 메서드에 대해 설명합니다. 이는 데이터베이스 명령 또는 Node.js와 같은 언어별 드라이버에 대한 설명서가 아닙니다.

데이터베이스 명령의 경우 $changeStream 집계 단계가 있는 aggregate 명령을 참조하세요.

MongoDB API 드라이버의 경우 언어별 MongoDB 드라이버 설명서를 참조하세요.

복제본 세트 및 샤딩된 클러스터 전용

컬렉션에서 변경 스트림 커서를 엽니다.

Parameter	유형	설명
pipeline	배열	선택 사항입니다. 다음 집계 단계 중 하나 이상으로 구성된 집계 파이프라인: $addFields $match $project $replaceRoot $replaceWith $redact $set $unset 변경 이벤트 출력을 필터링/수정할 파이프라인을 지정합니다. MongoDB 4.2부터 변경 스트림 집계 파이프라인이 이벤트의 _id 필드를 수정하면 변경 스트림이 예외를 스로우합니다.
options	문서	선택 사항. watch()의동작을 수정하는 추가 옵션입니다.

options 문서에는 다음과 같은 필드 및 값이 포함될 수 있습니다.

필드	유형	설명
resumeAfter	문서	선택 사항. watch() 재개 토큰에 지정된 작업 이후부터 알림 재개를 시도하도록 에 지시합니다. 각 변경 스트림 이벤트 문서에는 _id 필드로 이력서 토큰이 포함되어 있습니다. 이후에 재개하려는 작업을 나타내는 변경 이벤트 문서의 전체 _id 필드를 전달하십시오. resumeAfter 의 경우 startAfter 및 startAtOperationTime과 상호 배타적입니다.
startAfter	문서	선택 사항. watch() 재개 토큰에 지정된 작업 후 새 변경 스트림 시작을 시도하도록 에 지시합니다. 무효화 이벤트 후 알림 다시 시작할 수 있습니다. 각 변경 스트림 이벤트 문서에는 _id 필드로 이력서 토큰이 포함되어 있습니다. 이후에 재개하려는 작업을 나타내는 변경 이벤트 문서의 전체 _id 필드를 전달하십시오. startAfter 의 경우 resumeAfter 및 startAtOperationTime과 상호 배타적입니다.
fullDocument	문자열	선택 사항. watch() 기본값 으로 는 업데이트된 전체 문서 대신 업데이트 작업으로 수정된 필드의 델타를 반환합니다. fullDocument "updateLookup" watch() 를 로 설정하여 가 업데이트된 문서 의 최신 과반수 커밋 버전을 찾도록 지시합니다. watch() fullDocument 은 updateDescription 델타 외에 문서 조회가 포함된 필드 반환합니다.
batchSize	int	선택 사항입니다. MongoDB cluster의 각 응답 배치에서 반환할 최대 변경 이벤트 수를 지정합니다. cursor.batchSize()와 동일한 기능을 갖습니다.
maxAwaitTimeMS	int	선택 사항입니다. 서버가 빈 배치를 반환하기 전에 변경 스트림 커서에 보고할 새 데이터 변경 내용을 기다리는 최대 시간(밀리초)입니다. 기본값은 1000밀리초입니다.
collation	문서	선택 사항. 문서 를 전달하여 change stream 커서에 대한 데이터 정렬 을 지정합니다. 생략할 경우 기본값은 simple 바이너리 비교입니다.
startAtOperationTime	타임스탬프	선택 사항입니다. 변경 스트림의 시작 지점입니다. 지정된 시작점이 과거인 경우 oplog의 시간 범위 내에 있어야 합니다. oplog의 시간 범위를 확인하려면 rs.printReplicationInfo()를 참조하십시오. startAtOperationTime 의 경우 resumeAfter 및 startAfter과 상호 배타적입니다.

반환합니다:	MongoDB 배포에 대한 연결이 열려있고 동시에 컬렉션이 존재하는 한 커서가 열려 있는 상태로 유지됩니다. 변경 이벤트 문서의 예는 변경 이벤트 문서를 참조하세요.

팁

db.watch() 개인정보 정책에 Mongo.watch()

호환성

이 메서드는 다음 환경에서 호스팅되는 배포에서 사용할 수 있습니다.

• MongoDB Atlas: 클라우드에서의 MongoDB 배포를 위한 완전 관리형 서비스

• MongoDB Enterprise: MongoDB의 구독 기반 자체 관리 버전

• MongoDB Community: MongoDB의 소스 사용 가능 무료 자체 관리 버전

가용성

행동

• db.collection.watch() 대부분의 데이터 보유 멤버에게 지속된 데이터 변경 사항에 대해서만 알립니다.

• 변경 스트림 커서는 다음 중 하나가 발생할 때까지 열려 있습니다.

  • 커서가 명시적으로 닫힙니다.

  • 무효화 이벤트가 발생합니다. 예를 들어 컬렉션 제거 또는 이름 바꾸기가 있습니다.

  • MongoDB 배포에 대한 연결이 닫히거나 시간 초과됩니다. 자세한 내용은 Cursor Behaviors를 참조하십시오.

  • 배포가 샤딩된 클러스터인 경우 분할 제거로 인해 열린 변경 스트림 커서가 닫힐 수 있으며 닫힌 변경 스트림 커서가 완전히 재개되지 않을 수 있습니다.

액세스 제어

액세스 제어와 함께 실행하는 경우, 사용자는 컬렉션 리소스에 대해 find 및 changeStream 권한 조치를 가지고 있어야 합니다. 즉, 사용자의 역할이 다음 권한을 부여해야 합니다.

{ resource: { db: <dbname>, collection: <collection> }, actions: [ "find", "changeStream" ] }

기본 제공 read 역할은 적절한 권한을 제공합니다.

커서 반복

MongoDB는 커서를 반복하는 다양한 방법을 제공합니다.

cursor.hasNext() 메서드는 다음 이벤트를 차단하고 기다립니다. watchCursor 커서를 모니터링하고 이벤트를 반복하려면 hasNext()를 다음과 같이 사용합니다.

while (!watchCursor.isClosed()) {
   if (watchCursor.hasNext()) {
     firstChange = watchCursor.next();
     break;
   }
}

cursor.tryNext() 메서드는 비차단입니다. watchCursor 커서를 모니터링하고 이벤트를 반복하려면 tryNext()를 다음과 같이 사용합니다.

while (!watchCursor.isClosed()) {
  let next = watchCursor.tryNext()
  while (next !== null) {
    printjson(next);
    next = watchCursor.tryNext()
  }
}

예시

watchCursor = db.getSiblingDB("data").sensors.watch()

while (!watchCursor.isClosed()) {
  let next = watchCursor.tryNext()
  while (next !== null) {
    printjson(next);
    next = watchCursor.tryNext()
  }
}

watchCursor = db.getSiblingDB("data").sensors.watch(
   [],
   { fullDocument : "updateLookup" }
)

while (!watchCursor.isClosed()) {
  let next = watchCursor.tryNext()
  while (next !== null) {
    printjson(next);
    next = watchCursor.tryNext()
  }
}

watchCursor = db.getSiblingDB("data").sensors.watch(
   [
      { $match : {"operationType" : "insert" } }
   ]
)

while (!watchCursor.isClosed()){
   if (watchCursor.hasNext()){
      printjson(watchCursor.next());
   }
}

let watchCursor = db.getSiblingDB("data").sensors.watch();
let firstChange;
while (!watchCursor.isClosed()) {
   if (watchCursor.hasNext()) {
     firstChange = watchCursor.next();
     break;
   }
}
watchCursor.close();
let resumeToken = firstChange._id;
resumedWatchCursor = db.getSiblingDB("data").sensors.watch(
[],
   { resumeAfter : resumeToken }
)

while (!resumedWatchCursor.isClosed()){
   if (resumedWatchCursor.hasNext()){
      print(resumedWatchCursor.next());
   }
}