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