MongoDB220 课程

MongoDB

连接服务

连接 mflix cluster

mongo "mongodb+srv://mflix-d71ia.mongodb.net/test" --username m220student

上传文件到 Atlas

进入资源目录：

cd mflix-js

上传：

mongorestore --drop --gzip --uri "mongodb+srv://userName:password@<YOUR_CLUSTER_URI>" data

运行应用

修改配置 dotenv_win：

# Ticket: Connection
# Rename this file to .env after filling in your MFLIX_DB_URI and your SECRET_KEY
# Do not surround the URI with quotes
SECRET_KEY=super_secret_key_you_should_change
MFLIX_DB_URI=mongodb+srv://m220student:m220password@mflix-d71ia.mongodb.net/test
MFLIX_NS=mflix
PORT=5000

修改环境配置 .env 文件：

ren dotenv_win .env

再修改使用 notepad .env 才能进入

运行单元测试

npm test -t <testUnit>

课程中的单元测试通过 jest 和 async/await 编写。

MongoClient

创建一个 MongoClient 对象来初始化到数据库的连接，通过 testClient.db(dbname) 来查找数据库，并通过数据库方法操作集合。

简单的例子：

import { MongoClient } from "mongodb"
const URI = "mongodb+srv://m220-user:m220-pass@m220-test.mongodb.net/test"
const testClient = await MongoClient.connect(
  URI,
  { connectTimeoutMS: 100, retryWrites: true, useNewUrlParser: true },
)
const clientOptions = testClient.s.options

• 默认情况下，MongoClient 对象将根据 admin 数据库进行身份验证。要使用存储在另一个数据库中的登录凭据，可以在 URI 字符串的末尾添加 authSource=<some-other-DB>。
• 通过 SRV 字符串的连接默认使用 ssl。

import app from "./server"
import { MongoClient } from "mongodb"
import MoviesDAO from "../src/dao/moviesDAO"
import UsersDAO from "./dao/usersDAO"
import CommentsDAO from "./dao/commentsDAO"
const port = process.env.PORT || 8000
/**
Ticket: Connection Pooling
Please change the configuration of the MongoClient object by setting the
maximum connection pool size to 50 active connections.
*/
/**
Ticket: Timeouts
Please prevent the program from waiting indefinitely by setting the write
concern timeout limit to 2500 milliseconds.
*/
MongoClient.connect(
  process.env.MFLIX_DB_URI,
  // TODO: Connection Pooling
  // Set the poolSize to 50 connections.
  // TODO: Timeouts
  // Set the write timeout limit to 2500 milliseconds.
  { useNewUrlParser: true },
)
  .catch(err => {
    console.error(err.stack)
    process.exit(1)
  })
  .then(async client => {
    await MoviesDAO.injectDB(client)
    await UsersDAO.injectDB(client)
    await CommentsDAO.injectDB(client)
    app.listen(port, () => {
      console.log(`listening on port ${port}`)
    })
  })

CRUD

写入

collection.insertOne(obj) 方法为集合插入一行：

let insertResult = await collection.insertOne({
    title: 'Call of Duty',
    year: 2018,
})

通过结果对象的属性获取数据：

let { n, ok } = insertResult.result
console.log(n, ok)  // 1, 1
let count = insertResult.insertedCount
console.log(count)  // 1, equals to     ``insertResult.result.n``
let insertedId = insertResult.insertedId
// this is _id

insertedId 的用法是 collection.findOne({ _id: ObjectId(insertedId) })。

插入时如果没有传入 _id 属性的值，Mongodb 会自动分配一个 ObjectID，如果传入，则以传入为准（可以是字符串）。

注意：不允许 _id 重复，否则报错：

try {
    let dupIdResult = await collection.insertOne({
        _id: insertedId,
        title: "whatever",
        year: 2020,
    })
} catch(e) {
    console.log(e.errmsg)   // "E11000 duplicate key error collection"
}

collection.insertMany(ObjectArray) 插入多个文档：

let megaManYears = [
    1987,
    1988,
    1990,
    1991,
    1992,
      1993,
      1995,
      1996,
      2008,
    2010,
]
let docs = megaManYears.map((year, idx) => ({
    title: `Mega Man ${idx + 1}`,
    year,
}))
let insertResult = await videoGames.insertMany(docs)

不同之处在于传入多个文档的数组，返回结果类似单个写入，result.insertedIds 返回一个 _id 的数组。

默认是按顺序写入，如果报错会停止后续写入。

可以关闭顺序写入：

let insertResult = await videoGames.insertMany(docs, {"ordered": false})

写入安全机制（write concerns）

• writeConcern: { w: 0 }：不等待写入直接响应（可能未写入，但还是能获得连接错误等信息）。
• writeConcern: { w: 1 }：写入第一个服务器后响应成功（默认值）。
• writeConcern: { w: "majority" }：写入大多数服务器后响应成功。

try {
    await users.insertOne(
        {
            name: userInfo.name,
            email: userInfo.email,
            password: userInfo.password,
        },
        { w: "majority" }
    )
    return { success: true }
} catch(e) {
    if (String(e).startsWith("MongoError: E11000 duplicate key error")) {
        return { error: "A user with the given email already exists." }
    }
    console.error(`Error occurred while adding new user, ${e}.`)
    return { error: e }
}

writeConcern: { w: n } 的值不能大于副本集合的数量，否则报错！

更新

和写入不同之处在于，先查找是否已有文档，然后进行更新：

let updateResult = await collection.updateOne(
    { title: "Call of Duty" },
    { 
        $set: {
            title: "Call of Duty",
            year: 2020,
        }
    },
)

collection.updateOne(query, operation, flag) 和 collection.updateMany(query, operation, flag) 分别更新一个和多个文档； update(query, opration, flag) 默认更新一个，flag 为 {multi: true} 更新多个。

如果查找成功，则更新，否则不操作。
通过 updateResult.result.nModified 判断修改个数，updateResult.matchedCount 判断匹配个数。

update 操作符：

• $inc 增量
• $mul 乘于给定值
• $setOnInsert 如果使用 upsert 时，在插入条件下设置字段，更新则不设置
• $set 设置新值
• $unset 删除所有给出的字段

db.movieDetails.updateMany({
    rated: null
}, {
    $unset: {
        <filed1>: "",
        ...
    }
})

• $min 将原值和新值比较，取较小者
• $max 同上，取较大者
• $rename: {<field>: <name>}

对于数组字段：

• $addToSet 如果原数组中没有则添加
• $pop -1 移除第一个元素；1 移除最后一个元素
• $pullAll: { <arrField1>: [<value1>, <vaule2>...], ...} 移除所有值相同的元素
• $pull: { <arrField1>: <value|conditions>, ... }

如果没有该数组字段，则创建。

// {_id: 1, scores: [0, 2, 5, 5, 1, 0]}
db.survey.update({{ _id: 1} }, { $pullAll: { scores: [0, 5] } })
// {_id: 1, scores: [2, 1]}
db.stores.update({}, { $pull: { fruits: { $in: ["apples", "oranges"] }, vegetables: "carrots" } })

upsert

如果需要查找失败则创建新文档，否则进行更新操作，增加 upsert 选项：

let updateResult = await collection.updateOne(
    { title: "Call of Duty" },
    { 
        $set: {
            title: "Call of Duty",
        },
        $inc: {
            year: -1
        }
    },
    { upsert: true }
)

updateResult.result.upserted 对象包含了 _id 和 一个索引。

replaceOne(filter, doc)

在 shell 中 find 到需要修改的文档，用 JavaScript 修改后，再通过 replaceOne 更新数据库的数据。

只会拿到第一个匹配的文档。

查找

collection.findOne() 和 collection.find() 方法用于查找文档：

let result = await collection.findOne(
    { cast: "Johnny Depp" },
    { projection: { title: 1, year: 1, _id: 0 } },
)

第一个是查找对象，第二个是过滤结果，1 为需要，0 为不需要。
不用 projection 则返回所有 field 的结果；使用则默认带上 _id，可以手动去掉。

查找多个并返回 cursor：

let result = await collection.find(
    { cast: { $all: ["Johnny Depp", "Sam Hayek"] } }
)

返回的 cursor 对象有 .next() 方法以供遍历，或使用 .toArray() 方法获得所有结果的数组。

cursor 还有 .limit()，.sort()，.skip() 等方法，类似一个个过滤器。

！！cursor 方法顺序会被调整，例如 limit() 会在 skip() 后以保证返回 limit 的个数。

可以使用 collection.aggregate(pipeline) 查找起到相同效果：

const skippedPipeline = [
    { $match: { cast: "Johnney Depp" } },
    { $projection: { _id: 0, year: 1, title: 1, director: 1 } },
    { $sort: { year: -1 } },
    { $skip: 5 },
]
// ``$sort: { year: -1 }`` 表示按年递减排序，1 则为递增。
const skippedAggregation = await collection.aggregate(skippedPipeline)
const resultArr = skippedAggregation.toArray()

查找字段如果是嵌套等特殊名称，需要用引号包裹

读取安全机制（read concerns）

• "local"：读取第一个集合节点后响应。
• "majority"：读取大部分节点中的拷贝数据（随大流）。

db.restaurants.find({ _id: 5 }).readConcern("majority").maxTimeMS(10000)

合并

使用 $lookup 查找符合的结果并合并：

// in movies collection
const pipeline = [
  {
    '$match': {
      'year': {
        '$gte': 1980, 
        '$lt': 1990
      }
    }
  }, {
    '$lookup': {
      'from': 'comments',
      'let': { 'id': '$_id' },
      'pipeline': [
        {
            '$match': {
                '$expr': {
                    '$eq': [ '$movie_id', '$$id' ],
                },
            },
        }
      ],
      'as': 'movie_comments'
    }
  }
]

let 的作用是声明变量，否则 pipeline 中不能访问当前集合的字段（上例中是 movies 集合文档的 _id 字段），只能访问需要合并的集合的字段（上例中是 comments 集合）。
注意：$$ 前缀以使用 let 中声明的变量访问当前文档（movies），$ 前缀访问合并文档（comments）。

上例会从 comments 集合查找 movie_id 和当前集合的 _id 相同的文档，然后合并到当前文档的 movie_comments 字段中。

如果只关注评论数量：

const pipeline = {
    ...,
    '$lookup': {
        ...,
        'pipeline': [
            ...,
            {
                '$count': 'counts'
            }
        ]
    }
}

将使用 counts 字段存储评论数目。

删除

deleteOne() 和 deleteMany() 是两种删除操作。删除实际上是一种写入更新的操作：

• 集合内的数据改变
• 索引改变
• 操作日志中写入键值对

collection.count(query) 可以获得某集合中符合搜索条件的文档数目，传入 {} 可得到所有文档的总数。

删除一个或多个文档：

try {
    let deleteOneDoc = await videoGames.deleteOne({ year: 1998 })
    let deleteManyDocs = await videoGames.deleteMany({ year: { $lt: 1993 } })
} catch(e) {
    // ...
}

保证使用 collection.deleteOne() 时的查询结果唯一！否则可能错误删除文档。

弹性（Resilience）和健壮性（Robustness）

连接池

• 连接池允许复用连接，后续操作更快，避免大量操作发生大量连接而崩溃。
• 后续请求响应速度更快。
• MongoDB 连接池默认大小是 100 个。

每个连接池对应一个数据库客户端提供连接。

鲁棒性设置

• 要为重要数据的写入设置一个 wtimeout，尤其使用 w: "majority" 时，可能写入需要花费更多时间，此时如果有更多读写操作，而写入成功尚未响应，会导致系统堵塞。wtimeout 取决于网络和硬件。{ w: "majority", wtimeout: 5000 }
• 一定要处理 serverSelectionTimeout 错误。同时能被动地监控应用堆的健康，并及时发现软硬件问题。默认情况，等待 30s 会报错，应该根据实际情况设置。

使用比 w: 1 更重要的写入时强烈建议设置 wtimeout。

错误处理

• Dulplicate Key Errors
  使用新 _id。
• Timeout Errors
  增加 wtimeoutMS，或减少 w 值。
• Write Concern Errors
  减少 w 值。

try {
    let errOp = await db.collection.insertOne({
        _id: 0,
    },
    {
        w: 5,
        wtimeoutMS: 1,
    })
}