Skip to content

Commit 1651534

Browse files
rockyshi1993rockyshi1993
committed
添加stream流支持
1 parent c9caaba commit 1651534

13 files changed

Lines changed: 3012 additions & 23 deletions

‎README.md‎

Lines changed: 602 additions & 14 deletions
Large diffs are not rendered by default.

‎STATUS.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -93,7 +93,7 @@
9393
- ✅ count
9494
- 统计匹配文档数；totals.sync/async 会透传 hint/collation/maxTimeMS。
9595
- 性能优化：空查询自动使用 estimatedDocumentCount（基于元数据，速度快）；有查询条件使用 countDocuments（精确统计）。
96-
- ❌ stream（find 流式返回）
96+
- ✅ stream（流式返回）
9797
- 支持流式查询，适合处理大数据集；默认 batchSize=1000；支持 maxTimeMS/hint/collation/noCursorTimeout。
9898
- 自动记录慢查询日志；触发 slow-query 和 query 事件；不支持缓存（流式特性）。
9999
- ✅ 聚合（aggregate）

‎examples/exports/orders_export.csv‎

Lines changed: 0 additions & 1 deletion
This file was deleted.

‎examples/run-all.js‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
/**
2+
* 运行所有流式查询示例的测试脚本
3+
*/
4+
5+
const streamBasic = require('./stream-basic');
6+
const streamTransform = require('./stream-transform');
7+
const streamExport = require('./stream-export');
8+
const streamFindPage = require('./stream-findpage');
9+
10+
async function runAllExamples() {
11+
console.log('=' .repeat(70));
12+
console.log('🚀 开始运行所有流式查询示例');
13+
console.log('=' .repeat(70));
14+
console.log();
15+
16+
const examples = [
17+
{ name: '基础流式查询', fn: streamBasic },
18+
{ name: '流式数据转换', fn: streamTransform },
19+
{ name: '数据导出', fn: streamExport },
20+
{ name: 'findPage 流式查询', fn: streamFindPage },
21+
];
22+
23+
for (const example of examples) {
24+
try {
25+
console.log(`\n🎯 正在运行: ${example.name}`);
26+
console.log('='.repeat(70));
27+
await example.fn();
28+
console.log(`\n✅ ${example.name} 执行完成\n`);
29+
} catch (error) {
30+
console.error(`\n❌ ${example.name} 执行失败:`, error.message);
31+
console.error(error.stack);
32+
}
33+
}
34+
35+
console.log('\n' + '='.repeat(70));
36+
console.log('🎉 所有示例执行完成！');
37+
console.log('='.repeat(70));
38+
}
39+
40+
if (require.main === module) {
41+
runAllExamples().catch(err => {
42+
console.error('执行失败:', err);
43+
process.exit(1);
44+
});
45+
}
46+
47+
module.exports = runAllExamples;
48+

‎examples/stream-basic.js‎

Lines changed: 186 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,186 @@
1+
/**
2+
* 基础流式查询示例
3+
* 演示如何使用流式查询处理数据
4+
*/
5+
6+
const MonSQLize = require('../lib/index');
7+
8+
async function basicStreamExample() {
9+
console.log('📦 基础流式查询示例\n');
10+
11+
const msq = new MonSQLize({
12+
type: 'mongodb',
13+
databaseName: 'test',
14+
config: { uri: 'mongodb://localhost:27017' },
15+
slowQueryMs: 2000,
16+
});
17+
18+
try {
19+
const { collection } = await msq.connect();
20+
console.log('✅ 数据库连接成功\n');
21+
22+
// ============================================================
23+
// 示例 1: 使用 stream() 方法（推荐，最简洁）
24+
// ============================================================
25+
console.log('示例 1: 使用 stream() 方法');
26+
console.log('-'.repeat(60));
27+
28+
let count1 = 0;
29+
const stream1 = collection('orders').stream({
30+
query: { status: 'paid' },
31+
projection: { _id: 1, amount: 1, createdAt: 1 },
32+
sort: { createdAt: -1 },
33+
limit: 100,
34+
batchSize: 20
35+
});
36+
37+
stream1.on('data', (doc) => {
38+
count1++;
39+
if (count1 <= 5) {
40+
console.log(` 文档 ${count1}:`, doc);
41+
}
42+
});
43+
44+
await new Promise((resolve, reject) => {
45+
stream1.on('end', () => {
46+
console.log(`✅ 共处理 ${count1} 条数据\n`);
47+
resolve();
48+
});
49+
stream1.on('error', reject);
50+
});
51+
52+
// ============================================================
53+
// 示例 2: 使用 find({stream: true})（等价写法）
54+
// ============================================================
55+
console.log('示例 2: 使用 find({stream: true})');
56+
console.log('-'.repeat(60));
57+
58+
let count2 = 0;
59+
const stream2 = collection('orders').find({
60+
query: { status: 'pending' },
61+
stream: true,
62+
batchSize: 10
63+
});
64+
65+
stream2.on('data', (doc) => {
66+
count2++;
67+
});
68+
69+
await new Promise((resolve, reject) => {
70+
stream2.on('end', () => {
71+
console.log(`✅ 共处理 ${count2} 条数据\n`);
72+
resolve();
73+
});
74+
stream2.on('error', reject);
75+
});
76+
77+
// ============================================================
78+
// 示例 3: 使用 for await 语法（推荐，代码更简洁）
79+
// ============================================================
80+
console.log('示例 3: 使用 for await 语法');
81+
console.log('-'.repeat(60));
82+
83+
const stream3 = collection('orders').stream({
84+
query: { status: 'shipped' },
85+
limit: 50
86+
});
87+
88+
let count3 = 0;
89+
try {
90+
for await (const doc of stream3) {
91+
count3++;
92+
if (count3 <= 3) {
93+
console.log(` 处理文档 ${count3}:`, doc._id);
94+
}
95+
}
96+
console.log(`✅ 共处理 ${count3} 条数据\n`);
97+
} catch (error) {
98+
console.error('❌ 处理错误:', error);
99+
}
100+
101+
// ============================================================
102+
// 示例 4: 聚合管道流式处理
103+
// ============================================================
104+
console.log('示例 4: 聚合管道流式处理');
105+
console.log('-'.repeat(60));
106+
107+
const aggStream = collection('orders').aggregate([
108+
{ $match: { status: 'paid' } },
109+
{ $group: { _id: '$userId', total: { $sum: '$amount' }, count: { $sum: 1 } } },
110+
{ $sort: { total: -1 } },
111+
{ $limit: 10 }
112+
], {
113+
stream: true,
114+
allowDiskUse: true
115+
});
116+
117+
let count4 = 0;
118+
aggStream.on('data', (doc) => {
119+
count4++;
120+
if (count4 <= 3) {
121+
console.log(` 用户统计:`, doc);
122+
}
123+
});
124+
125+
await new Promise((resolve, reject) => {
126+
aggStream.on('end', () => {
127+
console.log(`✅ 共处理 ${count4} 条聚合结果\n`);
128+
resolve();
129+
});
130+
aggStream.on('error', reject);
131+
});
132+
133+
// ============================================================
134+
// 示例 5: 流式错误处理
135+
// ============================================================
136+
console.log('示例 5: 流式错误处理');
137+
console.log('-'.repeat(60));
138+
139+
const stream5 = collection('orders').stream({
140+
query: {},
141+
limit: 20
142+
});
143+
144+
let count5 = 0;
145+
let hasError = false;
146+
147+
stream5.on('data', (doc) => {
148+
count5++;
149+
});
150+
151+
stream5.on('error', (error) => {
152+
hasError = true;
153+
console.error('❌ 流错误:', error.message);
154+
});
155+
156+
await new Promise((resolve) => {
157+
stream5.on('end', () => {
158+
if (!hasError) {
159+
console.log(`✅ 流正常结束，处理 ${count5} 条数据\n`);
160+
}
161+
resolve();
162+
});
163+
stream5.on('error', resolve);
164+
});
165+
166+
console.log('=' .repeat(60));
167+
console.log('✅ 所有示例执行完成');
168+
console.log('=' .repeat(60));
169+
170+
} catch (error) {
171+
console.error('❌ 错误:', error.message);
172+
console.error(error.stack);
173+
process.exit(1);
174+
} finally {
175+
await msq.close();
176+
console.log('\n✅ 连接已关闭');
177+
}
178+
}
179+
180+
// 运行示例
181+
if (require.main === module) {
182+
basicStreamExample();
183+
}
184+
185+
module.exports = basicStreamExample;
186+

0 commit comments

Comments
 (0)