当前位置：首页 > article >正文

Node.js系列（4）--微服务架构实践

article 2025/3/20 12:41:10

Node.js微服务架构实践 🔄

引言

微服务架构已成为构建大规模Node.js应用的主流选择。本文将深入探讨Node.js微服务架构的设计与实现，包括服务拆分、服务治理、通信机制等方面，帮助开发者构建可扩展的微服务系统。

微服务架构概述

Node.js微服务架构主要包括以下方面：

服务拆分：业务领域划分与服务边界
服务治理：服务注册、发现与负载均衡
通信机制：同步与异步通信方案
数据管理：分布式事务与数据一致性
可观测性：监控、日志与链路追踪

微服务架构实现

服务注册中心

// 服务注册中心
class ServiceRegistry {
    private static instance: ServiceRegistry;
    private services: Map<string, ServiceInfo[]>;
    private config: RegistryConfig;
    private healthChecker: HealthChecker;
    
    private constructor() {
        this.services = new Map();
        this.config = {
            checkInterval: 10000,
            timeoutThreshold: 30000
        };
        this.healthChecker = new HealthChecker(this.config);
    }
    
    // 获取单例实例
    static getInstance(): ServiceRegistry {
        if (!ServiceRegistry.instance) {
            ServiceRegistry.instance = new ServiceRegistry();
        }
        return ServiceRegistry.instance;
    }
    
    // 注册服务
    registerService(serviceInfo: ServiceInfo): void {
        const { name, version } = serviceInfo;
        const key = `${name}@${version}`;
        
        if (!this.services.has(key)) {
            this.services.set(key, []);
        }
        
        this.services.get(key)!.push(serviceInfo);
        console.log(`Service registered: ${key}`);
        
        // 启动健康检查
        this.healthChecker.addService(serviceInfo);
    }
    
    // 注销服务
    deregisterService(serviceInfo: ServiceInfo): void {
        const { name, version } = serviceInfo;
        const key = `${name}@${version}`;
        
        const services = this.services.get(key);
        if (services) {
            const index = services.findIndex(
                s => s.instanceId === serviceInfo.instanceId
            );
            if (index !== -1) {
                services.splice(index, 1);
                console.log(`Service deregistered: ${key}`);
                
                // 停止健康检查
                this.healthChecker.removeService(serviceInfo);
            }
        }
    }
    
    // 发现服务
    discoverService(name: string, version: string): ServiceInfo[] {
        const key = `${name}@${version}`;
        return this.services.get(key) || [];
    }
    
    // 更新服务状态
    updateServiceStatus(
        serviceInfo: ServiceInfo,
        status: ServiceStatus
    ): void {
        const { name, version } = serviceInfo;
        const key = `${name}@${version}`;
        
        const services = this.services.get(key);
        if (services) {
            const service = services.find(
                s => s.instanceId === serviceInfo.instanceId
            );
            if (service) {
                service.status = status;
                service.lastUpdateTime = Date.now();
            }
        }
    }
    
    // 获取所有服务
    getAllServices(): Map<string, ServiceInfo[]> {
        return this.services;
    }
}

// 健康检查器
class HealthChecker {
    private config: RegistryConfig;
    private checkTimer: NodeJS.Timeout | null;
    private services: Set<ServiceInfo>;
    
    constructor(config: RegistryConfig) {
        this.config = config;
        this.checkTimer = null;
        this.services = new Set();
    }
    
    // 添加服务
    addService(serviceInfo: ServiceInfo): void {
        this.services.add(serviceInfo);
        
        if (!this.checkTimer) {
            this.startHealthCheck();
        }
    }
    
    // 移除服务
    removeService(serviceInfo: ServiceInfo): void {
        this.services.delete(serviceInfo);
        
        if (this.services.size === 0 && this.checkTimer) {
            this.stopHealthCheck();
        }
    }
    
    // 启动健康检查
    private startHealthCheck(): void {
        this.checkTimer = setInterval(() => {
            this.checkServices();
        }, this.config.checkInterval);
    }
    
    // 停止健康检查
    private stopHealthCheck(): void {
        if (this.checkTimer) {
            clearInterval(this.checkTimer);
            this.checkTimer = null;
        }
    }
    
    // 检查服务健康状态
    private async checkServices(): Promise<void> {
        const registry = ServiceRegistry.getInstance();
        
        for (const service of this.services) {
            try {
                const status = await this.checkServiceHealth(service);
                registry.updateServiceStatus(service, status);
            } catch (error) {
                console.error(
                    `Health check failed for service ${service.name}:`,
                    error
                );
                registry.updateServiceStatus(service, 'unhealthy');
            }
        }
    }
    
    // 检查单个服务健康状态
    private async checkServiceHealth(
        service: ServiceInfo
    ): Promise<ServiceStatus> {
        try {
            const response = await fetch(
                `${service.baseUrl}/health`,
                {
                    timeout: this.config.timeoutThreshold
                }
            );
            
            return response.ok ? 'healthy' : 'unhealthy';
        } catch (error) {
            return 'unhealthy';
        }
    }
}

// 服务发现客户端
class ServiceDiscoveryClient {
    private registry: ServiceRegistry;
    private loadBalancer: LoadBalancer;
    
    constructor() {
        this.registry = ServiceRegistry.getInstance();
        this.loadBalancer = new LoadBalancer();
    }
    
    // 获取服务实例
    async getServiceInstance(
        name: string,
        version: string
    ): Promise<ServiceInfo | null> {
        const services = this.registry.discoverService(name, version);
        
        // 过滤健康实例
        const healthyServices = services.filter(
            s => s.status === 'healthy'
        );
        
        if (healthyServices.length === 0) {
            return null;
        }
        
        // 使用负载均衡选择实例
        return this.loadBalancer.select(healthyServices);
    }
    
    // 调用服务
    async callService(
        name: string,
        version: string,
        path: string,
        options: RequestOptions = {}
    ): Promise<any> {
        const service = await this.getServiceInstance(name, version);
        if (!service) {
            throw new Error(`No healthy service instance found: ${name}@${version}`);
        }
        
        try {
            const response = await fetch(
                `${service.baseUrl}${path}`,
                {
                    ...options,
                    timeout: options.timeout || 5000
                }
            );
            
            if (!response.ok) {
                throw new Error(`Service call failed: ${response.statusText}`);
            }
            
            return await response.json();
        } catch (error) {
            // 标记服务不健康
            this.registry.updateServiceStatus(service, 'unhealthy');
            throw error;
        }
    }
}

// 负载均衡器
class LoadBalancer {
    private currentIndex: number;
    
    constructor() {
        this.currentIndex = 0;
    }
    
    // 选择服务实例
    select(services: ServiceInfo[]): ServiceInfo {
        if (services.length === 0) {
            throw new Error('No services available');
        }
        
        // 轮询算法
        const service = services[this.currentIndex];
        this.currentIndex = (this.currentIndex + 1) % services.length;
        
        return service;
    }
}

// 服务网关
class ServiceGateway {
    private registry: ServiceRegistry;
    private discoveryClient: ServiceDiscoveryClient;
    private routeConfig: RouteConfig[];
    
    constructor(routeConfig: RouteConfig[]) {
        this.registry = ServiceRegistry.getInstance();
        this.discoveryClient = new ServiceDiscoveryClient();
        this.routeConfig = routeConfig;
    }
    
    // 启动网关
    async start(port: number): Promise<void> {
        const app = express();
        
        // 配置中间件
        app.use(express.json());
        app.use(this.errorHandler.bind(this));
        
        // 注册路由
        this.registerRoutes(app);
        
        // 启动服务器
        app.listen(port, () => {
            console.log(`Gateway is running on port ${port}`);
        });
    }
    
    // 注册路由
    private registerRoutes(app: express.Application): void {
        for (const route of this.routeConfig) {
            app.use(
                route.path,
                this.createProxyMiddleware(route)
            );
        }
    }
    
    // 创建代理中间件
    private createProxyMiddleware(
        route: RouteConfig
    ): express.RequestHandler {
        return async (req, res, next) => {
            try {
                const response = await this.discoveryClient.callService(
                    route.service,
                    route.version,
                    req.path,
                    {
                        method: req.method,
                        headers: req.headers as any,
                        body: req.body
                    }
                );
                
                res.json(response);
            } catch (error) {
                next(error);
            }
        };
    }
    
    // 错误处理中间件
    private errorHandler(
        err: Error,
        req: express.Request,
        res: express.Response,
        next: express.NextFunction
    ): void {
        console.error('Gateway error:', err);
        
        res.status(500).json({
            error: 'Internal Server Error',
            message: err.message
        });
    }
}

// 接口定义
interface ServiceInfo {
    name: string;
    version: string;
    instanceId: string;
    baseUrl: string;
    status: ServiceStatus;
    lastUpdateTime: number;
    metadata?: Record<string, any>;
}

interface RegistryConfig {
    checkInterval: number;
    timeoutThreshold: number;
}

interface RouteConfig {
    path: string;
    service: string;
    version: string;
}

interface RequestOptions extends RequestInit {
    timeout?: number;
}

type ServiceStatus = 'healthy' | 'unhealthy';

// 使用示例
async function main() {
    // 创建服务注册中心
    const registry = ServiceRegistry.getInstance();
    
    // 注册服务
    registry.registerService({
        name: 'user-service',
        version: '1.0.0',
        instanceId: 'user-1',
        baseUrl: 'http://localhost:3001',
        status: 'healthy',
        lastUpdateTime: Date.now()
    });
    
    // 创建服务网关
    const gateway = new ServiceGateway([
        {
            path: '/api/users',
            service: 'user-service',
            version: '1.0.0'
        }
    ]);
    
    // 启动网关
    await gateway.start(3000);
    
    // 创建服务发现客户端
    const client = new ServiceDiscoveryClient();
    
    // 调用服务
    try {
        const result = await client.callService(
            'user-service',
            '1.0.0',
            '/users',
            { method: 'GET' }
        );
        console.log('Service call result:', result);
    } catch (error) {
        console.error('Service call failed:', error);
    }
}

main().catch(console.error);