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