本文目录导读：
Node.js Express 风格中间件
// 中间件管理器
class MiddlewareManager {
constructor() {
this.middlewares = [];
}
// 注册中间件
use(fn) {
this.middlewares.push(fn);
return this;
}
// 执行中间件链
async execute(context, callback) {
let index = -1;
const next = async () => {
index++;
if (index < this.middlewares.length) {
await this.middlewares[index](context, next);
} else if (callback) {
await callback(context);
}
};
await next();
}
}
// 使用示例
const app = new MiddlewareManager();
// 日志中间件
app.use(async (ctx, next) => {
console.log(`[${new Date().toISOString()}] ${ctx.method} ${ctx.path}`);
await next();
});
// 认证中间件
app.use(async (ctx, next) => {
if (ctx.token === 'valid-token') {
ctx.user = { name: 'User' };
await next();
} else {
ctx.error = 'Unauthorized';
}
});
// 路由中间件
app.use(async (ctx, next) => {
if (ctx.path === '/api/data') {
ctx.body = { data: ['item1', 'item2'] };
} else {
await next();
}
});
// 测试
const context = {
method: 'GET',
path: '/api/data',
token: 'valid-token'
};
app.execute(context, (ctx) => {
if (ctx.error) {
console.log('Error:', ctx.error);
} else {
console.log('Response:', ctx.body);
}
});
Python Flask 风格中间件
from functools import wraps
import time
class MiddlewareManager:
def __init__(self):
self.middlewares = []
self.handlers = {}
def use(self, fn):
"""注册中间件"""
self.middlewares.append(fn)
return self
def route(self, path, method='GET'):
"""注册路由处理函数"""
def decorator(fn):
key = f"{method}:{path}"
self.handlers[key] = fn
return fn
return decorator
def handle(self, request):
"""处理请求"""
# 构建中间件链
context = {
'request': request,
'response': None,
'error': None
}
def run_middleware(index):
if index < len(self.middlewares):
return self.middlewares[index](context, lambda: run_middleware(index + 1))
else:
# 执行实际路由处理
key = f"{request['method']}:{request['path']}"
handler = self.handlers.get(key)
if handler:
context['response'] = handler(request)
else:
context['response'] = {'error': 'Not Found', 'status': 404}
run_middleware(0)
return context['response']
# 使用示例
app = MiddlewareManager()
# 日志中间件
@app.use
def logging_middleware(ctx, next):
start = time.time()
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {ctx['request']['method']} {ctx['request']['path']}")
result = next()
duration = time.time() - start
print(f"Processed in {duration:.3f}s")
return result
# 认证中间件
@app.use
def auth_middleware(ctx, next):
headers = ctx['request'].get('headers', {})
if headers.get('Authorization') == 'Bearer valid-token':
ctx['user'] = {'username': 'demo'}
return next()
else:
ctx['response'] = {'error': 'Unauthorized', 'status': 401}
return ctx['response']
# 路由定义
@app.route('/api/hello', 'GET')
def hello(request):
return {'message': 'Hello World!'}
@app.route('/api/user', 'GET')
def get_user(request):
return {'username': 'demo', 'role': 'admin'}
# 测试
request1 = {
'method': 'GET',
'path': '/api/hello',
'headers': {'Authorization': 'Bearer valid-token'}
}
response = app.handle(request1)
print("Response:", response)
简单的中间件模式（通用）
class SimpleMiddleware:
"""通用中间件模式"""
def __init__(self):
self.chain = []
def add(self, middleware):
"""添加中间件"""
self.chain.append(middleware)
return self
def process(self, data, context=None):
"""处理数据"""
if context is None:
context = {}
def execute(index):
if index < len(self.chain):
middleware = self.chain[index]
return middleware(data, context, lambda: execute(index + 1))
return data
return execute(0)
# 定义中间件函数
def validator(data, context, next):
"""验证中间件"""
if not data.get('name'):
raise ValueError("Name is required")
print(f"Validation passed for: {data.get('name')}")
return next()
def transformer(data, context, next):
"""转换中间件"""
data['processed'] = True
data['timestamp'] = __import__('time').time()
print(f"Data transformed")
return next()
def logger(data, context, next):
"""日志中间件"""
print(f"Logging: {data}")
result = next()
print(f"Result: {result}")
return result
# 使用示例
pipeline = SimpleMiddleware()
pipeline.add(validator)
pipeline.add(transformer)
pipeline.add(logger)
# 处理数据
try:
input_data = {'name': 'Test', 'value': 123}
result = pipeline.process(input_data)
print("Final result:", result)
except Exception as e:
print("Error:", e)
前端 JavaScript 中间件（Web应用）
// 前端中间件管理器
class FrontendMiddleware {
constructor() {
this.middlewares = [];
}
// 注册中间件
use(fn) {
this.middlewares.push(fn);
return this;
}
// 执行中间件链
async execute(data) {
let index = -1;
const context = { data, state: {} };
const dispatch = async () => {
index++;
if (index < this.middlewares.length) {
await this.middlewares[index](context, () => dispatch());
}
};
await dispatch();
return context;
}
}
// 使用示例
const dataPipeline = new FrontendMiddleware();
// 数据清洗中间件
dataPipeline.use(async (ctx, next) => {
console.log('Step 1: Data cleaning');
ctx.data.email = ctx.data.email.toLowerCase().trim();
await next();
});
// 数据校验中间件
dataPipeline.use(async (ctx, next) => {
console.log('Step 2: Validation');
if (!ctx.data.email.includes('@')) {
throw new Error('Invalid email format');
}
await next();
});
// 数据处理中间件
dataPipeline.use(async (ctx, next) => {
console.log('Step 3: Processing');
ctx.state.processed = true;
ctx.data.processedAt = new Date().toISOString();
await next();
});
// 测试
async function test() {
const input = {
email: ' USER@EXAMPLE.COM ',
name: 'Test User'
};
try {
const result = await dataPipeline.execute(input);
console.log('Final result:', result.data);
console.log('State:', result.state);
} catch (error) {
console.error('Pipeline error:', error.message);
}
}
test();
这些实现展示了中间件模式的核心思想：链式处理、上下文传递、控制流管理,你可以根据具体需求选择最适合的实现方式。
