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

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