Change Streams变更流¶

Availability可用性¶

Change streams are available for replica sets and sharded clusters:更改流可用于副本集和分片群集：

• Storage Engine.存储引擎。

  The replica sets and sharded clusters must use the WiredTiger storage engine. 副本集和分片集群必须使用WiredTiger存储引擎。Change streams can also be used on deployments that employ MongoDB’s encryption-at-rest feature.变更流还可以用于使用MongoDB的静态加密功能的部署。

• Replica Set Protocol Version.副本集协议版本。

  The replica sets and sharded clusters must use replica set protocol version 1 (pv1).副本集和分片集群必须使用副本集协议版本1(pv1)。

• Read Concern “majority” Enablement.阅读关注“多数”启用。

  Starting in MongoDB 4.2, change streams are available regardless of the "majority" read concern support; that is, read concern majority support can be either enabled (default) or disabled to use change streams.从MongoDB 4.2开始，无论支持"majority"读关注点，都可以使用更改流；也就是说，可以启用（默认）或禁用读关注majority支持来使用更改流。

  In MongoDB 4.0 and earlier, change streams are available only if "majority" read concern support is enabled (default).在MongoDB 4.0及更早版本中，只有在启用了"majority"读关注点支持（默认）的情况下，更改流才可用。

Watch Collection/Database/Deployment监视收集/数据库/部署¶

You can open change streams against:您可以针对以下内容打开变更流：

Open A Change Stream打开变更流¶

cursor = db.inventory.watch()
document = next(cursor)

MongoCursor<ChangeStreamDocument<Document>> cursor = inventory.watch().iterator();
ChangeStreamDocument<Document> next = cursor.next();

const collection = db.collection('inventory');
const changeStream = collection.watch();
changeStream.on('change', next => {
  // process next document
});

const changeStreamIterator = collection.watch();
const next = await changeStreamIterator.next();

$changeStream = $db->inventory->watch();
$changeStream->rewind();

$firstChange = $changeStream->current();

$changeStream->next();

$secondChange = $changeStream->current();

cursor = db.inventory.watch()
document = await cursor.next()

mongoc_collection_t *collection;
bson_t *pipeline = bson_new ();
bson_t opts = BSON_INITIALIZER;
mongoc_change_stream_t *stream;
const bson_t *change;
const bson_t *resume_token;
bson_error_t error;

collection = mongoc_database_get_collection (db, "inventory");
stream = mongoc_collection_watch (collection, pipeline, NULL /* opts */);
mongoc_change_stream_next (stream, &change);
if (mongoc_change_stream_error_document (stream, &error, NULL)) {
   MONGOC_ERROR ("%s\n", error.message);
}

mongoc_change_stream_destroy (stream);

var cursor = inventory.Watch();
while (cursor.MoveNext() && cursor.Current.Count() == 0) { } // keep calling MoveNext until we've read the first batch
var next = cursor.Current.First();
cursor.Dispose();

cursor = inventory.watch.to_enum
next_change = cursor.next

cs, err := coll.Watch(ctx, mongo.Pipeline{})
require.NoError(t, err)
defer cs.Close(ctx)

ok := cs.Next(ctx)
next := cs.Current

let inventory = db.collection("inventory")
let changeStream = try inventory.watch()
let next = changeStream.next()

let inventory = db.collection("inventory")

// Option 1: retrieve next document via next()
let next = inventory.watch().flatMap { cursor in
    cursor.next()
}

// Option 2: register a callback to execute for each document
let result = inventory.watch().flatMap { cursor in
    cursor.forEach { event in
        // process event
        print(event)
    }
}

Modify Change Stream Output修改变更流输出¶

pipeline = [
    {'$match': {'fullDocument.username': 'alice'}},
    {'$addFields': {'newField': 'this is an added field!'}}
]
cursor = db.inventory.watch(pipeline=pipeline)
document = next(cursor)

MongoClient mongoClient = new MongoClient( new MongoClientURI("mongodb://host1:port1,host2:port2..."));

// Select the MongoDB database and collection to open the change stream against

MongoDatabase db = mongoClient.getDatabase("myTargetDatabase");

MongoCollection<Document> collection = db.getCollection("myTargetCollection");

// Create $match pipeline stage.
List<Bson> pipeline = singletonList(Aggregates.match(Filters.or(
    Document.parse("{'fullDocument.username': 'alice'}"),
    Filters.in("operationType", asList("delete")))));

// Create the change stream cursor, passing the pipeline to the
// collection.watch() method

MongoCursor<Document> cursor = collection.watch(pipeline).iterator();

const pipeline = [
  { $match: { 'fullDocument.username': 'alice' } },
  { $addFields: { newField: 'this is an added field!' } }
];

const collection = db.collection('inventory');
const changeStream = collection.watch(pipeline);
changeStream.on('change', next => {
  // process next document
});

const changeStreamIterator = collection.watch(pipeline);
const next = await changeStreamIterator.next();

$pipeline = [
    ['$match' => ['fullDocument.username' => 'alice']],
    ['$addFields' => ['newField' => 'this is an added field!']],
];
$changeStream = $db->inventory->watch($pipeline);
$changeStream->rewind();

$firstChange = $changeStream->current();

$changeStream->next();

$secondChange = $changeStream->current();

pipeline = [
    {'$match': {'fullDocument.username': 'alice'}},
    {'$addFields': {'newField': 'this is an added field!'}}
]
cursor = db.inventory.watch(pipeline=pipeline)
document = await cursor.next()

pipeline = BCON_NEW ("pipeline",
                     "[",
                     "{",
                     "$match",
                     "{",
                     "fullDocument.username",
                     BCON_UTF8 ("alice"),
                     "}",
                     "}",
                     "{",
                     "$addFields",
                     "{",
                     "newField",
                     BCON_UTF8 ("this is an added field!"),
                     "}",
                     "}",
                     "]");

stream = mongoc_collection_watch (collection, pipeline, &opts);
mongoc_change_stream_next (stream, &change);
if (mongoc_change_stream_error_document (stream, &error, NULL)) {
   MONGOC_ERROR ("%s\n", error.message);
}

mongoc_change_stream_destroy (stream);

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<BsonDocument>>()
    .Match(change =>
        change.FullDocument["username"] == "alice" ||
        change.OperationType == ChangeStreamOperationType.Delete)
    .AppendStage<ChangeStreamDocument<BsonDocument>, ChangeStreamDocument<BsonDocument>, BsonDocument>(
        "{ $addFields : { newField : 'this is an added field!' } }");

var collection = database.GetCollection<BsonDocument>("inventory");
using (var cursor = collection.Watch(pipeline))
{
    while (cursor.MoveNext() && cursor.Current.Count() == 0) { } // keep calling MoveNext until we've read the first batch
    var next = cursor.Current.First();
}

pipeline := mongo.Pipeline{bson.D{{"$match", bson.D{{"$or",
	bson.A{
		bson.D{{"fullDocument.username", "alice"}},
		bson.D{{"operationType", "delete"}}}}},
}}}
cs, err := coll.Watch(ctx, pipeline)
require.NoError(t, err)
defer cs.Close(ctx)

ok := cs.Next(ctx)
next := cs.Current

let pipeline: [BSONDocument] = [
    ["$match": ["fullDocument.username": "alice"]],
    ["$addFields": ["newField": "this is an added field!"]]
]
let inventory = db.collection("inventory")
let changeStream = try inventory.watch(pipeline, withEventType: BSONDocument.self)
let next = changeStream.next()

let pipeline: [BSONDocument] = [
    ["$match": ["fullDocument.username": "alice"]],
    ["$addFields": ["newField": "this is an added field!"]]
]
let inventory = db.collection("inventory")

// Option 1: use next() to iterate
let next = inventory.watch(pipeline, withEventType: BSONDocument.self).flatMap { changeStream in
    changeStream.next()
}

// Option 2: register a callback to execute for each document
let result = inventory.watch(pipeline, withEventType: BSONDocument.self).flatMap { changeStream in
    changeStream.forEach { event in
        // process event
        print(event)
    }
}

Lookup Full Document for Update Operations查找更新操作的完整文档¶

cursor = db.inventory.watch(full_document='updateLookup')
document = next(cursor)

cursor = inventory.watch().fullDocument(FullDocument.UPDATE_LOOKUP).iterator();
next = cursor.next();

const collection = db.collection('inventory');
const changeStream = collection.watch({ fullDocument: 'updateLookup' });
changeStream.on('change', next => {
  // process next document
});

const changeStreamIterator = collection.watch({ fullDocument: 'updateLookup' });
const next = await changeStreamIterator.next();

$changeStream = $db->inventory->watch([], ['fullDocument' => \MongoDB\Operation\Watch::FULL_DOCUMENT_UPDATE_LOOKUP]);
$changeStream->rewind();

$firstChange = $changeStream->current();

$changeStream->next();

$secondChange = $changeStream->current();

cursor = db.inventory.watch(full_document='updateLookup')
document = await cursor.next()

BSON_APPEND_UTF8 (&opts, "fullDocument", "updateLookup");
stream = mongoc_collection_watch (collection, pipeline, &opts);
mongoc_change_stream_next (stream, &change);
if (mongoc_change_stream_error_document (stream, &error, NULL)) {
   MONGOC_ERROR ("%s\n", error.message);
}

mongoc_change_stream_destroy (stream);

var options = new ChangeStreamOptions { FullDocument = ChangeStreamFullDocumentOption.UpdateLookup };
var cursor = inventory.Watch(options);
while (cursor.MoveNext() && cursor.Current.Count() == 0) { } // keep calling MoveNext until we've read the first batch
var next = cursor.Current.First();
cursor.Dispose();

cursor = inventory.watch([], full_document: 'updateLookup').to_enum
next_change = cursor.next

cs, err := coll.Watch(ctx, mongo.Pipeline{}, options.ChangeStream().SetFullDocument(options.UpdateLookup))
require.NoError(t, err)
defer cs.Close(ctx)

ok := cs.Next(ctx)
next := cs.Current

let inventory = db.collection("inventory")
let changeStream = try inventory.watch(options: ChangeStreamOptions(fullDocument: .updateLookup))
let next = changeStream.next()

let inventory = db.collection("inventory")

// Option 1: use next() to iterate
let next = inventory.watch(options: ChangeStreamOptions(fullDocument: .updateLookup))
    .flatMap { changeStream in
        changeStream.next()
    }

// Option 2: register a callback to execute for each document
let result = inventory.watch(options: ChangeStreamOptions(fullDocument: .updateLookup))
    .flatMap { changeStream in
        changeStream.forEach { event in
            // process event
            print(event)
        }
    }

Resume a Change Stream继续改变流程¶

Change streams are resumable by specifying a resume token to either resumeAfter or startAfter when opening the cursor.通过在打开游标时为resumeAfter或startAfter指定恢复令牌，可以恢复更改流。

resume_token = cursor.resume_token
cursor = db.inventory.watch(resume_after=resume_token)
document = next(cursor)

BsonDocument resumeToken = next.getResumeToken();
cursor = inventory.watch().resumeAfter(resumeToken).iterator();
next = cursor.next();

const collection = db.collection('inventory');
const changeStream = collection.watch();

let newChangeStream;
changeStream.once('change', next => {
  const resumeToken = changeStream.resumeToken;
  changeStream.close();

  newChangeStream = collection.watch({ resumeAfter: resumeToken });
  newChangeStream.on('change', next => {
    processChange(next);
  });
});

$resumeToken = $changeStream->getResumeToken();

if ($resumeToken === null) {
    throw new \Exception('Resume token was not found');
}

$changeStream = $db->inventory->watch([], ['resumeAfter' => $resumeToken]);
$changeStream->rewind();

$firstChange = $changeStream->current();

resume_token = cursor.resume_token
cursor = db.inventory.watch(resume_after=resume_token)
document = await cursor.next()

stream = mongoc_collection_watch (collection, pipeline, NULL);
if (mongoc_change_stream_next (stream, &change)) {
   resume_token = mongoc_change_stream_get_resume_token (stream);
   BSON_APPEND_DOCUMENT (&opts, "resumeAfter", resume_token);

   mongoc_change_stream_destroy (stream);
   stream = mongoc_collection_watch (collection, pipeline, &opts);
   mongoc_change_stream_next (stream, &change);
   mongoc_change_stream_destroy (stream);
} else {
   if (mongoc_change_stream_error_document (stream, &error, NULL)) {
      MONGOC_ERROR ("%s\n", error.message);
   }

   mongoc_change_stream_destroy (stream);
}

  var resumeToken = previousCursor.GetResumeToken();
  var options = new ChangeStreamOptions { ResumeAfter = resumeToken };
  var cursor = inventory.Watch(options);
  cursor.MoveNext();
  var next = cursor.Current.First();
  cursor.Dispose();

  change_stream = inventory.watch
  cursor = change_stream.to_enum
  next_change = cursor.next
  resume_token = change_stream.resume_token

  new_cursor = inventory.watch([], resume_after: resume_token).to_enum
  resumed_change = new_cursor.next

resumeToken := original.ResumeToken()

cs, err := coll.Watch(ctx, mongo.Pipeline{}, options.ChangeStream().SetResumeAfter(resumeToken))
require.NoError(t, err)
defer cs.Close(ctx)

ok = cs.Next(ctx)
result := cs.Current

let inventory = db.collection("inventory")
let changeStream = try inventory.watch(options: ChangeStreamOptions(fullDocument: .updateLookup))
let next = changeStream.next()

let resumeToken = changeStream.resumeToken
let resumedChangeStream = try inventory.watch(options: ChangeStreamOptions(resumeAfter: resumeToken))
let nextAfterResume = resumedChangeStream.next()

let inventory = db.collection("inventory")

inventory.watch(options: ChangeStreamOptions(fullDocument: .updateLookup))
    .flatMap { changeStream in
        changeStream.next().map { _ in
            changeStream.resumeToken
        }.always { _ in
            _ = changeStream.kill()
        }
    }.flatMap { resumeToken in
        inventory.watch(options: ChangeStreamOptions(resumeAfter: resumeToken)).flatMap { newStream in
            newStream.forEach { event in
                // process event
                print(event)
            }
        }
    }

{
   "_data" : <BinData|hex string>
}

Use Cases用例¶

Change streams can benefit architectures with reliant business systems, informing downstream systems once data changes are durable. 变更流可以使具有可靠业务系统的体系结构受益，一旦数据变更持久，就会通知下游系统。For example, change streams can save time for developers when implementing Extract, Transform, and Load (ETL) services, cross-platform synchronization, collaboration functionality, and notification services.例如，在实现提取、转换和加载（ETL）服务、跨平台同步、协作功能和通知服务时，变更流可以为开发人员节省时间。

Access Control访问控制¶

For deployments enforcing Authentication and authorization:对于强制执行身份验证和授权的部署：

• To open a change stream against specific collection, applications must have privileges that grant changeStream and find actions on the corresponding collection.要针对特定集合打开更改流，应用程序必须具有授予changeStream权限以及相应集合上find操作的权限。

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

• To open a change stream on a single databases, applications must have privileges that grant changeStream and find actions on all non-system collections in a database.若要在单个数据库上打开更改流，应用程序必须具有授予changeStream权限以及数据库中所有非系统集合上的find操作的权限。

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

• To open a change stream on an entire deployment, applications must have privileges that grant changeStream and find actions on all non-system collections for all databases in the deployment.要在整个部署中打开更改流，应用程序必须具有授予changeStream的权限以及在部署中所有数据库的所有非系统集合上find操作的权限。

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

Event Notification事件通知¶

Change streams only notify on data changes that have persisted to a majority of data-bearing members in the replica set. 更改流仅在副本集中的大多数数据承载成员都保留了数据更改时发出通知。This ensures that notifications are triggered only by majority-committed changes that are durable in failure scenarios.这确保了通知仅由大多数提交的更改触发，这些更改在故障场景中是持久的。

For example, consider a 3-member replica set with a change stream cursor opened against the primary. 例如，考虑一个带有primary打开的更改流游标的3个成员副本集。If a client issues an insert operation, the change stream only notifies the application of the data change once that insert has persisted to a majority of data-bearing members.如果客户机发出插入操作，更改流仅在插入持续存在于大多数数据承载成员之后通知应用程序数据更改。

If an operation is associated with a transaction, the change event document includes the txnNumber and the lsid.如果操作与事务相关，则变更事件文档包括txnNumber和lsid。

Collation排序规则¶

Starting in MongoDB 4.2, change streams use simple binary comparisons unless an explicit collation is provided. 从MongoDB 4.2开始，除非提供明确的排序规则，否则更改流使用simple的二进制比较。In earlier versions, change streams opened on a single collection (db.collection.watch()) would inherit that collection’s default collation.在早期版本中，在单个集合（db.collection.watch()）上打开的更改流将继承该集合的默认排序规则。