JavaScript系列（44）--微服务架构实现详解

JavaScript微服务架构实现详解 🏗️

今天，让我们来学习如何在JavaScript中实现微服务架构。微服务架构是一种将应用程序构建为一组小型服务的方法，每个服务运行在自己的进程中，并通过轻量级机制通信。

微服务基础概念 🌟

💡 小知识：微服务架构的核心是将大型应用拆分成多个独立的服务，每个服务都可以独立部署、扩展和维护。

基础架构实现 📊

// 1. 基础服务类
class MicroService {
    constructor(name, port) {
        this.name = name;
        this.port = port;
        this.express = require('express');
        this.app = this.express();
        this.routes = new Map();
    }
    
    // 配置中间件
    setupMiddleware() {
        this.app.use(this.express.json());
        this.app.use(this.express.urlencoded({ extended: true }));
    }
    
    // 注册路由
    registerRoute(method, path, handler) {
        this.routes.set(`${method}:${path}`, handler);
        this.app[method.toLowerCase()](path, handler);
    }
    
    // 启动服务
    async start() {
        return new Promise((resolve) => {
            this.server = this.app.listen(this.port, () => {
                console.log(`Service ${this.name} started on port ${this.port}`);
                resolve();
            });
        });
    }
    
    // 停止服务
    async stop() {
        return new Promise((resolve) => {
            this.server.close(() => {
                console.log(`Service ${this.name} stopped`);
                resolve();
            });
        });
    }
}

// 2. 服务注册中心
class ServiceRegistry {
    constructor() {
        this.services = new Map();
        this.healthChecks = new Map();
    }
    
    // 注册服务
    register(name, host, port) {
        const serviceId = `${name}-${host}:${port}`;
        this.services.set(serviceId, {
            name,
            host,
            port,
            status: 'healthy',
            lastHeartbeat: Date.now()
        });
        
        return serviceId;
    }
    
    // 注销服务
    unregister(serviceId) {
        this.services.delete(serviceId);
        this.healthChecks.delete(serviceId);
    }
    
    // 获取服务实例
    getService(name) {
        const services = Array.from(this.services.values())
            .filter(s => s.name === name && s.status === 'healthy');
            
        if (services.length === 0) {
            throw new Error(`No healthy instances of service ${name} found`);
        }
        
        // 简单的负载均衡：随机选择一个实例
        return services[Math.floor(Math.random() * services.length)];
    }
    
    // 健康检查
    startHealthCheck(serviceId, interval = 30000) {
        const check = setInterval(() => {
            const service = this.services.get(serviceId);
            if (!service) {
                clearInterval(check);
                return;
            }
            
            if (Date.now() - service.lastHeartbeat > interval) {
                service.status = 'unhealthy';
            }
        }, interval);
        
        this.healthChecks.set(serviceId, check);
    }
}

服务通信实现 🔄

// 1. HTTP通信适配器
class HttpCommunicator {
    constructor() {
        this.axios = require('axios');
    }
    
    async request(service, method, path, data) {
        const url = `http://${service.host}:${service.port}${path}`;
        try {
            const response = await this.axios({
                method,
                url,
                data
            });
            return response.data;
        } catch (error) {
            throw new Error(`Service communication failed: ${error.message}`);
        }
    }
}

// 2. 消息队列通信
class MessageQueueCommunicator {
    constructor() {
        this.amqp = require('amqplib');
        this.connection = null;
        this.channel = null;
    }
    
    async connect(url) {
        this.connection = await this.amqp.connect(url);
        this.channel = await this.connection.createChannel();
    }
    
    async publish(queue, message) {
        await this.channel.assertQueue(queue);
        return this.channel.sendToQueue(queue, Buffer.from(JSON.stringify(message)));
    }
    
    async subscribe(queue, callback) {
        await this.channel.assertQueue(queue);
        return this.channel.consume(queue, (msg) => {
            if (msg) {
                const content = JSON.parse(msg.content.toString());
                callback(content);
                this.channel.ack(msg);
            }
        });
    }
}

// 3. 事件总线
class EventBus {
    constructor() {
        this.events = new Map();
    }
    
    publish(event, data) {
        if (!this.events.has(event)) {
            return;
        }
        
        const handlers = this.events.get(event);
        handlers.forEach(handler => handler(data));
    }
    
    subscribe(event, handler) {
        if (!this.events.has(event)) {
            this.events.set(event, new Set());
        }
        
        this.events.get(event).add(handler);
    }
    
    unsubscribe(event, handler) {
        if (!this.events.has(event)) {
            return;
        }
        
        this.events.get(event).delete(handler);
    }
}

服务发现与负载均衡 ⚖️

// 1. 服务发现客户端
class ServiceDiscoveryClient {
    constructor(registry) {
        this.registry = registry;
        this.cache = new Map();
        this.cacheTimeout = 60000; // 1分钟缓存
    }
    
    async getService(name) {
        const now = Date.now();
        const cached = this.cache.get(name);
        
        if (cached && now - cached.timestamp < this.cacheTimeout) {
            return cached.service;
        }
        
        const service = this.registry.getService(name);
        this.cache.set(name, {
            service,
            timestamp: now
        });
        
        return service;
    }
}

// 2. 负载均衡器
class LoadBalancer {
    constructor() {
        this.algorithms = new Map();
        this.setupAlgorithms();
    }
    
    setupAlgorithms() {
        // 轮询算法
        this.algorithms.set('round-robin', {
            counter: new Map(),
            select: (services, serviceName) => {
                const count = this.algorithms.get('round-robin').counter;
                const current = count.get(serviceName) || 0;
                count.set(serviceName, (current + 1) % services.length);
                return services[current];
            }
        });
        
        // 随机算法
        this.algorithms.set('random', {
            select: (services) => {
                return services[Math.floor(Math.random() * services.length)];
            }
        });
        
        // 最少连接算法
        this.algorithms.set('least-connections', {
            connections: new Map(),
            select: (services) => {
                const conns = this.algorithms.get('least-connections').connections;
                return services.reduce((min, service) => {
                    const serviceConns = conns.get(service.id) || 0;
                    const minConns = conns.get(min.id) || 0;
                    return serviceConns < minConns ? service : min;
                });
            }
        });
    }
    
    select(algorithm, services, serviceName) {
        if (!this.algorithms.has(algorithm)) {
            throw new Error(`Unknown load balancing algorithm: ${algorithm}`);
        }
        
        return this.algorithms.get(algorithm).select(services, serviceName);
    }
}

容错与熔断机制 🔌

// 1. 断路器
class CircuitBreaker {
    constructor(options = {}) {
        this.failureThreshold = options.failureThreshold || 5;
        this.resetTimeout = options.resetTimeout || 60000;
        this.failures = 0;
        this.state = 'CLOSED';
        this.lastFailure = null;
    }
    
    async execute(fn) {
        if (this.state === 'OPEN') {
            if (Date.now() - this.lastFailure >= this.resetTimeout) {
                this.state = 'HALF-OPEN';
            } else {
                throw new Error('Circuit breaker is OPEN');
            }
        }
        
        try {
            const result = await fn();
            this.onSuccess();
            return result;
        } catch (error) {
            this.onFailure();
            throw error;
        }
    }
    
    onSuccess() {
        this.failures = 0;
        this.state = 'CLOSED';
    }
    
    onFailure() {
        this.failures++;
        this.lastFailure = Date.now();
        
        if (this.failures >= this.failureThreshold) {
            this.state = 'OPEN';
        }
    }
}

// 2. 重试机制
class RetryMechanism {
    constructor(options = {}) {
        this.maxRetries = options.maxRetries || 3;
        this.delay = options.delay || 1000;
        this.backoffFactor = options.backoffFactor || 2;
    }
    
    async execute(fn) {
        let retries = 0;
        let delay = this.delay;
        
        while (true) {
            try {
                return await fn();
            } catch (error) {
                retries++;
                
                if (retries >= this.maxRetries) {
                    throw error;
                }
                
                await new Promise(resolve => setTimeout(resolve, delay));
                delay *= this.backoffFactor;
            }
        }
    }
}

// 3. 超时控制
class TimeoutController {
    async execute(fn, timeout) {
        const timeoutPromise = new Promise((_, reject) => {
            setTimeout(() => reject(new Error('Operation timed out')), timeout);
        });
        
        return Promise.race([fn(), timeoutPromise]);
    }
}

监控与日志 📊

// 1. 性能监控
class PerformanceMonitor {
    constructor() {
        this.metrics = new Map();
        this.startTime = Date.now();
    }
    
    recordMetric(name, value) {
        if (!this.metrics.has(name)) {
            this.metrics.set(name, []);
        }
        
        this.metrics.get(name).push({
            timestamp: Date.now(),
            value
        });
    }
    
    getMetrics(name, timeRange) {
        const metrics = this.metrics.get(name) || [];
        const now = Date.now();
        
        return metrics.filter(m => now - m.timestamp <= timeRange);
    }
    
    calculateStatistics(name, timeRange) {
        const metrics = this.getMetrics(name, timeRange);
        const values = metrics.map(m => m.value);
        
        return {
            count: values.length,
            average: values.reduce((a, b) => a + b, 0) / values.length,
            min: Math.min(...values),
            max: Math.max(...values)
        };
    }
}

// 2. 分布式日志
class DistributedLogger {
    constructor(options = {}) {
        this.serviceName = options.serviceName;
        this.logLevel = options.logLevel || 'info';
        this.transport = options.transport || 'console';
    }
    
    log(level, message, metadata = {}) {
        const logEntry = {
            timestamp: new Date().toISOString(),
            level,
            service: this.serviceName,
            message,
            metadata,
            correlationId: metadata.correlationId || this.generateCorrelationId()
        };
        
        this.send(logEntry);
    }
    
    send(logEntry) {
        switch (this.transport) {
            case 'console':
                console.log(JSON.stringify(logEntry));
                break;
            case 'elasticsearch':
                // 实现Elasticsearch传输
                break;
            case 'fluentd':
                // 实现Fluentd传输
                break;
        }
    }
    
    generateCorrelationId() {
        return `${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
    }
}

实际应用示例 💼

// 1. 用户服务
class UserService extends MicroService {
    constructor() {
        super('user-service', 3000);
        this.setupRoutes();
    }
    
    setupRoutes() {
        this.registerRoute('GET', '/users', this.getUsers.bind(this));
        this.registerRoute('GET', '/users/:id', this.getUserById.bind(this));
        this.registerRoute('POST', '/users', this.createUser.bind(this));
    }
    
    async getUsers(req, res) {
        // 实现获取用户列表
    }
    
    async getUserById(req, res) {
        // 实现获取单个用户
    }
    
    async createUser(req, res) {
        // 实现创建用户
    }
}

// 2. 订单服务
class OrderService extends MicroService {
    constructor() {
        super('order-service', 3001);
        this.setupRoutes();
        this.userServiceClient = new ServiceDiscoveryClient(registry);
    }
    
    setupRoutes() {
        this.registerRoute('POST', '/orders', this.createOrder.bind(this));
        this.registerRoute('GET', '/orders/:id', this.getOrderById.bind(this));
    }
    
    async createOrder(req, res) {
        const { userId, items } = req.body;
        
        // 使用断路器调用用户服务
        const circuitBreaker = new CircuitBreaker();
        const user = await circuitBreaker.execute(async () => {
            const userService = await this.userServiceClient.getService('user-service');
            const communicator = new HttpCommunicator();
            return communicator.request(userService, 'GET', `/users/${userId}`);
        });
        
        // 创建订单逻辑
    }
}

// 3. API网关
class ApiGateway extends MicroService {
    constructor() {
        super('api-gateway', 8080);
        this.setupRoutes();
        this.loadBalancer = new LoadBalancer();
        this.registry = new ServiceRegistry();
    }
    
    setupRoutes() {
        this.app.use(this.routeRequest.bind(this));
    }
    
    async routeRequest(req, res) {
        const service = this.determineTargetService(req.path);
        const instances = await this.registry.getServiceInstances(service);
        
        if (instances.length === 0) {
            return res.status(503).json({ error: 'Service Unavailable' });
        }
        
        const target = this.loadBalancer.select('round-robin', instances, service);
        const communicator = new HttpCommunicator();
        
        try {
            const result = await communicator.request(target, req.method, req.path, req.body);
            res.json(result);
        } catch (error) {
            res.status(500).json({ error: error.message });
        }
    }
}