怎样实现中间件简易脚本

本文目录导读：

1. Node.js Express 风格中间件
2. Python Flask 风格中间件
3. 简单的中间件模式（通用）
4. 前端 JavaScript 中间件（Web应用）

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();

这些实现展示了中间件模式的核心思想：链式处理、上下文传递、控制流管理,你可以根据具体需求选择最适合的实现方式。