# Python 2 原始代码 import urllib2 from ConfigParser import SafeConfigParser class MyHandler: def __init__(self, config_path): self.parser = SafeConfigParser() self.parser.read(config_path) self.timeout = self.parser.getint('DEFAULT', 'timeout') def fetch_data(self, url): try: req = urllib2.Request(url) response = urllib2.urlopen(req, timeout=self.timeout) data = response.read() return data except urllib2.URLError, e: print "Network error: %s" % e.reason return None def process_items(self, items): results = [] for item in items: if item['status'] == 'active': results.append(item['name'].decode('utf-8')) return results

# Python 3 迁移后代码 import urllib.request import urllib.error from configparser import SafeConfigParser class MyHandler: def __init__(self, config_path): self.parser = SafeConfigParser() self.parser.read(config_path) self.timeout = self.parser.getint('DEFAULT', 'timeout') def fetch_data(self, url): try: req = urllib.request.Request(url) response = urllib.request.urlopen(req, timeout=self.timeout) data = response.read() return data except urllib.error.URLError as e: print(f"Network error: {e.reason}") return None def process_items(self, items): results = [] for item in items: if item['status'] == 'active': results.append(item['name']) return results

// JavaScript 原始代码 function calculateDiscount(price, customer) { if (customer.isPremium) { if (customer.years > 5) { return price * 0.2; } return price * 0.1; } if (price > 1000) { return price * 0.05; } return 0; } function processOrder(items, customer) { let total = 0; let discounts = 0; for (let i = 0; i < items.length; i++) { total += items[i].price * items[i].quantity; discounts += calculateDiscount(items[i].price, customer); } return { subtotal: total, discount: discounts, total: total - discounts }; }

// TypeScript 迁移后代码 interface Customer { isPremium: boolean; years: number; tier?: 'gold' | 'silver' | 'bronze'; } interface OrderItem { id: string; price: number; quantity: number; category: string; } interface OrderResult { subtotal: number; discount: number; total: number; } function calculateDiscount(price: number, customer: Customer): number { if (customer.isPremium) { return customer.years > 5 ? price * 0.2 : price * 0.1; } return price > 1000 ? price * 0.05 : 0; } function processOrder(items: OrderItem[], customer: Customer): OrderResult { const { subtotal, discounts } = items.reduce( (acc, item) => ({ subtotal: acc.subtotal + item.price * item.quantity, discounts: acc.discounts + calculateDiscount(item.price, customer) }), { subtotal: 0, discounts: 0 } ); return { subtotal, discount: discounts, total: subtotal - discounts }; }

# CoffeeScript 原始代码 class ShoppingCart constructor: (@items = []) -> addItem: (product, quantity = 1) -> existing = @items.filter((i) -> i.product.id == product.id)[0] if existing existing.quantity += quantity else @items.push { product, quantity } getTotal: -> @items.reduce ((sum, item) -> sum + item.product.price * item.quantity), 0 applyCoupon: (coupon) -> if coupon.isValid() and coupon.minPurchase <= @getTotal() @items.forEach (item) -> item.discount = coupon.calculateDiscount(item.product.price) return true false

// JavaScript (ES6+) 迁移后代码 class ShoppingCart { constructor(items = []) { this.items = items; } addItem(product, quantity = 1) { const existing = this.items.find(i => i.product.id === product.id); if (existing) { existing.quantity += quantity; } else { this.items.push({ product, quantity }); } } getTotal() { return this.items.reduce( (sum, item) => sum + item.product.price * item.quantity, 0 ); } applyCoupon(coupon) { if (coupon.isValid() && coupon.minPurchase <= this.getTotal()) { this.items.forEach(item => { item.discount = coupon.calculateDiscount(item.product.price); }); return true; } return false; } }

// Java 原始代码 public class UserService { private final UserRepository repository; private final EmailService emailService; public UserService(UserRepository repository, EmailService emailService) { this.repository = repository; this.emailService = emailService; } public Optional findUserById(Long id) { return repository.findById(id); } public User createUser(CreateUserRequest request) { User user = new User(); user.setName(request.getName()); user.setEmail(request.getEmail()); user.setRole(request.getRole() != null ? request.getRole() : Role.USER); user.setCreatedAt(LocalDateTime.now()); user.setActive(true); User saved = repository.save(user); emailService.sendWelcomeEmail(saved.getEmail(), saved.getName()); return saved; } public List searchUsers(String query, Pageable pageable) { return repository.findByNameContainingOrEmailContaining(query, query, pageable); } }

// Kotlin 迁移后代码 class UserService( private val repository: UserRepository, private val emailService: EmailService ) { fun findUserById(id: Long): Optional = repository.findById(id) fun createUser(request: CreateUserRequest): User { val user = User().apply { name = request.name email = request.email role = request.role ?: Role.USER createdAt = LocalDateTime.now() isActive = true } return repository.save(user).also { emailService.sendWelcomeEmail(it.email, it.name) } } fun searchUsers(query: String, pageable: Pageable): List = repository.findByNameContainingOrEmailContaining(query, query, pageable) }

// React 15 组件（class组件与旧生命周期） import React from 'react'; import PropTypes from 'prop-types'; class UserProfile extends React.Component { constructor(props) { super(props); this.state = { user: null, loading: true }; } componentWillMount() { this.fetchUser(this.props.userId); } componentWillReceiveProps(nextProps) { if (nextProps.userId !== this.props.userId) { this.setState({ loading: true }); this.fetchUser(nextProps.userId); } } fetchUser(userId) { fetch(`/api/users/${userId}`) .then(res => res.json()) .then(user => this.setState({ user, loading: false })); } render() { if (this.state.loading) return ; return ( ); } componentWillUnmount() { this.abortController?.abort(); } } UserProfile.propTypes = { userId: PropTypes.number.isRequired };

// React 18 迁移后组件（函数组件 + Hooks） import React, { useState, useEffect } from 'react'; interface UserProfileProps { userId: number; } const UserProfile: React.FC = ({ userId }) => { const [user, setUser] = useState(null); const [loading, setLoading] = useState(true); useEffect(() => { const abortController = new AbortController(); setLoading(true); fetch(`/api/users/${userId}`, { signal: abortController.signal }) .then(res => res.json()) .then(data => { setUser(data); setLoading(false); }) .catch(err => { if (err.name !== 'AbortError') { setLoading(false); } }); return () => abortController.abort(); }, [userId]); // React 18: 使用 useDeferredValue 优化渲染性能 const deferredLoading = React.useDeferredValue(loading); // React 18: 使用 useTransition 处理非紧急状态更新 const [isPending, startTransition] = React.useTransition(); if (deferredLoading) return

Loading...

; return (

{user?.name}

{user?.bio}

); };

// Vue 2 组件 (Options API) export default { name: 'ProductList', props: { categoryId: { type: Number, required: true }, filters: { type: Object, default: () => ({}) } }, data() { return { products: [], loading: false, error: null, pagination: { page: 1, total: 0 } }; }, computed: { filteredProducts() { return this.products.filter(p => { if (this.filters.minPrice && p.price < this.filters.minPrice) return false; if (this.filters.maxPrice && p.price > this.filters.maxPrice) return false; return true; }); } }, watch: { categoryId: { immediate: true, handler() { this.fetchProducts(); } }, filters: { deep: true, handler() { this.fetchProducts(); } } }, methods: { async fetchProducts() { this.loading = true; try { const { data } = await axios.get('/api/products', { params: { category: this.categoryId, page: this.pagination.page } }); this.products = data.items; this.pagination.total = data.total; } catch (e) { this.error = e.message; } finally { this.loading = false; } } } };

// Vue 3 迁移后组件 (Composition API + script setup) <script setup lang="ts"> import { ref, computed, watch, onMounted } from 'vue'; import axios from 'axios'; interface Product { id: number; name: string; price: number; } interface Filters { minPrice?: number; maxPrice?: number; } const props = defineProps<{ categoryId: number; filters?: Filters }>(); const products = ref<Product[]>([]); const loading = ref(false); const error = ref<string | null>(null); const pagination = ref({ page: 1, total: 0 }); const filteredProducts = computed(() => products.value.filter(p => { if (props.filters?.minPrice && p.price < props.filters.minPrice) return false; if (props.filters?.maxPrice && p.price > props.filters.maxPrice) return false; return true; }) ); async function fetchProducts() { loading.value = true; try { const { data } = await axios.get('/api/products', { params: { category: props.categoryId, page: pagination.value.page } }); products.value = data.items; pagination.value.total = data.total; } catch (e: any) { error.value = e.message; } finally { loading.value = false; } } // Vue 3：支持多个 watcher 合并 watch([() => props.categoryId, () => props.filters], fetchProducts, { immediate: true, deep: true }); </script>

// AngularJS (1.x) 服务与控制器 angular.module('app', []) .service('OrderService', ['$http', '$q', function($http, $q) { this.getOrders = function(userId) { return $http.get(`/api/users/${userId}/orders`) .then(response => response.data) .catch(error => $q.reject(error.data)); }; this.placeOrder = function(order) { return $http.post('/api/orders', order) .then(response => response.data); }; }]) .controller('OrderController', ['$scope', 'OrderService', function($scope, OrderService) { $scope.orders = []; $scope.loading = false; $scope.$watch('userId', function(newVal) { if (newVal) { $scope.loading = true; OrderService.getOrders(newVal) .then(function(orders) { $scope.orders = orders; }) .finally(function() { $scope.loading = false; }); } }); $scope.cancelOrder = function(orderId) { // 控制器逻辑 }; }]);

// Angular (2+) 迁移后代码：Service + Component import { Injectable } from '@angular/core'; import { HttpClient, HttpErrorResponse } from '@angular/common/http'; import { Observable, throwError } from 'rxjs'; import { catchError } from 'rxjs/operators'; @Injectable({ providedIn: 'root' }) export class OrderService { constructor(private http: HttpClient) {} getOrders(userId: number): Observable<Order[]> { return this.http.get<Order[]>(`/api/users/${userId}/orders`) .pipe(catchError(this.handleError)); } placeOrder(order: CreateOrderRequest): Observable<Order> { return this.http.post<Order>('/api/orders', order) .pipe(catchError(this.handleError)); } private handleError(error: HttpErrorResponse) { return throwError(() => new Error(error.error?.message || 'Server error')); } } @Component({ selector: 'app-orders', template: ` <div *ngIf="loading">Loading...</div> <div class="orders-list"> <div *ngFor="let order of orders$ | async" class="order-card"> <h3>Order #{{ order.id }}</h3> <p>Total: {{ order.total | currency }}</p> </div> </div> ` }) export class OrderComponent implements OnInit { @Input() userId!: number; orders$!: Observable<Order[]>; loading = true; constructor(private orderService: OrderService) {} ngOnInit() { this.orders$ = this.orderService.getOrders(this.userId).pipe( tap(() => this.loading = false), catchError(err => { this.loading = false; return throwError(() => err); }) ); } }

# Django 2 原始代码 from django.conf.urls import url, include from django.contrib import admin from django.views.generic import DetailView urlpatterns = [ url(r'^admin/', admin.site.urls), url(r'^api/', include('api.urls', namespace='api')), url(r'^articles/(?P<pk>\d+)/$', DetailView.as_view(model=Article)), ] # settings.py MIDDLEWARE = [ 'django.middleware.security.SecurityMiddleware', 'django.contrib.sessions.middleware.SessionMiddleware', 'django.middleware.common.CommonMiddleware', 'django.middleware.csrf.CsrfViewMiddleware', 'django.contrib.auth.middleware.AuthenticationMiddleware', 'django.contrib.messages.middleware.MessageMiddleware', 'django.middleware.clickjacking.XFrameOptionsMiddleware', ]

# Django 4 迁移后代码 from django.urls import path, include, re_path from django.contrib import admin from django.views.generic import DetailView urlpatterns = [ path('admin/', admin.site.urls), path('api/', include('api.urls')), re_path(r'^articles/(?P<pk>\d+)/$', DetailView.as_view(model=Article)), ] # settings.py - re_path替代url；MIDDLEWARE不再需要django.middleware.security等 MIDDLEWARE = [ 'django.middleware.security.SecurityMiddleware', 'django.contrib.sessions.middleware.SessionMiddleware', 'django.middleware.common.CommonMiddleware', 'django.middleware.csrf.CsrfViewMiddleware', 'django.contrib.auth.middleware.AuthenticationMiddleware', 'django.contrib.messages.middleware.MessageMiddleware', 'django.middleware.clickjacking.XFrameOptionsMiddleware', ] # Django 4.1+ 支持异步视图 from django.http import HttpResponse from django.views import View import asyncio class AsyncArticleView(View): async def get(self, request, *args, **kwargs): await asyncio.sleep(0.1) # 模拟异步IO return HttpResponse("Hello from async Django 4!")

-- SQLite Schema (原始) CREATE TABLE users ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, email TEXT NOT NULL UNIQUE, role TEXT DEFAULT 'user', metadata TEXT, -- JSON存储为文本 created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER REFERENCES users(id), total REAL NOT NULL, status TEXT DEFAULT 'pending', items TEXT, -- JSON数组存储为文本 created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX idx_orders_user_id ON orders(user_id); CREATE INDEX idx_orders_status ON orders(status);

-- PostgreSQL Schema (迁移后) CREATE TABLE users ( id SERIAL PRIMARY KEY, name VARCHAR(255) NOT NULL, email VARCHAR(255) NOT NULL UNIQUE, role VARCHAR(20) DEFAULT 'user', metadata JSONB DEFAULT '{}', -- 使用原生JSONB类型 created_at TIMESTAMPTZ DEFAULT NOW() ); CREATE TABLE orders ( id SERIAL PRIMARY KEY, user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, total NUMERIC(10, 2) NOT NULL, -- 使用精确数值类型 status VARCHAR(20) DEFAULT 'pending', items JSONB DEFAULT '[]', -- 使用JSONB存储 created_at TIMESTAMPTZ DEFAULT NOW() ); CREATE INDEX idx_orders_user_id ON orders(user_id); CREATE INDEX idx_orders_status ON orders(status); CREATE INDEX idx_users_metadata_gin ON users USING GIN (metadata); -- GIN索引加速JSON查询

# 数据迁移脚本：使用Claude Code生成的双写模式 import sqlite3 import psycopg2 from datetime import datetime import json def migrate_users(sqlite_conn, pg_conn, batch_size=1000): """将用户数据从SQLite批量迁移到PostgreSQL""" sqlite_cursor = sqlite_conn.cursor() pg_cursor = pg_conn.cursor() sqlite_cursor.execute("SELECT COUNT(*) FROM users") total = sqlite_cursor.fetchone()[0] print(f"Found {total} users to migrate") offset = 0 while offset < total: sqlite_cursor.execute( "SELECT id, name, email, role, metadata, created_at FROM users LIMIT ? OFFSET ?", (batch_size, offset) ) rows = sqlite_cursor.fetchall() if not rows: break for row in rows: pg_cursor.execute(""" INSERT INTO users (id, name, email, role, metadata, created_at) VALUES (%s, %s, %s, %s, %s::jsonb, %s) ON CONFLICT (id) DO NOTHING """, ( row[0], row[1], row[2], row[3], json.dumps(json.loads(row[4])) if row[4] else '{}', row[5] )) pg_conn.commit() offset += batch_size print(f"Migrated {min(offset, total)}/{total} users") sqlite_cursor.close() pg_cursor.close() # PostgreSQL特有功能：迁移后利用JSONB查询 def get_users_with_metadata(pg_conn): """利用PostgreSQL JSONB能力做高级查询""" with pg_conn.cursor() as cur: cur.execute(""" SELECT id, name, email FROM users WHERE metadata @> '{"vip": true}'::jsonb AND metadata->>'tier' = 'gold' ORDER BY created_at DESC LIMIT 100 """) return cur.fetchall()

// MongoDB 文档结构（原始） // orders collection { "_id": ObjectId("..."), "orderNumber": "ORD-2024-0001", "customer": { "name": "张三", "email": "zhangsan@example.com", "address": { "province": "上海", "city": "上海市", "district": "浦东新区" } }, "items": [ { "productId": "P001", "name": "笔记本电脑", "price": 5999, "qty": 1 }, { "productId": "P002", "name": "鼠标", "price": 99, "qty": 2 } ], "totalAmount": 6197, "status": "shipped", "createdAt": ISODate("2024-01-15T10:30:00Z"), "tags": ["urgent", "gift"] }

-- PostgreSQL 关系模型（迁移后） CREATE TABLE customers ( id SERIAL PRIMARY KEY, name VARCHAR(100) NOT NULL, email VARCHAR(255) UNIQUE NOT NULL, province VARCHAR(50), city VARCHAR(50), district VARCHAR(50), created_at TIMESTAMPTZ DEFAULT NOW() ); CREATE TABLE products ( id VARCHAR(20) PRIMARY KEY, name VARCHAR(200) NOT NULL, price NUMERIC(10, 2) NOT NULL, category VARCHAR(50), created_at TIMESTAMPTZ DEFAULT NOW() ); CREATE TABLE orders ( id SERIAL PRIMARY KEY, order_number VARCHAR(50) UNIQUE NOT NULL, customer_id INTEGER REFERENCES customers(id), total_amount NUMERIC(12, 2) NOT NULL, status VARCHAR(20) DEFAULT 'pending', tags TEXT[], -- Postgres数组类型 created_at TIMESTAMPTZ DEFAULT NOW() ); CREATE TABLE order_items ( id SERIAL PRIMARY KEY, order_id INTEGER REFERENCES orders(id) ON DELETE CASCADE, product_id VARCHAR(20) REFERENCES products(id), quantity INTEGER NOT NULL DEFAULT 1, unit_price NUMERIC(10, 2) NOT NULL, subtotal NUMERIC(12, 2) GENERATED ALWAYS AS (quantity * unit_price) STORED ); -- 复杂关联查询 SELECT o.order_number, c.name AS customer_name, c.city, json_agg( json_build_object('product', p.name, 'qty', oi.quantity, 'price', oi.unit_price) ) AS items, o.total_amount, o.status FROM orders o JOIN customers c ON c.id = o.customer_id JOIN order_items oi ON oi.order_id = o.id JOIN products p ON p.id = oi.product_id WHERE o.created_at >= NOW() - INTERVAL '30 days' AND o.status = 'shipped' GROUP BY o.id, c.id ORDER BY o.created_at DESC;

# Dockerfile 容器化迁移示例 FROM node:18-alpine AS builder WORKDIR /app COPY package*.json ./ RUN npm ci --only=production COPY . . RUN npm run build FROM node:18-alpine WORKDIR /app COPY --from=builder /app/dist ./dist COPY --from=builder /app/node_modules ./node_modules COPY --from=builder /app/package.json ./ # 非root用户运行，提升安全性 RUN addgroup -g 1001 app && adduser -u 1001 -G app -s /bin/sh -D app USER app EXPOSE 3000 HEALTHCHECK --interval=30s --timeout=3s --retries=3 \ CMD wget -qO- http://localhost:3000/health || exit 1 CMD ["node", "dist/server.js"]

# Kubernetes 部署配置 apiVersion: apps/v1 kind: Deployment metadata: name: api-gateway namespace: production labels: app: api-gateway tier: frontend spec: replicas: 3 strategy: type: RollingUpdate rollingUpdate: maxUnavailable: 1 maxSurge: 1 selector: matchLabels: app: api-gateway template: metadata: labels: app: api-gateway version: "2.0" spec: containers: - name: gateway image: registry.example.com/api-gateway:2.0.0 ports: - containerPort: 3000 name: http env: - name: NODE_ENV value: "production" - name: DB_CONNECTION valueFrom: secretKeyRef: name: db-credentials key: connection-string resources: requests: cpu: "500m" memory: "512Mi" limits: cpu: "1000m" memory: "1Gi" livenessProbe: httpGet: path: /health port: 3000 initialDelaySeconds: 10 periodSeconds: 15 readinessProbe: httpGet: path: /ready port: 3000 initialDelaySeconds: 5 periodSeconds: 10 --- apiVersion: v1 kind: Service metadata: name: api-gateway namespace: production spec: selector: app: api-gateway ports: - port: 80 targetPort: 3000 type: ClusterIP

# 单体架构原始路由（Python Flask） from flask import Flask, request, jsonify from flask_sqlalchemy import SQLAlchemy app = Flask(__name__) db = SQLAlchemy(app) @app.route('/api/orders', methods=['POST']) def create_order(): """单体应用中创建订单的完整逻辑""" data = request.json # 1. 验证用户 user = User.query.get(data['userId']) if not user or not user.is_active: return jsonify({'error': 'Invalid user'}), 400 # 2. 验证库存 for item in data['items']: product = Product.query.get(item['productId']) if not product or product.stock < item['quantity']: return jsonify({'error': f'Insufficient stock for {item["productId"]}'}), 400 # 3. 计算价格和折扣 subtotal = 0 for item in data['items']: product = Product.query.get(item['productId']) subtotal += product.price * item['quantity'] # 4. 处理支付 payment = Payment(user_id=user.id, amount=subtotal, method=data['paymentMethod']) db.session.add(payment) db.session.flush() payment_result = process_payment(payment) # 5. 创建订单 order = Order(user_id=user.id, total=subtotal, status='confirmed') db.session.add(order) db.session.commit() # 6. 通知（邮件 + 内部消息） send_email(user.email, 'order_confirmation', order.id) notify_inventory(data['items']) return jsonify({'orderId': order.id, 'status': 'confirmed'}), 201

# 微服务重构后：订单服务（从单体中拆分） from flask import Flask, request, jsonify import httpx import asyncio app = Flask(__name__) @app.route('/api/orders', methods=['POST']) async def create_order(): """微服务：订单服务（通过异步HTTP调用其他服务）""" data = request.json async with httpx.AsyncClient() as client: # 1. 调用用户服务验证用户 user_resp = await client.post( 'http://user-service/api/internal/verify', json={'userId': data['userId']} ) if user_resp.status_code != 200: return jsonify({'error': 'User verification failed'}), 400 # 2. 调用库存服务预占库存 inventory_resp = await client.post( 'http://inventory-service/api/internal/reserve', json={'items': data['items']} ) if inventory_resp.status_code != 200: return jsonify({'error': 'Inventory reservation failed'}), 400 # 3. 调用价格服务计算（含折扣） pricing_resp = await client.post( 'http://pricing-service/api/internal/calculate', json={'items': data['items'], 'coupon': data.get('couponCode')} ) pricing = pricing_resp.json() # 4. 调用支付服务处理支付 payment_resp = await client.post( 'http://payment-service/api/internal/charge', json={ 'userId': data['userId'], 'amount': pricing['total'], 'method': data['paymentMethod'] } ) if payment_resp.status_code != 200: # 库存回滚 await client.post( 'http://inventory-service/api/internal/release', json={'items': data['items']} ) return jsonify({'error': 'Payment failed'}), 402 # 5. 在当前服务创建订单记录 order = create_order_record(data, pricing) # 6. 异步通知（通过消息队列） await notify_services_async(order) return jsonify({'orderId': order.id, 'status': 'confirmed'}), 201

# 传统Flask API处理器 from flask import Flask, request, jsonify import redis import boto3 app = Flask(__name__) redis_client = redis.Redis(host='localhost', port=6379, db=0) s3 = boto3.client('s3') @app.route('/api/documents/upload', methods=['POST']) def upload_document(): file = request.files['document'] user_id = request.headers['X-User-Id'] # 文件处理 content = file.read() processed = process_content(content) # 上传到S3 key = f'documents/{user_id}/{file.filename}' s3.put_object(Bucket='my-bucket', Key=key, Body=processed) # 缓存状态 redis_client.setex(f'doc:{key}', 3600, 'processing') return jsonify({'key': key, 'status': 'processing'})

# AWS Lambda 函数迁移后 import json import boto3 import aiomysql from datetime import datetime s3 = boto3.client('s3') dynamodb = boto3.resource('dynamodb') def lambda_handler(event, context): """API Gateway + Lambda 的无服务器架构""" # 1. 解析API Gateway事件 http_method = event['httpMethod'] path = event['path'] headers = event.get('headers', {}) body = json.loads(event.get('body', '{}')) # 2. 路由分发 if http_method == 'POST' and path == '/api/documents/upload': return handle_upload(event, context) elif http_method == 'GET' and path.startswith('/api/documents/'): doc_id = path.split('/')[-1] return handle_get_document(doc_id, context) else: return { 'statusCode': 404, 'headers': {'Content-Type': 'application/json'}, 'body': json.dumps({'error': 'Not found'}, ensure_ascii=False) } def handle_upload(event, context): """无服务器文件上传处理""" import base64 body = json.loads(event.get('body', '{}')) user_id = event['requestContext']['authorizer']['claims']['sub'] file_content = base64.b64decode(body['fileBase64']) filename = body['filename'] # 直接上传到S3（Lambda的临时存储只有512MB） key = f'documents/{user_id}/{datetime.now().isoformat()}_{filename}' s3.put_object(Bucket='my-bucket', Key=key, Body=file_content) # 使用DynamoDB替代Redis进行状态跟踪 table = dynamodb.Table('document-status') table.put_item(Item={ 'documentKey': key, 'userId': user_id, 'status': 'processing', 'ttl': int(datetime.now().timestamp()) + 86400 # 24小时后自动过期 }) return { 'statusCode': 200, 'headers': { 'Content-Type': 'application/json', 'Access-Control-Allow-Origin': '*' }, 'body': json.dumps({ 'key': key, 'status': 'processing' }, ensure_ascii=False) }

# 逐步迁移的状态跟踪机制 class MigrationTracker: """跟踪每个迁移步骤的状态""" def __init__(self, migration_name: str): self.migration_name = migration_name self.steps = [] self.checkpoints = {} def add_step(self, step_id: str, description: str, depends_on: list = None): self.steps.append({ 'id': step_id, 'description': description, 'depends_on': depends_on or [], 'status': 'pending', # pending | in_progress | completed | failed | rolled_back 'result': None }) def mark_step(self, step_id: str, status: str, result: dict = None): checkpoint = { 'step_id': step_id, 'status': status, 'timestamp': datetime.utcnow().isoformat(), 'result': result } key = f"migration:{self.migration_name}:checkpoint:{step_id}" self.checkpoints[key] = checkpoint def can_execute_step(self, step_id: str) -> bool: step = next((s for s in self.steps if s['id'] == step_id), None) if not step: return False return all( self.checkpoints.get(f"migration:{self.migration_name}:checkpoint:{dep}")['status'] == 'completed' for dep in step['depends_on'] ) def rollback(self, failed_step: str): """失败时顺序回滚已完成的步骤""" completed = [ s for s in reversed(self.steps) if self.checkpoints.get(f"migration:{self.migration_name}:checkpoint:{s['id']}", {}).get('status') == 'completed' ] for step in completed: if step['id'] == failed_step: break self.rollback_step(step['id']) def generate_report(self): return { 'migration': self.migration_name, 'total_steps': len(self.steps), 'completed': sum(1 for s in self.steps if self.checkpoints.get(f"migration:{self.migration_name}:checkpoint:{s['id']}", {}).get('status') == 'completed'), 'failed': sum(1 for s in self.steps if self.checkpoints.get(f"migration:{self.migration_name}:checkpoint:{s['id']}", {}).get('status') == 'failed'), 'duration': None }

// 暗启动流量复制中间件 import express from 'express'; import http from 'http'; import { v4 as uuidv4 } from 'uuid'; const app = express(); // 暗启动配置 const darkLaunchConfig = { enabled: true, newServiceUrl: 'http://new-service.internal:4000', sampleRate: 0.1, // 10% 流量复制到新系统 compareResponses: true, timeout: 5000 }; async function darkLaunchMiddleware(req, res, next) { if (!darkLaunchConfig.enabled) return next(); const requestId = uuidv4(); const shouldDarkLaunch = Math.random() < darkLaunchConfig.sampleRate; if (!shouldDarkLaunch) return next(); // 复制请求到新服务（不阻塞主请求） const originalJson = res.json.bind(res); const startTime = Date.now(); res.json = function(body) { if (shouldDarkLaunch) { const payload = { requestId, originalResponse: body, request: { method: req.method, path: req.path, headers: req.headers, body: req.body }, timestamp: new Date().toISOString() }; // 异步调用新服务比较结果 const newServiceReq = http.request( `${darkLaunchConfig.newServiceUrl}${req.path}`, { method: req.method, headers: { 'Content-Type': 'application/json', 'X-Request-Id': requestId, 'X-Dark-Launch': 'true' }, timeout: darkLaunchConfig.timeout }, (newServiceRes) => { let data = ''; newServiceRes.on('data', chunk => data += chunk); newServiceRes.on('end', () => { const duration = Date.now() - startTime; try { const newBody = JSON.parse(data); payload.newServiceResponse = newBody; payload.newServiceStatus = newServiceRes.statusCode; payload.duration = duration; // 比较响应差异 payload.discrepancies = findDiscrepancies(body, newBody); payload.matched = payload.discrepancies.length === 0; // 记录暗启动结果 logDarkLaunchResult(payload); } catch (e) { payload.error = e.message; logDarkLaunchResult(payload); } }); } ); newServiceReq.on('error', (e) => { logDarkLaunchResult({ requestId, error: `New service error: ${e.message}`, timestamp: new Date().toISOString() }); }); newServiceReq.write(JSON.stringify(req.body)); newServiceReq.end(); } return originalJson.call(this, body); }; next(); } app.use(darkLaunchMiddleware); function findDiscrepancies(oldBody, newBody) { const discrepancies = []; // 递归比较两个响应对象的差异 function compare(o, n, path = '') { if (typeof o !== typeof n) { discrepancies.push({ path, expected: typeof o, actual: typeof n }); return; } if (typeof o === 'object' && o !== null && n !== null) { const allKeys = new Set([...Object.keys(o), ...Object.keys(n)]); for (const key of allKeys) { compare(o[key], n[key], `${path}.${key}`); } } else if (o !== n) { discrepancies.push({ path, expected: o, actual: n }); } } compare(oldBody, newBody); return discrepancies; } async function logDarkLaunchResult(result) { // 记录到专门的暗启动日志系统 await db.collection('dark_launch_logs').insertOne({ ...result, loggedAt: new Date() }); // 如果发现差异，触发告警 if (result.discrepancies?.length > 0) { await alertSystem.trigger({ level: 'warning', type: 'dark_launch_discrepancy', details: result }); } }

# 蓝绿迁移基础设施配置（Terraform） resource "aws_lb_target_group" "blue" { name = "api-blue" port = 8080 protocol = "HTTP" vpc_id = aws_vpc.main.id health_check { path = "/healthz" interval = 30 timeout = 5 healthy_threshold = 2 unhealthy_threshold = 2 } tags = { Environment = "blue", ManagedBy = "migration-tf" } } resource "aws_lb_target_group" "green" { name = "api-green" port = 8080 protocol = "HTTP" vpc_id = aws_vpc.main.id health_check { path = "/healthz" interval = 30 timeout = 5 healthy_threshold = 2 unhealthy_threshold = 2 } tags = { Environment = "green", ManagedBy = "migration-tf" } } # 默认指向蓝环境 resource "aws_lb_listener_rule" "main" { listener_arn = aws_lb_listener.main.arn priority = 100 action { type = "forward" target_group_arn = aws_lb_target_group.blue.arn # 切换时改为 green } condition { path_pattern { values = ["/*"] } } }

# 迁移切换与回滚脚本 #!/bin/bash # blue-green-switch.sh — 蓝绿迁移切换工具 set -euo pipefail CURRENT_ENV_FILE="/tmp/current-env.txt" APP_NAME="api-gateway" LISTENER_RULE_ARN="arn:aws:elasticloadbalancing:..." get_current_env() { if [ -f "$CURRENT_ENV_FILE" ]; then cat "$CURRENT_ENV_FILE" else echo "blue" # 默认为蓝环境 fi } switch_traffic() { local target_env=$1 echo "[$(date '+%Y-%m-%d %H:%M:%S')] Switching traffic to ${target_env} environment..." if [ "$target_env" = "green" ]; then NEW_TARGET_GROUP_ARN="arn:aws:elasticloadbalancing:...:targetgroup/api-green/..." else NEW_TARGET_GROUP_ARN="arn:aws:elasticloadbalancing:...:targetgroup/api-blue/..." fi # 切换负载均衡器规则 aws elbv2 modify-listener-rule \ --listener-rule-arn "$LISTENER_RULE_ARN" \ --actions "Type=forward,TargetGroupArn=$NEW_TARGET_GROUP_ARN" echo "$target_env" > "$CURRENT_ENV_FILE" echo "[$(date '+%Y-%m-%d %H:%M:%S')] Traffic switched to ${target_env}" } validate_environment() { local env=$1 local health_endpoint="${env}.internal:8080/healthz" echo "[$(date '+%Y-%m-%d %H:%M:%S')] Validating ${env} environment..." for i in $(seq 1 30); do STATUS=$(curl -s -o /dev/null -w "%{http_code}" "http://${health_endpoint}") if [ "$STATUS" = "200" ]; then echo "[$(date '+%Y-%m-%d %H:%M:%S')] ${env} environment is healthy" return 0 fi echo "[$(date '+%Y-%m-%d %H:%M:%S')] Waiting for ${env} to become healthy... (attempt $i)" sleep 5 done echo "[$(date '+%Y-%m-%d %H:%M:%S')] ERROR: ${env} environment failed health check" return 1 } perform_migration() { local current_env=$(get_current_env) local target_env if [ "$current_env" = "blue" ]; then target_env="green" else target_env="blue" fi echo "=== Blue-Green Migration ===" echo "Current: $current_env" echo "Target: $target_env" echo "" # 步骤1: 验证目标环境健康 validate_environment "$target_env" || { echo "ABORTING: Target environment not healthy" exit 1 } # 步骤2: 切换流量 switch_traffic "$target_env" # 步骤3: 验证切换后系统 echo "[$(date '+%Y-%m-%d %H:%M:%S')] Running smoke tests..." if npm run smoke-test; then echo "[$(date '+%Y-%m-%d %H:%M:%S')] Migration completed successfully" else echo "[$(date '+%Y-%m-%d %H:%M:%S')] Smoke tests FAILED! Initiating rollback..." switch_traffic "$current_env" exit 2 fi } rollback() { local current_env=$(get_current_env) echo "=== Initiating Rollback ===" if [ "$current_env" = "blue" ]; then switch_traffic "green" else switch_traffic "blue" fi echo "[$(date '+%Y-%m-%d %H:%M:%S')] Rollback complete" } # 主入口 case "${1:-migrate}" in migrate) perform_migration ;; rollback) rollback ;; status) echo "Current environment: $(get_current_env)" ;; *) echo "Usage: $0 {migrate|rollback|status}" ;; esac

// 适配器模式实现新旧API兼容 // 目标：将旧的基于回调的API适配为新的Promise/async API // 旧的回调式API（不可修改的遗留代码） interface LegacyDatabase { query(sql: string, callback: (err: Error | null, rows?: any[]) => void): void; insert(table: string, data: any, callback: (err: Error | null, id?: number) => void): void; update(table: string, id: number, data: any, callback: (err: Error | null) => void): void; delete(table: string, id: number, callback: (err: Error | null) => void): void; } // 适配器：将旧API转换为Promise接口 class DatabaseAdapter { private legacy: LegacyDatabase; constructor(legacy: LegacyDatabase) { this.legacy = legacy; } async query(sql: string): Promise { return new Promise((resolve, reject) => { this.legacy.query(sql, (err, rows) => { if (err) reject(err); else resolve(rows || []); }); }); } async insert(table: string, data: any): Promise { return new Promise((resolve, reject) => { this.legacy.insert(table, data, (err, id) => { if (err) reject(err); else resolve(id!); }); }); } async update(table: string, id: number, data: any): Promise { return new Promise((resolve, reject) => { this.legacy.update(table, id, data, (err) => { if (err) reject(err); else resolve(); }); }); } async delete(table: string, id: number): Promise { return new Promise((resolve, reject) => { this.legacy.delete(table, id, (err) => { if (err) reject(err); else resolve(); }); }); } } // 新数据库接口（目标态） interface NewDatabase { query(sql: string, params?: any[]): Promise; insert(table: string, data: any): Promise; update(table: string, id: number, data: any): Promise; delete(table: string, id: number): Promise; transaction(fn: (tx: Transaction) => Promise): Promise; } // 兼容层：在迁移期间同时支持新旧接口 class CompatibilityLayer implements NewDatabase { private adapter: DatabaseAdapter; private featureFlags: Map; constructor(legacy: LegacyDatabase, flags: Map) { this.adapter = new DatabaseAdapter(legacy); this.featureFlags = flags; } async query(sql: string, params?: any[]): Promise { // 使用特性开关决定走哪条路径 if (this.featureFlags.get('use_new_query_engine')) { return this.newQueryEngine(sql, params); } return this.adapter.query(sql) as Promise; } private async newQueryEngine(sql: string, params?: any[]): Promise { // 新查询引擎逻辑（逐步替换） console.log(`[Compatibility] Using new query engine: ${sql}`); // 此处可以调用新的ORM或查询构建器 return this.adapter.query(sql); } async insert(table: string, data: any): Promise { if (this.featureFlags.get('use_new_orm')) { return this.newInsert(table, data); } return this.adapter.insert(table, data); } private async newInsert(table: string, data: any): Promise { // 新的ORM插入逻辑 return this.adapter.insert(table, data); } // ...update, delete, transaction 类似模式 async transaction(fn: (tx: Transaction) => Promise): Promise { // 事务支持是新接口新增的能力 throw new Error('Transactions not supported in legacy mode'); } }

# 功能等价测试框架 import unittest import json import subprocess from typing import Any, Dict, List from dataclasses import dataclass @dataclass class TestCase: name: str input: Dict[str, Any] expected_output: Dict[str, Any] setup_sql: str = None teardown_sql: str = None class EquivalenceTester: """ 迁移功能等价测试器：同时运行新旧系统并比较输出 """ def __init__(self, old_system_url: str, new_system_url: str): self.old_url = old_system_url self.new_url = new_system_url self.discrepancies = [] async def run_test(self, test_case: TestCase) -> Dict: """对单个测试用例运行等价测试""" async with aiohttp.ClientSession() as session: # 同时调用新旧系统 old_resp, new_resp = await asyncio.gather( self.call_system(session, self.old_url, test_case), self.call_system(session, self.new_url, test_case) ) result = { 'test_name': test_case.name, 'input': test_case.input, 'old_output': old_resp, 'new_output': new_resp, 'match': old_resp == new_resp, 'differences': self.find_differences(old_resp, new_resp) } if not result['match']: self.discrepancies.append(result) print(f" DISCREPANCY: {test_case.name}") print(f" Fields differ: {list(result['differences'].keys())}") return result async def call_system(self, session, url, test_case): async with session.post( url + '/api/process', json=test_case.input, timeout=aiohttp.ClientTimeout(total=30) ) as resp: return await resp.json() def find_differences(self, old, new, path=''): diffs = {} if isinstance(old, dict) and isinstance(new, dict): all_keys = set(old.keys()) | set(new.keys()) for key in all_keys: new_path = f"{path}.{key}" if path else key if key not in old: diffs[new_path] = {'type': 'missing_in_old', 'new_value': new[key]} elif key not in new: diffs[new_path] = {'type': 'missing_in_new', 'old_value': old[key]} else: child_diffs = self.find_differences(old[key], new[key], new_path) diffs.update(child_diffs) elif old != new: diffs[path] = {'type': 'value_mismatch', 'old': old, 'new': new} return diffs def run_test_suite(self, test_cases: List[TestCase]): """运行全套测试并生成报告""" results = asyncio.run(self.run_all(test_cases)) summary = { 'total': len(results), 'passed': sum(1 for r in results if r['match']), 'failed': sum(1 for r in results if not r['match']), 'discrepancies': self.discrepancies } print(f"\n{'='*50}") print(f"Equivalence Test Summary") print(f"{'='*50}") print(f"Total: {summary['total']}") print(f"Passed: {summary['passed']}") print(f"Failed: {summary['failed']}") print(f"Pass rate: {summary['passed']/summary['total']*100:.1f}%") if summary['failed'] > 0: print(f"\nDetailed discrepancies:") for d in self.discrepancies: print(f" - {d['test_name']}: {d['differences']}") return summary

// 性能对比基准测试 import autocannon from 'autocannon'; import { writeFileSync } from 'fs'; interface BenchmarkConfig { oldUrl: string; newUrl: string; duration: number; // 测试持续时间（秒） connections: number; pipelining: number; endpoints: string[]; } async function runBenchmark(config: BenchmarkConfig): Promise { const results = { timestamp: new Date().toISOString(), config, old: {} as any, new: {} as any, comparison: {} as any }; for (const endpoint of config.endpoints) { console.log(`\nBenchmarking ${endpoint}...`); // 测试旧系统 console.log(` Old system (${config.oldUrl})...`); const oldResult = await runSingleBenchmark(config.oldUrl + endpoint, config); results.old[endpoint] = oldResult; // 测试新系统 console.log(` New system (${config.newUrl})...`); const newResult = await runSingleBenchmark(config.newUrl + endpoint, config); results.new[endpoint] = newResult; // 计算对比 results.comparison[endpoint] = { latencyP50_change: ((newResult.latency.p50 - oldResult.latency.p50) / oldResult.latency.p50 * 100).toFixed(2) + '%', latencyP99_change: ((newResult.latency.p99 - oldResult.latency.p99) / oldResult.latency.p99 * 100).toFixed(2) + '%', throughput_change: ((newResult.requests.average - oldResult.requests.average) / oldResult.requests.average * 100).toFixed(2) + '%', errorRate_change: ((newResult.errors - oldResult.errors) / (oldResult.errors || 1) * 100).toFixed(2) + '%' }; console.log(` Comparison:`, results.comparison[endpoint]); } writeFileSync('benchmark-results.json', JSON.stringify(results, null, 2)); console.log('\nBenchmark results saved to benchmark-results.json'); } function runSingleBenchmark(url: string, config: BenchmarkConfig): Promise { return new Promise((resolve, reject) => { const instance = autocannon({ url, connections: config.connections, duration: config.duration, pipelining: config.pipelining, title: url, timeout: 30 }, (err, result) => { if (err) return reject(err); resolve({ latency: { p50: result.latency.p50, p95: result.latency.p95, p99: result.latency.p99, min: result.latency.min, max: result.latency.max }, requests: { average: result.requests.average, total: result.requests.total }, throughput: result.throughput, errors: result.errors, timeouts: result.timeouts, non2xx: result.non2xx }); }); }); } // 生成HTML性能对比报告 function generateReport(results: any): string { let html = ``; html += ``; html += `

迁移性能对比报告

`; html += `

测试时间: ${results.timestamp}

`; for (const [endpoint, comp] of Object.entries(results.comparison)) { html += `

${endpoint}

`; html += ``; const metrics = [ ['P50延迟', results.old[endpoint].latency.p50, results.new[endpoint].latency.p50, comp.latencyP50_change], ['P99延迟', results.old[endpoint].latency.p99, results.new[endpoint].latency.p99, comp.latencyP99_change], ['吞吐量', results.old[endpoint].requests.average, results.new[endpoint].requests.average, comp.throughput_change], ]; for (const [name, old, nw, change] of metrics) { const isRegression = parseFloat(change) > 5 && (name.includes('延迟')); const cls = isRegression ? 'regression' : parseFloat(change) <= 0 ? 'improvement' : 'neutral'; html += ``; } html += `

指标	旧系统	新系统	变化	判定
${name}	${old.toFixed(2)}	${nw.toFixed(2)}	${change}	${isRegression ? 'REGRESSION' : 'OK'}

`; } html += ``; return html; }

# 流量录制与回放验证系统 import json import time import hashlib import aiohttp import asyncio from datetime import datetime from collections import defaultdict class TrafficRecorder: """生产流量录制器""" def __init__(self, record_path: str, sample_rate: float = 1.0): self.record_path = record_path self.sample_rate = sample_rate self.buffer = [] self.buffer_size = 100 self.session_id = datetime.now().strftime('%Y%m%d_%H%M%S') async def record_request(self, request_data: dict): """录制生产请求""" if not self.should_sample(): return entry = { 'id': hashlib.md5(json.dumps(request_data, sort_keys=True).encode()).hexdigest()[:16], 'timestamp': datetime.now().isoformat(), 'method': request_data['method'], 'path': request_data['path'], 'headers': {k: v for k, v in request_data.get('headers', {}).items() if k.lower() not in ('authorization', 'cookie', 'x-api-key')}, 'query': request_data.get('query', {}), 'body': request_data.get('body'), 'recorded_response': request_data.get('response') } self.buffer.append(entry) if len(self.buffer) >= self.buffer_size: await self.flush() def should_sample(self) -> bool: return random.random() < self.sample_rate async def flush(self): if not self.buffer: return filename = f"{self.record_path}/traffic_{self.session_id}_{int(time.time())}.jsonl" with open(filename, 'w', encoding='utf-8') as f: for entry in self.buffer: f.write(json.dumps(entry, ensure_ascii=False) + '\n') self.buffer = [] class TrafficReplayValidator: """流量回放验证器""" def __init__(self, old_url: str, new_url: str): self.old_url = old_url self.new_url = new_url self.stats = defaultdict(int) self.details = [] async def replay_traffic(self, traffic_file: str) -> dict: """回放录制的流量并验证""" print(f"Replaying traffic from {traffic_file}...") async with aiohttp.ClientSession() as session: with open(traffic_file, 'r', encoding='utf-8') as f: for line in f: entry = json.loads(line.strip()) await self.replay_one(session, entry) return self.generate_report() async def replay_one(self, session: aiohttp.ClientSession, entry: dict): """回放单个请求""" self.stats['total'] += 1 try: # 同时发送到新旧系统 old_resp, new_resp = await asyncio.gather( self.send_request(session, self.old_url, entry), self.send_request(session, self.new_url, entry) ) # 比较响应 comparison = self.compare_responses(entry, old_resp, new_resp) self.details.append(comparison) if comparison['match']: self.stats['matched'] += 1 else: self.stats['mismatched'] += 1 if comparison.get('status_code_mismatch'): self.stats['status_mismatch'] += 1 if comparison.get('body_mismatch'): self.stats['body_mismatch'] += 1 except Exception as e: self.stats['errors'] += 1 self.stats['total'] -= 1 async def send_request(self, session, base_url, entry): url = f"{base_url}{entry['path']}" method = entry['method'].lower() kwargs = { 'headers': entry.get('headers', {}), 'timeout': aiohttp.ClientTimeout(total=30) } if entry.get('body') and method in ('post', 'put', 'patch'): kwargs['json'] = entry['body'] if entry.get('query'): kwargs['params'] = entry['query'] async with getattr(session, method)(url, **kwargs) as resp: return { 'status': resp.status, 'headers': dict(resp.headers), 'body': await resp.json() } def compare_responses(self, entry, old, new): result = { 'request_id': entry['id'], 'path': entry['path'], 'method': entry['method'] } # 比较状态码 result['status_code_mismatch'] = old['status'] != new['status'] result['old_status'] = old['status'] result['new_status'] = new['status'] # 比较响应体（关键字段） result['body_mismatch'] = old['body'] != new['body'] if result['body_mismatch']: diffs = find_key_differences(old['body'], new['body']) result['differences'] = diffs[:10] # 只记录前10个差异 result['match'] = not result['status_code_mismatch'] and not result['body_mismatch'] return result def generate_report(self): total = self.stats['total'] matched = self.stats['matched'] mismatched = self.stats['mismatched'] errors = self.stats['errors'] return { 'total_requests': total, 'matched': matched, 'mismatched': mismatched, 'errors': errors, 'match_rate': f"{matched/total*100:.2f}%" if total > 0 else "N/A", 'details': self.details[:50], # 只包含前50条详情 'summary': f"Traffic replay complete: {matched}/{total} matched ({matched/total*100:.1f}%)" }