mm_mongodb

一个基于 MongoDB 的数据库操作模块，提供与 mm_redis / mm_cachebase 完全对齐 的异步缓存操作接口，同时保留 MongoDB 原生的 CRUD、事务与索引管理能力，支持连接池、TTL 自动过期和冒号风格的事件通知系统。

项目信息

• 当前版本: 2.0.0
• 更新日期: 2026-08-04
• Gitee地址: https://gitee.com/qiuwenwu91/mm_mongodb
• Node.js 版本要求: >= 14.0.0（推荐 16.0.0 及以上）
• MongoDB 版本要求: 4.0+

特点

• 接口与 mm_redis / mm_cachebase 完全对齐：全异步 API、冒号风格事件、统一方法签名与返回值，三模块可无缝切换
• 全异步设计：所有公共方法均为 async，返回 Promise
• 冒号风格事件：set:before / set:after / set:error，与 mm_redis / mm_cachebase 一致
• 完整的数据类型支持：字符串、哈希表、列表、集合、有序集合、发布订阅
• MongoDB 原生能力：保留核心 CRUD、事务（ACID）、索引与 TTL 索引管理
• 连接池管理：内置连接池，支持自动重连与重试
• TTL 自动过期：通过 expire_at 字段与 TTL 索引实现自动过期清理
• 查询注入防护：内置 _sanitizeQuery 防止 NoSQL 注入

安装

npm install mm_mongodb

基本使用

Redis 兼容 API（与 mm_redis / mm_cachebase 对齐）

const { MongoDB, mongodbAdmin } = require('mm_mongodb');

// 通过连接池管理器获取实例（推荐）
const cache = mongodbAdmin('myapp');

async function example() {
    // 连接数据库
    await cache.open();

    // 设置缓存，10 秒过期
    await cache.set("name", "张三", 10);

    // 获取缓存
    const value = await cache.get("name");
    $.log.debug(value); // 输出: 张三

    // 删除缓存（返回删除数量）
    const deleted = await cache.del("name");
    $.log.debug(deleted); // 输出: 1

    // 关闭连接
    await cache.close();
}

MongoDB 原生 API

const { MongoDB } = require('mm_mongodb');

const db = new MongoDB({ database: 'mydb' });
await db.open();

// 插入文档
await db.addObj({ key: 'user:1', name: 'John', age: 25 });

// 查询文档
const user = await db.findOne({ key: 'user:1' });

// 更新文档
await db.setting({ key: 'user:1' }, { age: 26 });

// 删除文档
await db.delete({ key: 'user:1' });

await db.close();

配置

环境变量

可通过环境变量配置 MongoDB 连接：

MONGO_URL=mongodb://username:password@host:port/database
MONGO_HOST=localhost
MONGO_PORT=27017
MONGO_USERNAME=username
MONGO_PASSWORD=password
MONGO_DATABASE=database
MONGO_AUTH_SOURCE=admin

构造函数配置

const db = new MongoDB({
    host: 'localhost',
    port: 27017,
    database: 'mm',
    user: '',
    password: '',
    min_pool_size: 10,
    max_pool_size: 100,
    connect_timeout_ms: 10000,
    server_selection_timeout_ms: 5000,
    retry_writes: true
});

连接管理

与 mm_redis / mm_cachebase 一致的连接管理接口：

const cache = mongodbAdmin('myapp');
await cache.open();           // 打开连接
const connected = cache.isConnected(); // 检查连接状态，返回 Boolean
await cache.close();          // 关闭连接，清理发布订阅频道

连接特性：

• 内置连接 Promise 缓存（connectPromise），避免重复连接
• 最大重试次数 maxConnectAttempts = 3，自动重连
• 连接成功后自动创建 key 唯一索引与 expire_at TTL 索引
• 监听连接 error / close 事件，自动更新连接状态

API 说明

所有方法均为 async，返回 Promise。事件名采用冒号风格（xxx:before / xxx:after / xxx:error）。

键值操作

| 方法 | 签名 | 返回值 | |------|------|--------| | set | set(key, value, seconds=0) | Promise<Boolean> | | get | get(key) | Promise<Object\|null> | | del | del(key) | Promise<Number> 删除数量，支持键名数组 | | exists | exists(key) | Promise<Number> 0/1（单键）或计数（多键） | | has | has(key) | Promise<Boolean> | | getset | getset(key, value, seconds=0) | Promise<Object\|null> 旧值 | | add | add(key, value, seconds=0) | Promise<Boolean> SETNX 语义 |

批量操作

| 方法 | 签名 | 返回值 | |------|------|--------| | mset | mset(obj, seconds=0) | Promise<Boolean> | | mget | mget(keys) | Promise<Array> 不存在的键返回 null |

过期时间管理

| 方法 | 签名 | 返回值 | |------|------|--------| | expire | expire(key, seconds) | Promise<Boolean> | | pexpire | pexpire(key, milliseconds) | Promise<Boolean> | | persist | persist(key) | Promise<Boolean> | | ttl | ttl(key, [seconds]) | Promise<Number\|Boolean> 不传 seconds 返回剩余秒数（-1 无过期，-2 已过期） |

过期通过 expire_at 字段配合 TTL 索引（expireAfterSeconds: 0）由 MongoDB 自动清理。

哈希表操作

| 方法 | 签名 | 返回值 | |------|------|--------| | hset | hset(key, field, value) | Promise<Boolean> | | hget | hget(key, field) | Promise<Object\|null> | | hgetall | hgetall(key) | Promise<Object> | | hdel | hdel(key, fields) | Promise<Number> 删除字段数，支持字段数组 | | hmset | hmset(key, obj) | Promise<Boolean> | | hmget | hmget(key, fields) | Promise<Array> 不存在的字段返回 null | | hlen | hlen(key) | Promise<Number> | | hexists | hexists(key, field) | Promise<Boolean> | | hkeys | hkeys(key) | Promise<Array> 字段名数组 | | hvals | hvals(key) | Promise<Array> 值数组 |

数值操作

| 方法 | 签名 | 返回值 | |------|------|--------| | incr | incr(key, seconds=0) | Promise<Number> 错误返回 null | | decr | decr(key, seconds=0) | Promise<Number> 错误返回 null | | incrby | incrby(key, increment, seconds=0) | Promise<Number> 错误返回 null | | decrby | decrby(key, decrement, seconds=0) | Promise<Number> 错误返回 null | | incrbyfloat | incrbyfloat(key, increment, seconds=0) | Promise<Number> 错误返回 null | | addInt | addInt(key, value, seconds=0) | Promise<Number> 兼容方法 | | addFloat | addFloat(key, num, seconds=0) | Promise<Number> 兼容方法 |

字符串操作

| 方法 | 签名 | 返回值 | |------|------|--------| | setrange | setrange(key, value, seconds=0) | Promise<Number> 追加后长度 | | append | append(key, value) | Promise<Number> 长度 | | strlen | strlen(key) | Promise<Number> |

列表操作

| 方法 | 签名 | 返回值 | |------|------|--------| | setForList | setForList(key, value, seconds=0) | Promise<Number> 列表长度 | | addForList | addForList(key, value, seconds=0) | Promise<Number> 列表长度 | | getForList | getForList(key, start=0, end=-1) | Promise<Array> 支持 Redis 风格负索引 | | hasForList | hasForList(key, value) | Promise<Boolean> | | clearForList | clearForList(key) | Promise<Boolean> |

集合操作

| 方法 | 签名 | 返回值 | |------|------|--------| | sadd | sadd(key, members) | Promise<Number> 添加数量，支持成员数组 | | srem | srem(key, members) | Promise<Number> 移除数量，支持成员数组 | | sismember | sismember(key, member) | Promise<Boolean> | | smembers | smembers(key) | Promise<Array> | | scard | scard(key) | Promise<Number> | | sinter | sinter(...keys) | Promise<Array> 交集 | | sunion | sunion(...keys) | Promise<Array> 并集 | | sdiff | sdiff(...keys) | Promise<Array> 差集 |

有序集合操作

| 方法 | 签名 | 返回值 | |------|------|--------| | zadd | zadd(key, score, member) | Promise<Boolean> 新成员返回 true | | zrem | zrem(key, members) | Promise<Number> 移除数量，支持成员数组 | | zscore | zscore(key, member) | Promise<Number\|null> | | zrank | zrank(key, member) | Promise<Number\|null> | | zrange | zrange(key, start, stop, withScores=false) | Promise<Array> | | zcard | zcard(key) | Promise<Number> | | zincrby | zincrby(key, increment, member) | Promise<Number> |

发布订阅

注意：发布订阅为进程内实现（基于 _channels 回调表），无跨进程能力，与 mm_redis 的跨进程语义不同，与 mm_cachebase 一致。

| 方法 | 签名 | 返回值 | |------|------|--------| | subscribe | subscribe(channel, func) | Promise<Boolean> | | unsubscribe | unsubscribe(channel) | Promise<Boolean> | | publish | publish(channel, message) | Promise<Number> 接收者数量 |

键管理

| 方法 | 签名 | 返回值 | |------|------|--------| | keys | keys(pattern) | Promise<Array> 支持 * 通配符 | | dbsize | dbsize() | Promise<Number> | | flushdb | flushdb() | Promise<Boolean> | | clear | clear([pattern]) | Promise<Boolean> flushdb 扩展，支持按模式清空 | | clearExpire | clearExpire() | Promise<Boolean> 主动清理过期数据 |

MongoDB 原生 API

除 Redis 兼容 API 外，本模块保留 MongoDB 原生操作能力。

核心 CRUD

| 方法 | 签名 | 返回值 | |------|------|--------| | findOne | findOne(query, options) | Promise<Object\|null> | | find | find(query, options) | Promise<Array> 自动防注入 | | addObj | addObj(obj) | Promise<Object> 插入结果 | | addList | addList(list) | Promise<Object> 批量插入结果 | | setting | setting(query, body, options) | Promise<Object\|Boolean> 修改结果 | | delete | delete(query) | Promise<Object> 删除结果 | | count | count(query) | Promise<Number> |

表与索引管理

| 方法 | 签名 | 返回值 | |------|------|--------| | addTable | addTable(table) | Promise<Object> 创建集合 | | createIndex | createIndex(keys, options) | Promise<Object> | | createTTLIndex | createTTLIndex(keys, expireAfterSeconds) | Promise<Object> | | getIndexes | getIndexes() | Promise<Array> | | dropIndex | dropIndex(indexName) | Promise<Boolean> | | dropTable | dropTable() | Promise<Boolean> |

事务管理

| 方法 | 签名 | 返回值 | |------|------|--------| | startTransaction | startTransaction() | Promise<Object> 会话对象 | | commitTransaction | commitTransaction(session) | Promise<void> | | rollbackTransaction | rollbackTransaction(session) | Promise<void> |

const session = await db.startTransaction();
try {
    await db.addObj({ key: 'a', value: 1 }, { session });
    await db.addObj({ key: 'b', value: 2 }, { session });
    await db.commitTransaction(session);
} catch (err) {
    await db.rollbackTransaction(session);
}

事件系统

事件名采用冒号风格，与 mm_redis / mm_cachebase 完全一致。每个操作有 :before、:after、:error 三种事件。

const cache = mongodbAdmin('myapp');
await cache.open();

// 监听 set 前置事件，可修改参数或取消操作
cache.on('set:before', (ctx) => {
    $.log.debug('即将设置键:', ctx.key, '值:', ctx.value);
    // 可修改 ctx.key / ctx.value / ctx.seconds
    // 设置 ctx.cancel = true 可取消操作
});

// 监听 set 后置事件
cache.on('set:after', (data) => {
    $.log.debug('设置完成:', data.key, '成功:', data.success);
});

// 监听错误事件
cache.on('set:error', (data) => {
    $.log.error('设置出错:', data.key, data.error);
});

支持的事件

所有公共方法均有对应事件，命名规则为 方法名:before / 方法名:after / 方法名:error：

• 键值：set、get、del、exists、add
• 批量：mset、mget
• 过期：expire、pexpire、persist、ttl
• 哈希：hset、hget、hgetall、hdel、hmset、hmget、hlen
• 数值：incr、decr、incrby、decrby、incrbyfloat、addInt、addFloat
• 字符串：setrange
• 列表：setForList、addForList、getForList、hasForList、clearForList
• 集合：sadd、srem、sismember、smembers
• 有序集合：zadd、zrem
• 键管理：keys、clear、clearExpire
• 原生操作：addObj、addList、delete、setting

注意：destroy 事件不可使用冒号格式（destroy:after），因为 mm_expand 的 Base.emit 会将冒号格式解释为 do 流程的阶段，会触发 _destroyCore。

使用示例

哈希表操作

async function hashExample() {
    await cache.hset('user:1', 'name', '张三');
    await cache.hmset('user:1', { age: 25, city: '北京' });
    const name = await cache.hget('user:1', 'name'); // 张三
    const all = await cache.hgetall('user:1'); // { name: '张三', age: 25, city: '北京' }
    await cache.hdel('user:1', ['age', 'city']); // 2
}

集合操作

async function setExample() {
    await cache.sadd('tags', ['node', 'js', 'cache']);
    const isMember = await cache.sismember('tags', 'node'); // true
    const members = await cache.smembers('tags'); // ['node', 'js', 'cache']
    const count = await cache.scard('tags'); // 3
    await cache.srem('tags', ['js']);
}

有序集合操作

async function zsetExample() {
    await cache.zadd('ranking', 100, 'user1');
    await cache.zadd('ranking', 200, 'user2');
    const rank = await cache.zrank('ranking', 'user1'); // 0
    const top = await cache.zrange('ranking', 0, -1, true); // [{member:'user1', score:100}, ...]
    await cache.zincrby('ranking', 50, 'user1'); // 新分数 150
}

发布订阅

async function pubsubExample() {
    await cache.subscribe('news', (message) => {
        $.log.debug('收到消息:', message);
    });
    await cache.publish('news', 'hello world'); // 接收者数量: 1
}

与 mm_redis / mm_cachebase 的接口一致性

本模块与 mm_redis、mm_cachebase 提供完全对齐的 Redis 兼容接口，调用方可通过统一的方式在内存缓存、Redis、MongoDB 之间无缝切换：

• 全异步：所有方法返回 Promise
• 事件命名：冒号风格 xxx:before / xxx:after / xxx:error
• 方法签名：set(key, value, seconds=0)、mset(obj, seconds=0)、incrby(key, increment, seconds=0) 等
• 返回值：hset 返回 Boolean，del/exists/hdel 返回 Number，hget 不存在返回 null

差异点：

• 发布订阅为进程内实现（mm_redis 支持跨进程，mm_cachebase 与本模块一致为进程内）
• 集合/有序集合用 JS 数组实现，存于 MongoDB 文档的 value 字段，无 Redis 的 O(log N) 复杂度保证
• 过期依赖 MongoDB 的 TTL 索引后台清理，与 mm_redis 的精确过期、mm_cachebase 的定时轮询语义略有差异
• 本模块额外提供 MongoDB 原生 CRUD、事务、索引管理能力，为 mm_redis / mm_cachebase 所不具备

性能优化

1. 使用连接池管理器 - 始终通过 mongodbAdmin 获取实例，复用连接池
2. 创建合适的索引 - 为频繁查询的字段创建索引
3. 使用批量操作 - 多个操作时优先使用 mset、mget、addList
4. 设置过期时间 - 临时数据使用 seconds 参数或 expire，配合 TTL 索引自动清理
5. 优化查询 - 避免全表扫描，使用精确查询条件

安全考虑

• 密码自动 URI 编码，安全传输
• 内置 _sanitizeQuery 查询注入防护，过滤 $ 操作符注入
• 敏感配置通过环境变量管理
• 自动连接重试，提升稳定性
• 避免在查询中直接使用原始用户输入

依赖

• mm_logs - 日志模块
• mongodb - 官方 MongoDB 驱动
• mm_expand - 基础类扩展

兼容性

• Node.js: >= 14.0.0（推荐 16.0.0+）
• MongoDB: 4.0+

从 Redis 迁移

从 mm_redis 迁移到 mm_mongodb 非常简单，接口完全一致，仅需替换模块与初始化方式：

// mm_redis（原）
// const { redisAdmin } = require('mm_redis');
// const cache = redisAdmin('default');
// await cache.open();

// mm_mongodb（替换）
const { mongodbAdmin } = require('mm_mongodb');
const cache = mongodbAdmin('default');
await cache.open();

// Redis 命令保持不变
await cache.set('key', 'value');
const value = await cache.get('key');

贡献

欢迎贡献代码！请遵循以下步骤：

1. Fork 仓库
2. 创建功能分支：git checkout -b feature/your-feature
3. 提交更改：git commit -m "feat: add your feature"
4. 推送到分支：git push origin feature/your-feature
5. 提交 Pull Request

问题与支持

如遇问题或有疑问：

• Gitee Issues: https://gitee.com/qiuwenwu91/mm_mongodb/issues

许可证

本项目基于 MIT 许可证开源 - 详见 LICENSE 文件。

作者

邱文武 - qww.elins.cn

致谢

感谢所有贡献者和用户对 mm_mongodb 的改进与支持！