目录
1. 背景与动机
在微服务架构或前后端分离的系统中，往往会出现这样一个中间层（Gateway、API 层或 BFF—Backend For Frontend）：客户端发起 N 个请求到中间层，由中间层再统一转发到后端服务。若不加控制，短时间内大量重复或类似请求会导致后端压力骤增、网络带宽浪费、响应延迟飙升，甚至引发“雪崩”故障。
**请求合并（Request Coalescing）**意在中间层将多个对同一资源的并发请求合并成一次后端调用，其他请求“排队”等待同一次调用的返回结果。这样可以大幅减少后端调用次数，降低整体延迟并保护后端系统的稳定性。
2. 请求合并概述
2.1 什么是请求合并？
- 去重（Duplicate Suppression）
当短时间内出现多个对同一资源（同一 URL、同一参数）的请求时，只发起一次后端调用，将结果“广播”给所有等待的请求。 - 批量（Batching）
将多个不同但兼容的请求合并成一次批量调用，例如客户端请求：/user/1、/user/2、/user/3可以合并成后端调用：/users?ids=1,2,3。
两者在具体场景中常常结合使用。去重侧重于“同一资源多次请求只发一次”，批量侧重于“多个资源请求合并成一个多异步调用”。
2.2 为何需要请求合并？
- 降低后端负载：避免短时间内同一资源被重复查询。
- 减少网络开销：一次批量调用往往比分别多次调用更省时省带宽。
- 降低响应延迟：合并后减少排队时间，总体完成更快。
- 提高系统稳定性：在高并发场景下防止对后端的瞬时洪峰，避免雪崩。
3. 核心思想与基本模式
3.1 去重（Duplicate Suppression）
思路：在内存中维护一个待处理请求列表（pending map），以资源标识（Key）为索引。收到请求后：
判断该 Key 是否已有正在执行的后端调用。
- 是 → 将当前请求的 Promise 或回调加入待通知队列，不再发起新调用。
- 否 → 发起一次后端调用，并将 Key 与“当前 Promise”注册到 pending map。
- 后端调用返回后，将结果或错误通知给 pending map 中所有注册的请求，再清理 pending map。
3.2 批量（Batching）
思路：将 N 个对不同资源但满足合并条件的请求，聚合成一次批量调用。例如在 10ms 内收到 5 个用户查询，请求 /users/:id，可以合并成 /users?ids=[...]：
- 队列缓存：收到请求后，将其 Key（如 id）与回调存入一个数组。
- 定时触发：设置一个短暂定时（如 5–10ms），到时将队列中所有 Key 合并并发起后端批量调用。
- 结果分发：后端返回批量结果后，遍历队列，将对应子结果依次回调给各请求。
3.3 缓存（Caching）
对于频繁访问但更新不太频繁的资源，还可以引入缓存（内存或外部缓存如 Redis）：
- 先查缓存：若命中，直接返回，短路后端。
- 若未命中，再执行合并或批量调用；并将结果存入缓存。
合理的缓存与合并策略结合，可以进一步削峰。
4. 第一种方案：基于 Promise 的请求合并
4.1 单一资源并发去重
假设我们在 Node.js 中，针对同一 URL /user/:id，可能在短时间内出现多次并发请求（来自不同客户端或前端同一页面多次渲染）。我们希望“同 id 的并发请求只打一次后端接口”。
4.2 实现思路与代码示例
下面以 Express 为例，演示如何在中间层实现并发去重。示例假设后端服务地址为：https://api.example.com/user/:id。
// app.js
import express from 'express';
import fetch from 'node-fetch'; // 或 axios
const app = express();
const PORT = 3000;
/**
* pendingMap 存储当前正在进行的请求
* key: userId
* value: {
* promise: Promise // 正在进行的后端调用 Promise
* resolvers: [] // 其他并发请求的 resolve
* rejecters: [] // 并发请求的 reject
* }
*/
const pendingMap = new Map();
/**
* fetchUserFromBackend：实际调用后端 API
*/
async function fetchUserFromBackend(userId) {
const response = await fetch(`https://api.example.com/user/${userId}`);
if (!response.ok) {
throw new Error(`后端请求失败，状态：${response.status}`);
}
return response.json();
}
/**
* getUser：合并并发请求
*/
function getUser(userId) {
// 如果已有 pending 调用，加入队列，返回同一个 Promise
if (pendingMap.has(userId)) {
return new Promise((resolve, reject) => {
const entry = pendingMap.get(userId);
entry.resolvers.push(resolve);
entry.rejecters.push(reject);
});
}
// 否则，先创建 entry，并发起后端调用
let resolvers = [];
let rejecters = [];
const promise = new Promise(async (resolve, reject) => {
try {
const data = await fetchUserFromBackend(userId);
// 通知所有等待者
resolve(data);
resolvers.forEach(r => r(data));
} catch (err) {
reject(err);
rejecters.forEach(r => r(err));
} finally {
pendingMap.delete(userId);
}
});
// 注册在 map 中
pendingMap.set(userId, {
promise,
resolvers,
rejecters,
});
return promise;
}
/**
* Express 路由
*/
app.get('/user/:id', async (req, res) => {
const userId = req.params.id;
try {
const userData = await getUser(userId);
res.json(userData);
} catch (err) {
res.status(500).json({ error: err.message });
}
});
/**
* 启动服务器
*/
app.listen(PORT, () => {
console.log(`中间层服务启动，端口 ${PORT}`);
});关键点说明：
pendingMap：
用于存储当前正在进行的后端调用。Key 为userId，Value 为一个对象{ promise, resolvers, rejecters }，其中：promise：代表第一次发起的后端调用的 Promise。resolvers：用于存放在调用过程中加入的并发请求的 resolve 回调。rejecters：用于存放并发请求的 reject 回调。
第一次请求时：
pendingMap.has(userId)为false，创建新条目，并发起一次后端调用fetchUserFromBackend(userId)，将其封装为promise。这个
promise内部调用fetchUserFromBackend，并在 resolve/reject 时：- 调用原始的
resolve(data)或reject(err)。 - 遍历
resolvers/rejecters数组，通知并发请求。
- 调用原始的
- 调用完毕后，
finally中pendingMap.delete(userId)，清理 map。
后续并发请求时：
pendingMap.has(userId)为true，直接返回一个新的 Promise，将其 resolve/reject 回调推入正在进行的条目的resolvers/rejecters队列，等待同一次后端返回。
这样便能实现：对于同一 userId，同一时刻不论来多少并发请求，都只发起一次后端请求，并将结果分发给所有等待的请求。
4.3 流程图解
┌────────────────────────────────────────────────────────┐
│ 客户端 A 获取 /user/123 │
│ 客户端 B 获取 /user/123（几乎同时） │
└──────────────┬─────────────────────────────────────────┘
│
▼
┌──────────────────────────────────┐
│ Router: getUser('123') │
│ pendingMap.has('123') == false │
└───────────────┬──────────────────┘
│
▼
┌─────────────────────────────────────┐
│ 第一次调用 fetchUserFromBackend(123)│
└──────────────┬──────────────────────┘
│
▼
┌─────────────┐ ┌──────────────────────┐
│ pendingMap: │<--------│ store entry with │
│ '123' -> {│ promise, │ resolvers=[], │
│ resolvers, rejecters } │ rejecters=[] } │
└─────────────┘ └──────────────────────┘
│
▼
后端调用发起 ──▶ 服务器: 获取 user 123 数据
│
▼
┌───────────────────────────────────────┐
│ 返回用户数据 data │
└───────────────┬───────────────────────┘
│
▼
┌─────────────────────────────────────────┐
│ resolve(data); │
│ 遍历 resolvers 并执行 (当前为空) │
└───────────────┬─────────────────────────┘
│
▼
pendingMap.delete('123')
│
▼
┌──────────────────────────────┐
│ 客户端 A 收到 data，并响应 │
└──────────────────────────────┘
（此时如果 B 来自并发加入，B 会获得存在于 pendingMap 的同一 Promise，并复用返回值） 5. 第二种方案：批量请求（Batching）实现
当合并多个不同资源的请求时，去重不再适用，因为请求针对不同 ID。此时需要批量化（Batching）。
5.1 适用场景与原理
- 示例场景：前端一次性加载页面，需要展示多个用户信息：
/user/1、/user/2、/user/3。若中间层对每个请求单独转发到后端，就要发起 3 次 HTTP 请求。若后端暴露了批量接口/users?ids=1,2,3，中间层可合并为一次请求，批量返回所有用户数据。 原理：
- 请求队列：中间层收到
/user/:id请求时，不立即转发，而先将其缓存到“批量队列”。 - 定时/容量触发：当队列中累积到一定数量（如 10 个），或等待时间超过阈值（如 10ms）时，将队列中的所有 ID 一次性发往后端批量接口。
- 结果分发：批量接口返回结果（一个数组或 map），中间层遍历队列，将对应的子结果发送给每个请求。
- 请求队列：中间层收到
5.2 基于队列与定时器的批量策略
常见策略：
- 固定时间窗 (Time Window)：收到第一个请求后，启动一个定时器（如 10ms）。在这个时间窗内所有新的请求都进入同一个批次。定时器到期后，一并发起批量调用。
- 固定容量 (Capacity Trigger)：当队列长度达到阈值 N（如 50），立即批量发起调用，不再等待时间窗结束。
两者可以结合，取“先到者”：时间窗先到则发起，容量先到也发起。
5.3 代码示例：Express 中间层批量转发
以下示例结合上述两种触发策略，演示如何在 Express 中实现批量请求合并。
// batch-app.js
import express from 'express';
import fetch from 'node-fetch'; // 或 axios
const app = express();
const PORT = 3000;
/**
* 批量队列：存储待合并请求
* queueItems: [{ userId, resolve, reject }, ...]
*/
let queueItems = [];
let timer = null;
const BATCH_SIZE = 5; // 容量阈值
const TIME_WINDOW = 10; // 时间窗：10ms
/**
* batchFetchUsers：一次性调用后端批量接口 /users?ids=...
*/
async function batchFetchUsers(userIds) {
// 真实环境中：`https://api.example.com/users?ids=1,2,3`
const query = userIds.join(',');
const response = await fetch(`https://api.example.com/users?ids=${query}`);
if (!response.ok) throw new Error(`后端批量请求失败：${response.status}`);
const data = await response.json(); // 假设返回 [{id, name, ...}, ...]
// 转为以 id 为 key 的 map，方便查找
const map = new Map(data.map(item => [String(item.id), item]));
return map;
}
/**
* scheduleBatch：调度批量任务
*/
function scheduleBatch() {
if (timer) return; // 已有定时器在等候
timer = setTimeout(async () => {
// 取出当前队列
const items = queueItems;
queueItems = [];
timer = null;
// 提取 userId 列表
const userIds = items.map(item => item.userId);
let resultMap;
try {
resultMap = await batchFetchUsers(userIds);
} catch (err) {
// 后端请求失败，统一 reject
items.forEach(item => item.reject(err));
return;
}
// 根据结果逐一 resolve
items.forEach(item => {
const data = resultMap.get(item.userId);
if (data !== undefined) {
item.resolve(data);
} else {
item.reject(new Error(`未找到用户 ${item.userId}`));
}
});
}, TIME_WINDOW);
}
/**
* getUserBatch：中间层对外接口，返回 Promise
*/
function getUserBatch(userId) {
return new Promise((resolve, reject) => {
queueItems.push({ userId, resolve, reject });
// 若已达容量阈值，立即发起批量
if (queueItems.length >= BATCH_SIZE) {
clearTimeout(timer);
timer = null;
// 立即触发批量
scheduleBatch();
} else {
// 启动定时等待
scheduleBatch();
}
});
}
/**
* Express 路由：对外 /user/:id
*/
app.get('/user/:id', async (req, res) => {
const userId = req.params.id;
try {
const userData = await getUserBatch(userId);
res.json(userData);
} catch (err) {
res.status(500).json({ error: err.message });
}
});
app.listen(PORT, () => {
console.log(`批量合并中间层启动，端口 ${PORT}`);
});说明：
queueItems数组：缓存所有待批量合并的请求条目（包含userId、resolve、reject）。timer定时器：在收到首个请求后启动一个 10ms 定时器，时间窗结束则批量发起后端调用。- 容量触发：若在时间窗内，队列长度达到
BATCH_SIZE（如 5），则立即清除定时器并批量发起。 - 批量调用：使用
batchFetchUsers(userIds)向后端批量接口发起请求，并将返回的数组转换为Map以便快速匹配各个子请求。 - 结果分发：批量返回后，遍历缓存队列，将对应
userId的 data 分发给每个请求的resolve。若后端缺少某个 ID，则对应该请求reject。
5.4 ASCII 批量流转图
客户端 A 请求 /user/1 客户端 B 请求 /user/2
│ │
▼ ▼
┌────────────────┐ ┌────────────────┐
│ Router: getUserBatch(1) │ │ Router: getUserBatch(2) │
└───────┬─────────┘ └───────┬─────────┘
│ │
▼ │
queueItems.push({1, resA}) │
│ │
│───> queueItems = [{1,A}] │
│ │
│ timer 启动 (10ms) │
│ ▼
│ queueItems.push({2, resB})
│ │
│ queueItems = [{1,A},{2,B}]
│ │
│<───── 容量或时间窗触发 ────────┘
▼
清除定时器、复制当前队列到 items
queueItems 清空
提取 userIds = [1,2]
│
▼
┌─────────────────────────────────┐
│ batchFetchUsers([1,2]) │
│ → 统一调用 后端 /users?ids=1,2 │
└─────────────────────────────────┘
│
后端返回 [{id:1,name...},{id:2,name...}]
▼
转为 Map：{ "1": data1, "2": data2 }
│
▼
遍历 items：
├─ item{1,A}.resolve(data1) → 客户端 A 响应
└─ item{2,B}.resolve(data2) → 客户端 B 响应6. 解决并发边界与超时问题
在实际生产环境中，需考虑并发边界、超时控制与错误隔离，避免单一批次或去重逻辑出现不可控延迟。
6.1 Promise 过期与超时控制
若后端接口偶尔出现迟滞或卡死，需要在中间层对单次请求设置超时，避免中间层请求一直挂起，导致后续请求也被阻塞。以下示例展示如何为 getUser 或 getUserBatch 增加超时逻辑。
/**
* 带超时的 fetchWithTimeout
*/
function fetchWithTimeout(url, options = {}, timeoutMs = 500) {
return new Promise((resolve, reject) => {
const timer = setTimeout(() => {
reject(new Error('请求超时'));
}, timeoutMs);
fetch(url, options)
.then(res => {
clearTimeout(timer);
if (!res.ok) {
reject(new Error(`状态码 ${res.status}`));
} else {
resolve(res.json());
}
})
.catch(err => {
clearTimeout(timer);
reject(err);
});
});
}
/**
* 在批量/去重逻辑中使用 fetchWithTimeout
*/
async function safeFetchUser(userId) {
return fetchWithTimeout(`https://api.example.com/user/${userId}`, {}, 300);
}- 如果 300ms 内未收到后端响应，就会
reject(new Error('请求超时'))，触发并发请求的 reject 回调。 - 对批量请求也可采用相似方式，对
batchFetchUsers包装超时：
function batchFetchUsersWithTimeout(userIds, timeoutMs = 500) {
const url = `https://api.example.com/users?ids=${userIds.join(',')}`;
return fetchWithTimeout(url, {}, timeoutMs).then(dataArray => {
const map = new Map(dataArray.map(item => [String(item.id), item]));
return map;
});
}6.2 并发量限制与节流
当中间层本身也成为高并发入口，为避免瞬时“大洪峰”导致 Node.js 进程内存/CPU 突增，可对进入中间层的并发请求量做限制。例如用 p-limit、bottleneck 等库进行并发数控制、节流或排队：
import pLimit from 'p-limit';
const limit = pLimit(50); // 最多并发 50 个批量请求
async function handleUserRequest(userId) {
return limit(() => getUserBatch(userId));
}
app.get('/user/:id', async (req, res) => {
try {
const data = await handleUserRequest(req.params.id);
res.json(data);
} catch (err) {
res.status(500).json({ error: err.message });
}
});- 这样即使瞬时有几千个请求拼到中间层，也只会同时发起 50 个批量任务，其他请求在队列中等待。
6.3 错误处理与降级策略
在高可用设计中，一旦批量或去重逻辑发生错误，需及时隔离故障并给出降级响应。常见策略：
降级为直接转发
如果合并触发错误（如超时），可以退回到“每个请求各自打后端”的简单模式。app.get('/user/:id', async (req, res) => { const userId = req.params.id; try { const data = await getUserBatch(userId); res.json(data); } catch (err) { console.error('合并调用失败，降级为单次请求:', err); // 直接调用后端单个接口 try { const fallbackData = await fetchWithTimeout(`https://api.example.com/user/${userId}`, {}, 500); res.json(fallbackData); } catch (e) { res.status(500).json({ error: '后端不可用' }); } } });快速失败
如果中间层负载过高，直接返回一个 503 或提示客户端稍后重试，避免排队堆积。const QUEUE_MAX = 1000; app.get('/user/:id', async (req, res) => { if (queueItems.length > QUEUE_MAX) { return res.status(503).json({ error: '系统繁忙，请稍后重试' }); } // 继续合并逻辑 ... });- 熔断与限流
配合opossum、breaker等熔断库，对批量调用封装熔断逻辑，当后端错误率或延迟过高时，短路并快速失败或降级。
7. 性能优化与监控
7.1 监控关键指标：QPS、延迟、命中率
在生产环境中，需持续关注以下指标：
- 请求量（QPS）：中间层每秒接入请求量高峰、平均值。
后端调用次数：原始请求数 vs. 合并后实际调用数，可计算合并命中率：
合并命中率 = 1 – （实际后端调用数 / 原始请求数）
响应延迟：
- 中间层延迟：从接收请求到返回数据的时延，包括合并等待、后端调用、内部处理。
- 后端延迟：中间层发起的后端调用耗时。
- 错误率：中间层/后端调用失败率，用于触发熔断或扩容策略。
7.2 日志与指标埋点示例
以下示例展示如何在上述合并/批量逻辑中埋点日志与指标，以便后续用 Prometheus、Grafana 等系统采集并可视化。
import { Counter, Histogram } from 'prom-client';
// Prometheus 监控指标
const batchCounter = new Counter({
name: 'batch_requests_total',
help: '批量请求总数',
labelNames: ['status']
});
const batchLatency = new Histogram({
name: 'batch_request_duration_ms',
help: '批量请求耗时（毫秒）',
buckets: [50, 100, 200, 500, 1000, 2000],
});
// 在 batchFetchUsers 中添加监控
async function batchFetchUsers(userIds) {
const start = Date.now();
try {
const map = await fetchWithTimeout(`https://api.example.com/users?ids=${userIds.join(',')}`, {}, 500);
batchCounter.inc({ status: 'success' });
batchLatency.observe(Date.now() - start);
return map;
} catch (err) {
batchCounter.inc({ status: 'error' });
batchLatency.observe(Date.now() - start);
throw err;
}
}
// 在 Express 中提供 /metrics 端点以供 Prometheus 抓取
import client from 'prom-client';
app.get('/metrics', async (req, res) => {
res.set('Content-Type', client.register.contentType);
res.end(await client.register.metrics());
});batchCounter：统计成功与失败的批量调用次数。batchLatency：分布式直方图，记录每次批量调用耗时。- 通过
/metrics端点，Prometheus 可以周期性抓取并产生监控面板。
7.3 Node.js 性能调优要点
- 避免同步阻塞：所有 I/O 操作必须使用异步 API（
fs.promises、fetch、axios、数据库驱动的异步方法）。 内存管理：
pendingMap、queueItems必须按需清理，防止内存泄漏。- 批量队列长度受限，如前文所示，通过阈值限制队列最大长度。
事件循环负载：
- 过多微任务（
process.nextTick、Promise.then）会让事件循环某阶段饥饿，应节制使用。 - 批量合并的时间窗不宜过长，否则客户端响应时延上升；也不宜过短，否则达不到合并效果。
- 过多微任务（
CPU & 网络带宽：
- 在合并层启用 gzip 压缩传输或轻量序列化。
- 对后端返回数据做简化，只保留必要字段，减少网络传输开销。
扩展性：
- 使用
cluster模式或 Docker/Kubernetes 部署多实例，分担高并发压力。 - 配合负载均衡（如 Nginx、Envoy），将请求均匀分配到不同中间层实例。
- 使用
8. 实战案例：GraphQL DataLoader 与自定义合并
8.1 DataLoader 简介与原理
DataLoader 是 Facebook 出品的一个用于 GraphQL 的批量与缓存库，但其核心原理也可用于普通 REST 中间层的合并：
- 批量函数（Batch Load Function）：收集若干个 load 请求，将其合并成一次批量调用。
- 缓存层：同一请求上下文内对同一 Key 的重复调用只做一次。
- 执行时机：每个事件循环 Tick 结束后，DataLoader 会批量调用一轮。
8.2 自定义 DataLoader 批量实现示例
以下示例展示如何在 Express 中将 DataLoader 与 REST 中间层结合：
// dataloader-app.js
import express from 'express';
import DataLoader from 'dataloader';
import fetch from 'node-fetch';
const app = express();
const PORT = 3000;
/**
* 定义 batchLoadUsers：一次性批量调用后端
*/
async function batchLoadUsers(keys) {
// keys: ['1','2','3']
const query = keys.join(',');
const res = await fetch(`https://api.example.com/users?ids=${query}`);
if (!res.ok) throw new Error('后端批量请求失败');
const dataArray = await res.json(); // [{id,name,...}, ...]
// 构建 Map：key => data
const dataMap = new Map(dataArray.map(item => [String(item.id), item]));
// 根据原始 keys 顺序返回数据，若某个 id 未命中，则返回 null
return keys.map(key => dataMap.get(key) || null);
}
const userLoader = new DataLoader(batchLoadUsers, {
cache: true, // 缓存同一 id 的结果
maxBatchSize: 5 // 最大每批次 5 个
});
app.get('/user/:id', async (req, res) => {
const userId = req.params.id;
try {
const data = await userLoader.load(userId);
if (data) res.json(data);
else res.status(404).json({ error: 'User Not Found' });
} catch (err) {
res.status(500).json({ error: err.message });
}
});
app.listen(PORT, () => {
console.log(`DataLoader 中间层启动，端口 ${PORT}`);
});要点：
- DataLoader 自动批量：在同一 Tick 内多次调用
userLoader.load(id)，DataLoader 会自动聚合为一次batchLoadUsers调用（最多maxBatchSize个）。 - 缓存：相同
id多次load只会触发一次后端调用。 - 调用时机：DataLoader 会在当前事件循环 Tick 的末尾（微任务队列）执行批量函数，确保“短时间合并”效果。
8.3 与 REST 中间层对比
- 本教程前文方案：手动管理队列与定时器，灵活但需要自己处理触发逻辑。
- DataLoader：开箱即用，适合 GraphQL 与简单 REST 批量场景，但对非 GraphQL 场景需要手动在每个请求上下文中新建 Loader，以免跨请求污染缓存。
9. 总结与最佳实践
合理选择合并策略
- 去重：针对同一资源的并发请求，直观且易实现，适合如缓存击穿防护场景。
- 批量：针对多个不同资源（ID 列表）请求，减少后端调用次数，适用于批量接口成熟的后端系统。
- 可将二者结合使用：先去重，再批量，进一步提高命中率与合并效率。
控制批量窗口与容量
- 时间窗不宜过长，否则延迟升高；不宜过短，否则合并效果差。常见取值 5–20ms。
- 容量阈值根据后端吞吐能力与中间层资源情况调优。
超时与错误隔离
- 为单次后端或批量调用设置超时，避免中间层长时间挂起。
- 失败时可降级至“直接转发”或快速失败（503）。
监控与报警
- 对批量调用次数、去重命中率、延迟、错误率等指标做实时监控。
- 一旦批量合并命中率下降或后端延迟飙升，及时告警并扩容或切换策略。
缓存与更新策略
- 对不常变动的资源可加内存或分布式缓存（如 Redis），先查缓存再合并。
- 缓存失效后，需要防止缓存击穿，可结合去重策略。
适时使用成熟库
- 对 GraphQL 场景，可直接使用 DataLoader；对 REST 场景，也可参考其实现原理，或使用
batch-request等社区方案。
- 对 GraphQL 场景，可直接使用 DataLoader；对 REST 场景，也可参考其实现原理，或使用
注意线程安全
- 若使用
cluster或多实例部署，内存级别的合并或缓存只能局限于单实例；跨实例需要使用共享缓存（Redis）或 API 网关方案。
- 若使用
通过以上章节，您已经掌握了 Node.js 中实现请求合并与批量转发的核心技巧：从最基础的并发去重、批量队列+定时器，到加入超时控制、并发限制与监控指标，并且了解了如何结合 DataLoader 等库进行二次开发。希望本文能帮助你在实际项目中打造高性能、可扩展的中间层，提升系统吞吐与稳定性。