db.collection.watch()

배포

db.collection.watch() 는 복제본 세트 및 샤드 클러스터 배포에 사용할 수 있습니다.

• 복제본 세트의 경우 모든 데이터 보유 멤버에 대해 db.collection.watch() 를 실행할 수 있습니다.

• 샤드 클러스터의 db.collection.watch() mongos 경우 인스턴스에서 를 실행해야 합니다.

스토리지 엔진

Wired Tiger 스토리지 엔진 에는 만 db.collection.watch()사용할 수 있습니다.

읽기 고려 majority 지원

MongoDB 4.2부터 변경 스트림은 "majority" 읽기 고려 지원과 무관하게 이용 가능합니다. 즉, 읽기 고려 majority 지원은 변경 스트림 사용 시 활성화(기본값)하거나 비활성화할 수 있습니다.

MongoDB 4.0 이하 버전에서는 읽기 우려 지원을 "majority"를 사용하도록 설정한 경우에만변경 스트림을 사용할 수 있습니다(기본값).

재개 가능성

MongoDB 드라이버 와 달리 mongosh 은 오류 발생 후 변경 스트림 커서를 자동으로 재개하려고 시도하지 않습니다. MongoDB 드라이버는 특정 오류 발생 후 변경 스트림 커서를 자동으로 재개하려고 한 번 시도합니다.

db.collection.watch() 는 oplog에 저장된 정보를 사용하여 변경 이벤트 설명을 생성하고 해당 작업과 관련된 재개 토큰을 생성합니다. 재개 토큰으로 식별되어 resumeAfter 또는 startAfter 옵션에 전달된 작업이 이미 oplog 에서 삭제된 경우 db.collection.watch() 는 변경 스트림을 재개할 수 없습니다.

변경 스트림 재개에 대한 자세한 내용은 변경 스트림 재개를 참조하십시오.

업데이트 작업의 전체 문서 조회

기본적으로 변경 스트림 커서는 업데이트 작업에 대한 특정 필드 변경 사항/델타를 반환합니다. 또한 변경된 문서의 현재 과반수가 커밋된 버전을 조회하고 반환하도록 변경 스트림을 구성할 수 있습니다. 업데이트와 조회 사이에 발생했을 수 있는 다른 쓰기 작업에 따라 반환된 문서는 업데이트 당시의 문서와 크게 다를 수 있습니다.

업데이트 작업 중에 적용된 변경 사항 수와 전체 문서의 크기에 따라 업데이트 작업에 대한 변경 이벤트 문서의 크기가 16MB BSON 문서 제한보다 클 위험이 있습니다. 이런 일이 발생하면 서버는 변경 스트림 커서를 닫고 오류를 반환합니다.

Change Stream 열기

다음 작업으로 data.sensors collection에 대한 변경 스트림 커서를 엽니다.

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

커서를 반복하여 새 이벤트를 확인합니다. 2} 메서드를 cursor.isClosed() 메서드와 함께 cursor.tryNext()를 사용하여 변경 스트림 커서가 닫혀 있고 최신 배치에 개체가 남아 있지 않은 경우에만 루프가 종료되도록 합니다:

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

변경 스트림 출력에 대한 전체 문서는 변경 이벤트를 참조하십시오.

전체 문서 업데이트 조회를 통한 Change Stream

fullDocument 옵션을 "updateLookup"(으)로 설정하여 change stream 커서가 업데이트 change stream 이벤트와 연관된 문서의 가장 최근의 다수 커밋 버전을 조회하도록 지시합니다.

다음 연산은 fullDocument : "updateLookup" 옵션을 사용하여 data.sensors collection에 대해 change stream 커서를 엽니다.

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

커서를 반복하여 새 이벤트를 확인합니다. 2} 메서드를 cursor.isClosed() 메서드와 함께 cursor.tryNext()를 사용하여 변경 스트림 커서가 닫혀 있고 최신 배치에 개체가 남아 있지 않은 경우에만 루프가 종료되도록 합니다:

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

모든 업데이트 작업의 경우 변경 이벤트는 fullDocument 필드의 문서 조회 결과를 반환합니다.

전체 문서 업데이트 출력의 예는 change stream 업데이트 이벤트를 참조하세요.

변경 스트림 출력에 대한 전체 문서는 변경 이벤트를 참조하십시오.

전후 이미지를 포함하는 문서의 Change Streams

MongoDB 6.0부터는 변경 스트림 이벤트를 사용하여 문서의 변경 전후 버전(문서의 전후 이미지)을 출력할 수 있습니다.

• 사전 이미지는 문서가 교체, 업데이트 또는 삭제되기 전의 문서입니다. 삽입된 문서에는 사전 이미지가 없습니다.

• 사후 이미지는 문서가 삽입, 교체, 업데이트된 후의 문서입니다. 삭제된 문서에 대한 사후 이미지가 없습니다.

• db.createCollection(), create 또는 collMod을(를) 사용하는 컬렉션에 대해 changeStreamPreAndPostImages을(를) 활성화하세요.

이미지가 다음과 같은 경우 change stream 이벤트에 사전 및 사후 이미지를 사용할 수 없습니다.

• 문서 업데이트 또는 삭제 작업 시 collection에서 활성화되지 않았습니다.

• expireAfterSeconds에서 전후 이미지 보존 시간 설정 이후에 제거됩니다.

  • 다음 예에서는 전체 cluster에서 expireAfterSeconds 을 100 초로 설정합니다.

    use admin
    db.runCommand( {
       setClusterParameter:
          { changeStreamOptions: { preAndPostImages: { expireAfterSeconds: 100 } } }
    } )

  • 다음 예에서는 특정 collection에서 expireAfterSeconds ~ 100 초를 설정합니다.

    use admin
    db.getSiblingDB("my_collection")
      .sensors.watch({ changeStreamOptions:
        { preAndPostImages: { expireAfterSeconds: 100 } } })

  • 다음 예제에서는 expireAfterSeconds 등 현재 changeStreamOptions 설정을 반환합니다.

    db.adminCommand( { getClusterParameter: "changeStreamOptions" } )

  • expireAfterSeconds를 off로 설정하면 기본 보존 정책이 사용되며, 해당 change stream 이벤트가 oplog에서 제거될 때까지 사전 및 사후 이미지가 보존됩니다.

  • change stream 이벤트가 oplog에서 제거되면 expireAfterSeconds 사전 및 사후 이미지 보존 시간에 관계없이 해당 사전 및 사후 이미지도 삭제됩니다.

추가 고려 사항

• 전후 이미지를 활성화하면 저장 공간이 소모되고 처리 시간이 늘어납니다. 필요한 경우에만 전후 이미지를 활성화하세요.

• 변경 스트림 이벤트 크기를 16메가바이트 미만으로 제한합니다. 이벤트 크기를 제한하려면 다음을 수행하면 됩니다.

  • 문서 크기를 8메가바이트로 제한합니다. updateDescription과 같은 다른 change stream 이벤트 필드가 크지 않은 경우 change stream 출력에서 사전 및 사후 이미지를 동시에 요청할 수 있습니다.

  • updateDescription과 같은 다른 change stream 이벤트 필드가 크지 않은 경우 최대 16메가바이트 문서에 대해 change stream 출력에서 사후 이미지만 요청합니다.

  • 다음과 같은 경우 최대 16메가바이트의 문서에 대해 change stream 출력에서 사전 이미지만 요청합니다.

    • 문서 업데이트가 문서 구조나 내용의 작은 부분에만 영향을 미칩니다. 그리고

    • replace 변경 이벤트를 발생시키지 않습니다. replace 이벤트에는 항상 후이미지가 포함됩니다.

• 사전 이미지를 요청하려면 db.collection.watch() 에서 fullDocumentBeforeChange 를 required 또는 whenAvailable 로 설정합니다. 사후 이미지를 요청하려면 동일한 방법을 사용하여 fullDocument 를 설정합니다.

• 사전 이미지가 config.system.preimages 컬렉션에 기록됩니다.

  • config.system.preimages collection은 커질 수 있습니다. collection 크기를 제한하려면 앞서 표시된 대로 사전 이미지에 대해 expireAfterSeconds 시간을 설정할 수 있습니다.

  • 사전 이미지는 백그라운드 프로세스가 비동기적으로 제거합니다.

db.createCollection(
   "temperatureSensor",
   { changeStreamPreAndPostImages: { enabled: true } }
)

db.temperatureSensor.insertMany( [
   { "_id" : 0, "reading" : 26.1 },
   { "_id" : 1, "reading" : 25.9 },
   { "_id" : 2, "reading" : 24.3 },
   { "_id" : 3, "reading" : 22.4 },
   { "_id" : 4, "reading" : 24.6 }
] )

watchCursorFullDocumentBeforeChange = db.temperatureSensor.watch(
   [],
   { fullDocumentBeforeChange: "whenAvailable" }
)

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

db.temperatureSensor.updateOne(
   { _id: 2 },
   { $set: { reading: 22.1 } }
)

{
   "_id" : {
      "_data" : "82624B21...",
      "_typeBits" : BinData(0,"QA==")
   },
   "operationType" : "update",
   "clusterTime" : Timestamp(1649090957, 1),
   "ns" : {
      "db" : "test",
      "coll" : "temperatureSensor"
   },
   "documentKey" : {
      "_id" : 2
   },
   "updateDescription" : {
      "updatedFields" : {
         "reading" : 22.1
      },
      "removedFields" : [ ],
      "truncatedArrays" : [ ]
   },
   "fullDocumentBeforeChange" : {
      "_id" : 2,
      "reading" : 24.3
   }
}

watchCursorFullDocument = db.temperatureSensor.watch(
   [],
   { fullDocument: "whenAvailable" }
)

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

db.temperatureSensor.updateOne(
   { _id: 1 },
   { $set: { reading: 29.5 } }
)

{
   "_id" : {
      "_data" : "8262474D...",
      "_typeBits" : BinData(0,"QA==")
   },
   "operationType" : "update",
   "clusterTime" : Timestamp(1648840090, 1),
   "fullDocument" : {
      "_id" : 1,
      "reading" : 29.5
   },
   "ns" : {
      "db" : "test",
      "coll" : "temperatureSensor"
   },
   "documentKey" : {
      "_id" : 1
   },
   "updateDescription" : {
      "updatedFields" : {
         "reading" : 29.5
      },
      "removedFields" : [ ],
      "truncatedArrays" : [ ]
   }
}

aggregation pipeline 필터로 스트림 변경하기

다음 작업에서는 insert 이벤트만 필터링하기 위해 aggregation pipeline을 사용하여 data.sensors collection에 대해 변경 스트림 커서를 엽니다.

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

커서를 반복하여 새 이벤트를 확인합니다. 2} 메서드를 cursor.isClosed() 메서드와 함께 cursor.hasNext()를 사용하여 변경 스트림 커서가 닫혀 있고 최신 배치에 개체가 남아 있지 않은 경우에만 루프가 종료되도록 합니다:

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

change stream 커서는 operationType이 insert인 변경 이벤트만 반환합니다. change stream 출력에 대한 전체 문서는 변경 이벤트를 참조하세요.

변경 스트림 재개하기

변경 스트림 커서가 반환하는 모든 문서에는 _id 필드로 재개 토큰이 포함되어 있습니다. 변경 스트림을 재개하려면watch() 의 resumeAfter 또는 startAfter 옵션에 다시 시작하려는 변경 이벤트의 전체 _id 문서를 전달합니다

다음 작업에서는 재개 토큰을 사용하여 data.sensors collection에 대해 변경 스트림 커서를 재개합니다. 여기서는 재개 토큰을 생성한 작업이 클러스터의 oplog에서 롤오프되지 않았다고 가정합니다.

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 }
)

커서를 반복하여 새 이벤트를 확인합니다. 2} 메서드를 cursor.isClosed() 메서드와 함께 cursor.hasNext()를 사용하여 변경 스트림 커서가 닫혀 있고 최신 배치에 개체가 남아 있지 않은 경우에만 루프가 종료되도록 합니다:

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

변경 스트림 재개에 대한 전체 문서는 변경 스트림 재개를 참조하세요.