引言：Node.js在高并发场景下的优势
在现代Web开发中，实时通信和高并发处理是两个核心挑战。Node.js凭借其事件驱动、非阻塞I/O模型，成为构建这类系统的理想选择。本文将通过两个实战案例——实时聊天室和API网关系统，深入解析如何利用JavaScript和Node.js构建高性能的分布式系统。
为什么选择Node.js？
Node.js的核心优势在于：
单线程事件循环
：避免了多线程编程的复杂性
非阻塞I/O
：能够高效处理大量并发连接
统一技术栈
：前后端使用同一种语言，降低学习成本
丰富的生态系统
：NPM提供了大量高质量的第三方模块
案例一：从零构建高并发实时聊天室
1. 系统架构设计
一个高并发实时聊天室需要解决以下核心问题：
实时消息推送
用户身份管理
消息持久化
水平扩展能力
我们采用以下架构：
客户端(WebSocket) → 聊天服务器(Node.js + Socket.IO) → Redis(消息队列/状态存储) → MongoDB(消息持久化)
2. 环境准备与项目初始化
# 创建项目目录
mkdir realtime-chatroom
cd realtime-chatroom
# 初始化Node.js项目
npm init -y
# 安装核心依赖
npm install express socket.io redis mongoose cors
npm install –save-dev nodemon dotenv
3. 核心服务实现
3.1 WebSocket服务器搭建
// server.js
const express = require(‘express’);
const http = require(‘http’);
const socketIo = require(‘socket.io’);
const redis = require(‘redis’);
const mongoose = require(‘mongoose’);
const cors = require(‘cors’);
require(‘dotenv’).config();
const app = express();
const server = http.createServer(app);
const io = socketIo(server, {
cors: {
origin: “*”,
methods: [“GET”, “POST”]
}
});
// 连接Redis
const redisClient = redis.createClient({
url: process.env.REDIS_URL || ‘redis://localhost:6379’
});
redisClient.on(‘error’, (err) => console.log(‘Redis Client Error’, err));
redisClient.connect();
// 连接MongoDB
mongoose.connect(process.env.MONGODB_URI || ‘mongodb://localhost:27017/chatroom’, {
useNewUrlParser: true,
useUnifiedTopology: true
});
// 消息模型
const messageSchema = new mongoose.Schema({
room: String,
user: String,
content: String,
timestamp: { type: Date, default: Date.now }
});
const Message = mongoose.model(‘Message’, messageSchema);
app.use(cors());
app.use(express.json());
// REST API for message history
app.get(‘/api/messages/:room’, async (req, res) => {
try {
const messages = await Message.find({ room: req.params.room })
.sort({ timestamp: -1 })
.limit(50);
res.json(messages.reverse());
} catch (error) {
res.status(500).json({ error: error.message });
}
});
// Socket.IO 实时通信
io.on(‘connection’, (socket) => {
console.log(`用户连接: ${socket.id}`);
// 加入房间
socket.on(‘joinRoom’, async ({ username, room }) => {
socket.join(room);
// 存储用户状态到Redis
await redisClient.hSet(‘online_users’, socket.id, JSON.stringify({
username,
room,
joinedAt: Date.now()
}));
// 广播用户加入
socket.to(room).emit(‘userJoined’, { username });
// 发送欢迎消息
socket.emit(‘message’, {
user: ‘系统’,
content: `欢迎 ${username} 加入 ${room} 房间！`,
timestamp: new Date()
});
});
// 处理消息
socket.on(‘sendMessage’, async ({ room, user, content }) => {
// 验证用户是否在房间
const userData = await redisClient.hGet(‘online_users’, socket.id);
if (!userData) {
socket.emit(‘error’, { message: ‘请先加入房间’ });
return;
}
// 保存到MongoDB
const message = new Message({ room, user, content });
await message.save();
// 发布到Redis Pub/Sub以便其他服务器实例处理
await redisClient.publish(‘chat_messages’, JSON.stringify({
room,
user,
content,
timestamp: message.timestamp
}));
// 广播给房间内所有用户
io.to(room).emit(‘message’, {
user,
content,
timestamp: message.timestamp
});
});
// 断开连接处理
socket.on(‘disconnect’, async () => {
const userData = await redisClient.hGet(‘online_users’, socket.id);
if (userData) {
const { username, room } = JSON.parse(userData);
await redisClient.hDel(‘online_users’, socket.id);
socket.to(room).emit(‘userLeft’, { username });
}
console.log(`用户断开: ${socket.id}`);
});
});
const PORT = process.env.PORT || 3000;
server.listen(PORT, () => {
console.log(`聊天服务器运行在端口 ${PORT}`);
});
3.2 消息队列消费者（用于多服务器实例）
// messageConsumer.js
const redis = require(‘redis’);
const mongoose = require(‘mongoose’);
// 连接MongoDB
mongoose.connect(process.env.MONGODB_URI || ‘mongodb://localhost:27017/chatroom’, {
useNewUrlParser: true,
useUnifiedTopology: true
});
const messageSchema = new mongoose.Schema({
room: String,
user: String,
content: String,
timestamp: { type: Date, default: Date.now }
});
const Message = mongoose.model(‘Message’, messageSchema);
async function startConsumer() {
const subscriber = redis.createClient({ url: process.env.REDIS_URL });
const publisher = subscriber.duplicate();
await subscriber.connect();
await publisher.connect();
// 订阅消息频道
await subscriber.subscribe(‘chat_messages’, async (message) => {
try {
const data = JSON.parse(message);
// 可以在这里进行额外的处理，如消息审核、统计等
console.log(`处理消息: ${data.user} – ${data.content}`);
// 如果需要，可以发布到其他频道进行进一步处理
await publisher.publish(‘processed_messages’, message);
} catch (error) {
console.error(‘消息处理错误:’, error);
}
});
console.log(‘消息消费者已启动’);
}
startConsumer();
4. 客户端实现
加入聊天室
5. 性能优化与扩展策略
5.1 水平扩展
当单机无法承载更多用户时，可以使用PM2进行进程管理：
# 安装PM2
npm install -g pm2
# 启动多个实例
pm2 start server.js -i 4 –name chat-server
# 保存配置
pm2 save
pm2 startup
5.2 Redis优化配置
// config/redis.js
const redis = require(‘redis’);
const createRedisClient = () => {
const client = redis.createClient({
url: process.env.REDIS_URL,
socket: {
reconnectStrategy: (retries) => {
// 指数退避策略
return Math.min(retries * 100, 3000);
},
connectTimeout: 10000
}
});
client.on(‘error’, (err) => {
console.error(‘Redis错误:’, err);
});
return client;
};
module.exports = { createRedisClient };
5.3 消息压缩
// 消息压缩中间件
const zlib = require(‘zlib’);
function compressMessage(message) {
return new Promise((resolve, reject) => {
zlib.gzip(JSON.stringify(message), (err, result) => {
if (err) reject(err);
else resolve(result.toString(‘base64’));
});
});
}
function decompressMessage(compressed) {
return new Promise((resolve, reject) => {
zlib.gunzip(Buffer.from(compressed, ‘base64’), (err, result) => {
if (err) reject(err);
else resolve(JSON.parse(result.toString()));
});
});
}
6. 安全考虑
// 安全中间件
const rateLimit = require(‘express-rate-limit’);
// 限制IP连接频率
const connectionLimiter = rateLimit({
windowMs: 60 * 1000, // 1分钟
max: 10, // 最多10次连接
message: ‘连接过于频繁，请稍后再试’
});
app.use(‘/socket.io/’, connectionLimiter);
// 消息内容验证
function validateMessage(content) {
// 防止XSS攻击
const clean = content.replace(/
// 限制消息长度
if (clean.length > 500) {
throw new Error(‘消息过长’);
}
// 过滤敏感词
const sensitiveWords = [‘暴力’, ‘色情’, ‘赌博’];
for (const word of sensitiveWords) {
if (clean.includes(word)) {
throw new Error(‘消息包含敏感内容’);
}
}
return clean;
}
案例二：构建高性能API网关系统
1. API网关的核心功能
API网关作为微服务架构的入口，需要提供：
路由转发
：将请求分发到后端服务
负载均衡
：均衡分配流量
认证授权
：统一的安全控制
限流熔断
：保护后端服务
日志监控
：请求追踪和分析
2. 系统架构
客户端 → API网关(Node.js + Express) → 服务发现(Etcd/Consul) → 微服务集群
3. 核心实现
3.1 网关主服务
// gateway.js
const express = require(‘express’);
const httpProxy = require(‘http-proxy’);
const rateLimit = require(‘express-rate-limit’);
const helmet = require(‘helmet’);
const jwt = require(‘jsonwebtoken’);
const Redis = require(‘ioredis’);
const { createClient } = require(‘redis’);
const app = express();
const proxy = httpProxy.createProxyServer({});
// 安全中间件
app.use(helmet());
app.use(express.json());
// Redis客户端（用于限流和缓存）
const redis = new Redis(process.env.REDIS_URL || ‘redis://localhost:6379’);
// 服务注册表（实际项目中使用Etcd或Consul）
const serviceRegistry = {
‘user-service’: [‘http://localhost:3001’, ‘http://localhost:3002’],
‘order-service’: [‘http://localhost:3003’],
‘product-service’: [‘http://localhost:3004’, ‘http://localhost:3005’, ‘http://localhost:3006’]
};
// 负载均衡器
class LoadBalancer {
constructor(servers) {
this.servers = servers;
this.currentIndex = 0;
}
getNext() {
const server = this.servers[this.currentIndex];
this.currentIndex = (this.currentIndex + 1) % this.servers.length;
return server;
}
// 健康检查
async healthCheck() {
const healthyServers = [];
for (const server of this.servers) {
try {
const response = await fetch(`${server}/health`, { timeout: 2000 });
if (response.ok) {
healthyServers.push(server);
}
} catch (error) {
console.warn(`服务器 ${server} 不健康:`, error.message);
}
}
this.servers = healthyServers.length > 0 ? healthyServers : this.servers;
return this.servers;
}
}
// 初始化负载均衡器
const loadBalancers = {};
Object.keys(serviceRegistry).forEach(service => {
loadBalancers[service] = new LoadBalancer(serviceRegistry[service]);
});
// 认证中间件
const authenticate = async (req, res, next) => {
const token = req.headers.authorization?.split(‘ ‘)[1];
if (!token) {
return res.status(401).json({ error: ‘缺少认证令牌’ });
}
try {
// 验证JWT
const decoded = jwt.verify(token, process.env.JWT_SECRET);
// 检查Redis中的令牌黑名单
const blacklisted = await redis.get(`blacklist:${token}`);
if (blacklisted) {
return res.status(401).json({ error: ‘令牌已失效’ });
}
req.user = decoded;
next();
} catch (error) {
return res.status(401).json({ error: ‘无效的令牌’ });
}
};
// 限流中间件
const createRateLimiter = (windowMs, maxRequests) => {
return rateLimit({
windowMs,
max: maxRequests,
keyGenerator: (req) => {
// 基于用户ID或IP限流
return req.user?.userId || req.ip;
},
handler: (req, res) => {
res.status(429).json({
error: ‘请求过于频繁’,
retryAfter: Math.ceil(windowMs / 1000)
});
},
store: new (require(‘rate-limit-redis’))({
client: redis,
prefix: ‘rl:’
})
});
};
// 路由配置
const routes = [
{
path: ‘/api/users’,
service: ‘user-service’,
auth: true,
rateLimit: createRateLimiter(60000, 100) // 每分钟100次
},
{
path: ‘/api/orders’,
service: ‘order-service’,
auth: true,
rateLimit: createRateLimiter(60000, 50)
},
{
path: ‘/api/products’,
service: ‘product-service’,
auth: false,
rateLimit: createRateLimiter(60000, 200)
}
];
// 动态路由处理
app.use(‘/api/:service/*’, async (req, res) => {
const serviceName = req.params.service;
const route = routes.find(r => r.path.includes(serviceName));
if (!route) {
return res.status(404).json({ error: ‘服务不存在’ });
}
// 认证检查
if (route.auth) {
await authenticate(req, res, () => {});
if (res.headersSent) return; // 认证失败
}
// 限流检查
if (route.rateLimit) {
await route.rateLimit(req, res, () => {});
if (res.headersSent) return; // 触发限流
}
// 获取健康的服务实例
const lb = loadBalancers[serviceName];
await lb.healthCheck();
if (lb.servers.length === 0) {
return res.status(503).json({ error: ‘服务暂时不可用’ });
}
const target = lb.getNext();
// 日志记录
const logEntry = {
timestamp: new Date().toISOString(),
method: req.method,
path: req.path,
user: req.user?.userId,
ip: req.ip,
target
};
// 异步写入日志（不阻塞请求）
redis.lpush(‘api_logs’, JSON.stringify(logEntry));
// 代理转发
proxy.web(req, res, {
target,
changeOrigin: true,
timeout: 5000,
proxyTimeout: 5000
}, (err) => {
console.error(`代理错误: ${req.path} -> ${target}`, err);
if (!res.headersSent) {
res.status(502).json({ error: ‘网关错误’, details: err.message });
}
});
});
// 错误处理
proxy.on(‘error’, (err, req, res) => {
console.error(‘代理服务器错误:’, err);
if (!res.headersSent) {
res.status(500).json({ error: ‘代理错误’ });
}
});
// 健康检查端点
app.get(‘/health’, (req, res) => {
res.json({ status: ‘healthy’, timestamp: new Date().toISOString() });
});
// 统计信息
app.get(‘/metrics’, async (req, res) => {
const totalRequests = await redis.get(‘total_requests’) || 0;
const activeConnections = await redis.scard(‘active_connections’);
res.json({
totalRequests,
activeConnections,
services: Object.keys(loadBalancers).map(service => ({
name: service,
servers: loadBalancers[service].servers
}))
});
});
const PORT = process.env.GATEWAY_PORT || 8080;
app.listen(PORT, () => {
console.log(`API网关运行在端口 ${PORT}`);
});
3.2 服务发现与注册
// service-registry.js
const Redis = require(‘ioredis’);
const crypto = require(‘crypto’);
class ServiceRegistry {
constructor(redisUrl) {
this.redis = new Redis(redisUrl);
this.heartbeatInterval = 10000; // 10秒心跳
}
// 服务注册
async register(serviceName, instanceId, address, metadata = {}) {
const key = `service:${serviceName}:${instanceId}`;
const data = {
address,
metadata,
lastHeartbeat: Date.now(),
status: ‘healthy’
};
await this.redis.hset(key, data);
// 设置过期时间（自动清理失效服务）
await this.redis.expire(key, 30);
// 加入服务列表
await this.redis.sadd(`services:${serviceName}`, instanceId);
console.log(`服务注册: ${serviceName} – ${instanceId} at ${address}`);
}
// 服务注销
async deregister(serviceName, instanceId) {
const key = `service:${serviceName}:${instanceId}`;
await this.redis.del(key);
await this.redis.srem(`services:${serviceName}`, instanceId);
console.log(`服务注销: ${serviceName} – ${instanceId}`);
}
// 心跳维持
async heartbeat(serviceName, instanceId) {
const key = `service:${serviceName}:${instanceId}`;
await this.redis.hset(key, ‘lastHeartbeat’, Date.now());
await this.redis.expire(key, 30); // 重置过期时间
}
// 获取健康的服务实例
async getHealthyInstances(serviceName) {
const instances = await this.redis.smembers(`services:${serviceName}`);
const healthyInstances = [];
for (const instanceId of instances) {
const key = `service:${serviceName}:${instanceId}`;
const data = await this.redis.hgetall(key);
if (data.status === ‘healthy’ &&
(Date.now() – parseInt(data.lastHeartbeat)) < 30000) {
healthyInstances.push({
id: instanceId,
address: data.address,
metadata: JSON.parse(data.metadata || '{}')
});
}
}
return healthyInstances;
}
// 自动清理失效服务
async startCleanup() {
setInterval(async () => {
const serviceKeys = await this.redis.keys(‘service:*’);
const now = Date.now();
for (const key of serviceKeys) {
const lastHeartbeat = await this.redis.hget(key, ‘lastHeartbeat’);
if (lastHeartbeat && (now – parseInt(lastHeartbeat)) > 30000) {
const [, serviceName, instanceId] = key.split(‘:’);
await this.deregister(serviceName, instanceId);
}
}
}, this.heartbeatInterval);
}
}
// 使用示例
const registry = new ServiceRegistry(process.env.REDIS_URL);
// 微服务启动时注册
async function startService(serviceName, port) {
const instanceId = crypto.randomUUID();
const address = `http://localhost:${port}`;
await registry.register(serviceName, instanceId, address, {
version: ‘1.0.0’,
region: ‘us-east-1’
});
// 启动心跳
const heartbeatTimer = setInterval(() => {
registry.heartbeat(serviceName, instanceId);
}, 5000);
// 优雅关闭
process.on(‘SIGTERM’, async () => {
clearInterval(heartbeatTimer);
await registry.deregister(serviceName, instanceId);
process.exit(0);
});
return instanceId;
}
module.exports = { ServiceRegistry, startService };
3.3 熔断器实现
// circuit-breaker.js
class CircuitBreaker {
constructor(serviceName, options = {}) {
this.serviceName = serviceName;
this.failureThreshold = options.failureThreshold || 5; // 5次失败
this.successThreshold = options.successThreshold || 3; // 3次成功恢复
this.timeout = options.timeout || 60000; // 60秒半开状态
this.resetTimeout = options.resetTimeout || 30000; // 30秒后尝试恢复
this.state = ‘CLOSED’; // CLOSED, OPEN, HALF_OPEN
this.failures = 0;
this.successes = 0;
this.nextAttempt = Date.now();
}
async execute(fn) {
if (this.state === ‘OPEN’) {
if (Date.now() < this.nextAttempt) {
throw new Error(`Circuit breaker is OPEN for ${this.serviceName}`);
}
// 进入半开状态
this.state = 'HALF_OPEN';
this.successes = 0;
this.failures = 0;
}
try {
const result = await fn();
this.onSuccess();
return result;
} catch (error) {
this.onFailure();
throw error;
}
}
onSuccess() {
if (this.state === 'HALF_OPEN') {
this.successes++;
if (this.successes >= this.successThreshold) {
this.state = ‘CLOSED’;
this.failures = 0;
console.log(`Circuit breaker ${this.serviceName} CLOSED`);
}
} else if (this.state === ‘CLOSED’) {
this.failures = 0; // 重置失败计数
}
}
onFailure() {
this.failures++;
if (this.state === ‘HALF_OPEN’ || this.failures >= this.failureThreshold) {
this.state = ‘OPEN’;
this.nextAttempt = Date.now() + this.resetTimeout;
console.warn(`Circuit breaker ${this.serviceName} OPENED`);
}
}
getState() {
return {
serviceName: this.serviceName,
state: this.state,
failures: this.failures,
successes: this.successes,
nextAttempt: this.nextAttempt
};
}
}
// 使用示例
const breakers = {};
function getCircuitBreaker(serviceName) {
if (!breakers[serviceName]) {
breakers[serviceName] = new CircuitBreaker(serviceName);
}
return breakers[serviceName];
}
// 在网关中使用
async function proxyWithCircuitBreaker(req, res, target, serviceName) {
const breaker = getCircuitBreaker(serviceName);
try {
const result = await breaker.execute(() => {
return new Promise((resolve, reject) => {
proxy.web(req, res, { target, changeOrigin: true }, (err) => {
if (err) reject(err);
else resolve();
});
});
});
return result;
} catch (error) {
if (breaker.state === ‘OPEN’) {
return res.status(503).json({
error: ‘服务暂时不可用’,
retryAfter: Math.ceil((breaker.nextAttempt – Date.now()) / 1000)
});
}
throw error;
}
}
4. 网关监控与日志
// monitoring.js
const { createLogger, format, transports } = require(‘winston’);
const Redis = require(‘ioredis’);
// Winston日志配置
const logger = createLogger({
level: ‘info’,
format: format.combine(
format.timestamp(),
format.errors({ stack: true }),
format.json()
),
transports: [
new transports.File({ filename: ‘error.log’, level: ‘error’ }),
new transports.File({ filename: ‘combined.log’ })
]
});
if (process.env.NODE_ENV !== ‘production’) {
logger.add(new transports.Console({
format: format.combine(
format.colorize(),
format.simple()
)
}));
}
// 实时监控
class GatewayMonitor {
constructor(redisUrl) {
this.redis = new Redis(redisUrl);
this.metrics = {
totalRequests: 0,
errors: 0,
latency: []
};
}
async recordRequest(service, path, status, latency) {
const timestamp = Date.now();
// 更新计数器
await this.redis.incr(‘total_requests’);
if (status >= 400) {
await this.redis.incr(‘errors’);
}
// 记录延迟
await this.redis.lpush(‘latency’, latency);
await this.redis.ltrim(‘latency’, 0, 999); // 只保留最近1000条
// 按服务统计
await this.redis.zincrby(`service:${service}:requests`, 1, timestamp);
// 记录日志
logger.info(‘API请求’, { service, path, status, latency });
}
async getMetrics() {
const totalRequests = await this.redis.get(‘total_requests’) || 0;
const errors = await this.redis.get(‘errors’) || 0;
const latencyData = await this.redis.lrange(‘latency’, 0, -1);
const latencies = latencyData.map(l => parseInt(l)).sort((a, b) => a – b);
const p50 = latencies[Math.floor(latencies.length * 0.5)] || 0;
const p95 = latencies[Math.floor(latencies.length * 0.95)] || 0;
const p99 = latencies[Math.floor(latencies.length * 0.99)] || 0;
return {
totalRequests: parseInt(totalRequests),
errors: parseInt(errors),
errorRate: totalRequests > 0 ? (errors / totalRequests * 100).toFixed(2) + ‘%’ : ‘0%’,
latency: {
p50: p50 + ‘ms’,
p95: p95 + ‘ms’,
p99: p99 + ‘ms’
}
};
}
// 告警检查
async checkAlerts() {
const metrics = await this.getMetrics();
// 错误率超过5%触发告警
if (metrics.errorRate > 5) {
logger.warn(‘高错误率告警’, metrics);
// 这里可以集成钉钉、Slack等通知
}
// P99延迟超过1秒触发告警
const p99 = parseInt(metrics.latency.p99);
if (p99 > 1000) {
logger.warn(‘高延迟告警’, metrics);
}
}
}
module.exports = { logger, GatewayMonitor };
高并发优化策略详解
1. Node.js事件循环优化
1.1 避免阻塞操作
// ❌ 错误示例：阻塞操作
function blockingOperation() {
const start = Date.now();
while (Date.now() – start < 100) {
// 阻塞100ms
}
}
// ✅ 正确示例：异步化
async function nonBlockingOperation() {
return new Promise(resolve => {
setTimeout(resolve, 100);
});
}
// ✅ 使用worker threads处理CPU密集型任务
const { Worker } = require(‘worker_threads’);
function heavyComputation(data) {
return new Promise((resolve, reject) => {
const worker = new Worker(‘./heavy-worker.js’, {
workerData: data
});
worker.on(‘message’, resolve);
worker.on(‘error’, reject);
worker.on(‘exit’, (code) => {
if (code !== 0) {
reject(new Error(`Worker stopped with exit code ${code}`));
}
});
});
}
// heavy-worker.js
const { parentPort, workerData } = require(‘worker_threads’);
// 执行CPU密集型计算
function fibonacci(n) {
if (n <= 1) return n;
return fibonacci(n - 1) + fibonacci(n - 2);
}
const result = fibonacci(workerData);
parentPort.postMessage(result);
1.2 事件循环监控
// 监控事件循环延迟
const { monitorEventLoopDelay } = require('perf_hooks');
const histogram = monitorEventLoopDelay({ resolution: 20 });
histogram.enable();
setInterval(() => {
const p99 = histogram.percentile(99);
const max = histogram.max;
if (p99 > 100) {
console.warn(`事件循环延迟过高: p99=${p99}ms, max=${max}ms`);
}
// 重置直方图
histogram.reset();
}, 5000);
2. 内存管理优化
// 内存泄漏检测
const v8 = require(‘v8’);
function monitorMemory() {
const stats = v8.getHeapStatistics();
// 使用率超过80%触发GC
const usage = (stats.used_heap_size / stats.heap_size_limit) * 100;
if (usage > 80) {
if (global.gc) {
global.gc(); // 手动触发GC
console.log(‘手动触发GC’);
}
}
// 记录内存使用
console.log(`内存使用: ${(stats.used_heap_size / 1024 / 1024).toFixed(2)}MB`);
}
// 每30秒检查一次
setInterval(monitorMemory, 30000);
// 启动时需要添加 –expose-gc 标志
// node –expose-gc server.js
3. 连接池优化
// 数据库连接池配置
const mysql = require(‘mysql2/promise’);
const pool = mysql.createPool({
host: ‘localhost’,
user: ‘root’,
database: ‘chatroom’,
waitForConnections: true,
connectionLimit: 100, // 根据服务器配置调整
queueLimit: 0,
enableKeepAlive: true,
keepAliveInitialDelay: 0,
// 连接超时设置
acquireTimeout: 60000,
timeout: 60000
});
// 连接池监控
pool.on(‘acquire’, (connection) => {
console.log(‘获取连接:’, connection.threadId);
});
pool.on(‘release’, (connection) => {
console.log(‘释放连接:’, connection.threadId);
});
// 使用示例
async function getUser(userId) {
const [rows] = await pool.execute(‘SELECT * FROM users WHERE id = ?’, [userId]);
return rows[0];
}
部署与运维
1. Docker部署
# Dockerfile
FROM node:18-alpine
WORKDIR /app
# 复制package.json并安装依赖
COPY package*.json ./
RUN npm ci –only=production
# 复制源代码
COPY . .
# 创建非root用户
RUN addgroup -g 1001 -S nodejs
RUN adduser -S nodeuser -u 1001
# 更改文件所有者
RUN chown -R nodeuser:nodejs /app
USER nodeuser
# 暴露端口
EXPOSE 3000
# 健康检查
HEALTHCHECK –interval=30s –timeout=3s \
CMD node healthcheck.js || exit 1
# 启动命令
CMD [“node”, “server.js”]
2. Docker Compose配置
# docker-compose.yml
version: ‘3.8’
services:
redis:
image: redis:7-alpine
ports:
– “6379:6379”
volumes:
– redis_data:/data
command: redis-server –appendonly yes
healthcheck:
test: [“CMD”, “redis-cli”, “ping”]
interval: 10s
timeout: 3s
retries: 3
mongodb:
image: mongo:6
ports:
– “27017:27017”
volumes:
– mongodb_data:/data/db
environment:
MONGO_INITDB_ROOT_USERNAME: root
MONGO_INITDB_ROOT_PASSWORD: password
healthcheck:
test: echo ‘db.runCommand(“ping”).ok’ | mongosh localhost:27017/test –quiet
interval: 10s
timeout: 3s
retries: 3
chat-server:
build: .
ports:
– “3000:3000”
environment:
– NODE_ENV=production
– REDIS_URL=redis://redis:6379
– MONGODB_URI=mongodb://root:password@mongodb:27017/chatroom
depends_on:
redis:
condition: service_healthy
mongodb:
condition: service_healthy
deploy:
replicas: 3
resources:
limits:
cpus: ‘1’
memory: 512M
reservations:
cpus: ‘0.5’
memory: 256M
healthcheck:
test: [“CMD”, “node”, “healthcheck.js”]
interval: 30s
timeout: 10s
retries: 3
api-gateway:
build: .
command: node gateway.js
ports:
– “8080:8080”
environment:
– NODE_ENV=production
– REDIS_URL=redis://redis:6379
– JWT_SECRET=your-secret-key
depends_on:
– redis
deploy:
replicas: 2
resources:
limits:
cpus: ‘0.5’
memory: 256M
volumes:
redis_data:
mongodb_data:
3. PM2配置
// ecosystem.config.js
module.exports = {
apps: [
{
name: ‘chat-server’,
script: ‘./server.js’,
instances: ‘max’, // 根据CPU核心数自动调整
exec_mode: ‘cluster’,
env: {
NODE_ENV: ‘production’,
REDIS_URL: ‘redis://localhost:6379’,
MONGODB_URI: ‘mongodb://localhost:27017/chatroom’
},
error_file: ‘./logs/err.log’,
out_file: ‘./logs/out.log’,
log_file: ‘./logs/combined.log’,
time: true,
max_memory_restart: ‘500M’,
listen_timeout: 5000,
kill_timeout: 5000,
wait_ready: true,
max_restarts: 10,
min_uptime: ’10s’
},
{
name: ‘api-gateway’,
script: ‘./gateway.js’,
instances: 2,
exec_mode: ‘cluster’,
env: {
NODE_ENV: ‘production’,
REDIS_URL: ‘redis://localhost:6379’,
JWT_SECRET: ‘your-secret-key’
},
error_file: ‘./logs/gateway-err.log’,
out_file: ‘./logs/gateway-out.log’,
max_memory_restart: ‘300M’
}
]
};
性能测试与基准
1. 使用Artillery进行压力测试
# artillery-config.yml
config:
target: ‘http://localhost:3000’
phases:
– duration: 60
arrivalRate: 10
name: Warm up
– duration: 120
arrivalRate: 50
name: Load test
– duration: 60
arrivalRate: 100
name: Stress test
plugins:
ensure: {}
ensure:
thresholds:
– “http.response_time.p99”: 500
– “http.response_time.p95”: 200
– “http.response_time.p50”: 100
scenarios:
– name: “Chat room test”
flow:
– get:
url: “/api/messages/general”
– think: 1
– post:
url: “/api/messages”
json:
room: “general”
user: “testuser”
content: “Hello World”
capture:
– json: “$.id”
as: “messageId”
运行测试：
npm install -g artillery
artillery run artillery-config.yml
2. 使用autocannon进行快速基准测试
npm install -g autocannon
# 测试聊天服务器
autocannon -c 100 -d 30 -p 10 http://localhost:3000/api/messages/general
# 测试API网关
autocannon -c 500 -d 60 http://localhost:8080/api/products
总结
通过本文的两个实战案例，我们详细解析了如何使用JavaScript和Node.js构建高并发的实时系统和API网关。关键要点包括：
架构设计
：合理使用Redis、MongoDB等组件，采用微服务架构
性能优化
：事件循环监控、内存管理、连接池优化
高可用性
：负载均衡、熔断器、健康检查
安全性
：认证授权、限流、输入验证
可扩展性
：水平扩展、服务发现、容器化部署
这些技术和架构模式可以组合使用，构建出能够承载百万级并发的生产级系统。在实际项目中，还需要根据具体业务场景进行调整和优化。