如果数据库关闭一段时间，MongoDB会更改流超时 [英] MongoDB change stream timeouts if database is down for some time

问题描述

我在nodejs中使用mongoDB变更流，一切正常，但是如果数据库关闭花费了10到5秒钟以上的时间来启动变更流抛出超时错误，这是我的变更流观察者代码

I am using mongoDB change stream in nodejs, everything works fine but if database is down has taken more than 10 5 seconds to get up change stream throws timeout error, here is my change stream watcher code

Service.prototype.watcher = function( db ){

let collection = db.collection('tokens');
let changeStream = collection.watch({ fullDocument: 'updateLookup' });
let resumeToken, newChangeStream;

changeStream.on('change', next => {
    resumeToken = next._id;
    console.log('data is ', JSON.stringify(next))
    changeStream.close();
    // console.log('resumeToken is ', JSON.stringify(resumeToken))
    newChangeStream = collection.watch({ resumeAfter : resumeToken });
    newChangeStream.on('change', next => {
        console.log('insert called ', JSON.stringify( next ))
    });
});

无论如何，我已经在数据库端进行了处理，即，如果数据库已关闭或通过使用此代码重新连接

however on database end i have handled it, i.e if database is down or reconnected by using this code

 this.db.on('reconnected', function () {
    console.info('MongoDB reconnected!');
});
this.db.on('disconnected', function() {
    console.warn('MongoDB disconnected!');
});

但是我无法处理变更流观察程序以在数据库关闭时将其停止，并在重新连接数据库时或者是否还有其他更好的方法来重新启动它?

but i am not able to handle change stream watcher to stop it when database is down and start it again when database is reconnected or if there is any other better way to do it ?

推荐答案

您想要做的是将watch()调用封装在一个函数中.然后，此函数将在错误时自行调用，以使用先前保存的恢复令牌重新监视集合.您所拥有的代码中缺少的是错误处理程序.例如:

What you want to do is to encapsulate the watch() call in a function. This function will then call itself on error, to rewatch the collection using a previously saved resume token. What is missing from the code you have is the error handler. For example:

const MongoClient = require('mongodb').MongoClient
const uri = 'mongodb://localhost:27017/test?replicaSet=replset'
var resume_token = null

run()

function watch_collection(con, db, coll) {
  console.log(new Date() + ' watching: ' + coll)
  con.db(db).collection(coll).watch({resumeAfter: resume_token})
    .on('change', data => {
      console.log(data)
      resume_token = data._id
    })
    .on('error', err => {
      console.log(new Date() + ' error: ' + err)
      watch_collection(con, coll)
    })
}

async function run() {
  con = await MongoClient.connect(uri, {"useNewUrlParser": true})
  watch_collection(con, 'test', 'test')
}

请注意，watch_collection()包含watch()方法及其处理程序.更改时，它将打印更改并存储恢复令牌.错误时，它会再次调用自己以重新监视该集合.

Note that watch_collection() contains the watch() method along with its handler. On change, it will print the change and store the resume token. On error, it will call itself to rewatch the collection again.

这篇关于如果数据库关闭一段时间，MongoDB会更改流超时的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！