nodejs+mongoose操作mongodb副本集实例

继上一篇设置mongodb副本集之后，开始使用nodejs访问mongodb副本集：

1：创建项目     express 项目名称    

2：npm install mongoose    安装mongoose包

3：mongoose连接数据库配置 （需要写primary，读从secondary读取数据，需要设置读取偏好设置）

理解MongDB驱动支持5中读偏好模式。

读偏好模式	描述
primary	默认模式，所有的读操作都从当前副本集主节点
primaryPreferred	多数情况下,从主节点读取数据，但是如果主节点不可用了，会从从节点读取
secondary	所有读操作都从副本集的从节点读取
secondaryPreferred	多数情况下， 从从节点进行读操作，但是如果从节点都不可用了，从主节点读取
nearest	从副本集中延迟最低的成员读取，不考虑成员的类型

读偏好在通过mongos连接到分片集群时也可以使用。mongos实例在连接到集群中副本集提供的分片时遵守读偏好。

【一般来说，不要使用secondary和secondaryPreferred来提供额外的读能力，因为：

• 所有的副本集大致拥有相同的写流量，所以，从节点服务读操作的速率大致和主节点相同。
• 副本集是异步同步的，因此在写操作成功到复制到从节点有一定的延迟。从从节点读取数据返回的可能是过期的数据，从不同从节点读取可能导致非单调读(不同节点返回不一致的数据)。
• 对于分片集合的查询，对于均衡器开启的集群，由于不完整或被终止的块合并，从节点可能返回丢失或重复的过期数据。

分片通过增加一组机器分散读写操作增加了读写性能，这通常是增加性能更好的方法。】

var mongoose = require('mongoose');
mongoose.Promise = Promise;

// Promise.promisifyAll(mongoose);

var uri = 'mongodb://172.16.9.241/replSetTest,mongodb://172.16.9.240/replSetTest,mongodb://172.16.9.84/replSetTest';

var opts = {
  db: { native_parser: true },
  server: {
    poolSize: 5 ,
    auto_reconnect: true,
    readPreference: "secondaryPreferred",   //读偏好设置

    socketOptions: {keepAlive: 1}
  },
  replset: { rs_name:"testrepl",
    readPreference: "secondaryPreferred"}    //读偏好设置
}

//mongoose连接 global.db = mongoose.createConnection(uri,opts); mongoose.connection = global.db; db.on("error",function(err){ console.error(err); }); db.on("open",function(){ console.log("dbopen"); });

4：schema/user.js

var mongoose = require('mongoose');
var Schema = mongoose.Schema;

var moment = require("moment");

var schema = new Schema({
  index: {type: Number, default: 0},
  name: {type: String, default: ""},
  age: {type: Number},
  createdAt:{type:Date, default:Date.now()},
  remark:{type:String,default:""}
});

var formatCreatedAt = schema.virtual('formatCreatedAt');
formatCreatedAt.get(function () {
  return new moment(this.createdAt).format("YYYY-MM-DD HH:mm:ss:SSS");
});
schema.pre('save',function(next){
  this.createdAt = Date.now();
  next();
});
module.exports = schema;

5: 写数据操作，每隔一秒写入一条数据

//写数据操作
var index=0;
function getMaxIndex() {
  return new Promise(function (resolve,reject) {
    user.find({},function (err, items) {
      console.log(items);
      var maxItem = ++items.length;
      console.log("maxItem:",maxItem);
      if(err){
        console.log("getMaxIndex:",err);
        return reject(err);
      }
      return resolve(maxItem);
    });
  })
}
function createUser(index) {
  console.log("index:",index);
  return new Promise(function (resolve,reject) {
    var u = {
      index:index,
      name:'user'+index,
      age:22,
      remark:"测试文字" };
    user.create(u,function (err,res) {
      if(err){
        console.error(err);
        return reject(err);
      }
      console.log("createUser:",res);
      return resolve('ok');
    })

  })
}
setInterval(function () {
  getMaxIndex().then(createUser);
},1000);

可以测试当primary节点下线，写数据操作会自动写入其他选举后的主节点

6：读操作，不停读出最大Index的记录，并将操作的数据写入文件，比对读文件时间和写数据的时间，算出同步时间长。

//读数据操作
function getUserByIndex() {
  user.findOne({index:nextIndex},function (err, item) {
    if (err) {
      console.log("getUserByIndex:", err);
    }
    if(item) {
      console.log(JSON.stringify(item));
      let str = " Index: " + item.index + "  read time: " + new moment(Date.now()).format("YYYY-MM-DD HH:mm:ss:SSS")+
        " write time: " + item.formatCreatedAt + "
";
      content += str;
      ++nextIndex;
      //console.log("user存在的nextIndex: ", nextIndex);

    }
    if(nextIndex==100){
      writeFile(content);
    }else {
      process.nextTick(getUserByIndex);
    }
  });
}
getUserByIndex();