📚 课程目标

通过本课程的学习，学员将能够：

• 深入理解微服务架构的核心设计原则
• 掌握服务拆分的方法和策略
• 学会服务间通信机制的设计
• 具备负载均衡与容错设计的能力

🎯 课程大纲

1. 微服务架构概述
2. 服务拆分原则与方法
3. 服务间通信机制
4. 负载均衡设计
5. 容错与熔断机制
6. 微服务治理

📖 课程内容

1. 微服务架构概述

微服务架构是一种将单一应用程序开发为一组小型服务的方法，每个服务运行在自己的进程中，并通过轻量级机制进行通信。

1.1 微服务架构特点

服务独立性

• 独立开发部署
• 独立技术栈
• 独立数据存储
• 独立团队负责

去中心化

• 数据管理去中心化
• 治理去中心化
• 技术栈去中心化
• 团队去中心化

故障隔离

• 服务间故障隔离
• 独立故障处理
• 快速故障恢复
• 系统整体可用性

1.2 微服务核心实现

服务注册与发现

import asyncio
import aiohttp
from typing import Dict, List, Optional
import json
import time

class ServiceRegistry:
    """服务注册中心"""
    def __init__(self):
        self.services: Dict[str, List[Dict]] = {}
        self.health_check_interval = 30
    
    async def register_service(self, service_name: str, service_info: Dict):
        """注册服务"""
        if service_name not in self.services:
            self.services[service_name] = []
        
        service_info['registered_at'] = time.time()
        self.services[service_name].append(service_info)
        print(f"服务 {service_name} 注册成功: {service_info}")
    
    async def discover_services(self, service_name: str) -> List[Dict]:
        """发现服务"""
        return self.services.get(service_name, [])
    
    async def health_check(self):
        """健康检查"""
        while True:
            for service_name, instances in self.services.items():
                for instance in instances[:]:  # 创建副本避免修改时迭代
                    try:
                        async with aiohttp.ClientSession() as session:
                            async with session.get(f"http://{instance['host']}:{instance['port']}/health") as resp:
                                if resp.status != 200:
                                    self.services[service_name].remove(instance)
                                    print(f"移除不健康服务: {instance}")
                    except Exception as e:
                        self.services[service_name].remove(instance)
                        print(f"健康检查失败，移除服务: {instance}, 错误: {e}")
            
            await asyncio.sleep(self.health_check_interval)

class ServiceProvider:
    """服务提供者"""
    def __init__(self, service_name: str, host: str, port: int, registry_url: str):
        self.service_name = service_name
        self.host = host
        self.port = port
        self.registry_url = registry_url
        self.registry = ServiceRegistry()
    
    async def start_service(self):
        """启动服务"""
        # 注册到注册中心
        service_info = {
            'host': self.host,
            'port': self.port,
            'status': 'healthy'
        }
        await self.registry.register_service(self.service_name, service_info)
        print(f"服务 {self.service_name} 启动在 {self.host}:{self.port}")

class ServiceConsumer:
    """服务消费者"""
    def __init__(self, registry_url: str):
        self.registry_url = registry_url
        self.registry = ServiceRegistry()
    
    async def call_service(self, service_name: str, endpoint: str, data: Dict = None):
        """调用服务"""
        # 发现服务
        services = await self.registry.discover_services(service_name)
        if not services:
            raise Exception(f"未找到服务: {service_name}")
        
        # 负载均衡选择服务
        service = self._load_balance(services)
        
        # 调用服务
        url = f"http://{service['host']}:{service['port']}/{endpoint}"
        async with aiohttp.ClientSession() as session:
            async with session.post(url, json=data) as resp:
                return await resp.json()
    
    def _load_balance(self, services: List[Dict]) -> Dict:
        """负载均衡策略 - 轮询"""
        # 简单的轮询策略
        return services[0]  # 实际应该维护状态

服务间通信实现

import asyncio
from abc import ABC, abstractmethod
from typing import Any, Dict, List
import aiohttp
import json

class ServiceCommunication:
    """服务间通信基类"""
    def __init__(self, service_name: str):
        self.service_name = service_name
        self.session = None
    
    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()
    
    @abstractmethod
    async def send_request(self, target_service: str, endpoint: str, data: Dict) -> Any:
        """发送请求"""
        pass

class HTTPCommunication(ServiceCommunication):
    """HTTP通信实现"""
    def __init__(self, service_name: str, registry: ServiceRegistry):
        super().__init__(service_name)
        self.registry = registry
    
    async def send_request(self, target_service: str, endpoint: str, data: Dict) -> Any:
        """发送HTTP请求"""
        # 发现目标服务
        services = await self.registry.discover_services(target_service)
        if not services:
            raise Exception(f"服务 {target_service} 不可用")
        
        # 选择服务实例
        service = services[0]  # 简化处理
        
        # 发送请求
        url = f"http://{service['host']}:{service['port']}/{endpoint}"
        async with self.session.post(url, json=data) as resp:
            if resp.status == 200:
                return await resp.json()
            else:
                raise Exception(f"请求失败: {resp.status}")

class MessageQueueCommunication(ServiceCommunication):
    """消息队列通信实现"""
    def __init__(self, service_name: str, queue_url: str):
        super().__init__(service_name)
        self.queue_url = queue_url
    
    async def send_request(self, target_service: str, endpoint: str, data: Dict) -> Any:
        """发送消息队列请求"""
        message = {
            'target_service': target_service,
            'endpoint': endpoint,
            'data': data,
            'source_service': self.service_name,
            'timestamp': time.time()
        }
        
        # 发送到消息队列
        async with self.session.post(f"{self.queue_url}/publish", json=message) as resp:
            if resp.status == 200:
                return await resp.json()
            else:
                raise Exception(f"消息发送失败: {resp.status}")

# 使用示例
async def main():
    # 创建注册中心
    registry = ServiceRegistry()
    
    # 启动健康检查
    asyncio.create_task(registry.health_check())
    
    # 创建服务提供者
    user_service = ServiceProvider("user-service", "localhost", 8001, "http://localhost:8000")
    await user_service.start_service()
    
    # 创建服务消费者
    consumer = ServiceConsumer("http://localhost:8000")
    
    # 调用服务
    try:
        result = await consumer.call_service("user-service", "api/users", {"action": "list"})
        print(f"调用结果: {result}")
    except Exception as e:
        print(f"调用失败: {e}")

if __name__ == "__main__":
    asyncio.run(main())

1.2 微服务 vs 单体架构

对比维度	单体架构	微服务架构
开发复杂度	简单	复杂
部署复杂度	简单	复杂
扩展性	垂直扩展	水平扩展
技术栈	统一	多样化
数据一致性	强一致性	最终一致性
故障影响	全局影响	局部影响
团队协作	集中式	分布式

1.3 微服务适用场景

适合微服务的场景

• 大型复杂应用
• 多团队开发
• 快速业务变化
• 技术栈多样化需求

不适合微服务的场景

• 小型简单应用
• 单体团队开发
• 性能要求极高
• 强一致性要求

2. 服务拆分原则与方法

2.1 服务拆分原则

单一职责原则

• 每个服务只负责一个业务功能
• 高内聚，低耦合
• 清晰的边界定义

业务能力拆分

• 按业务领域拆分
• 按用户界面拆分
• 按数据拆分

数据驱动拆分

• 按数据模型拆分
• 按数据访问模式拆分
• 按数据一致性需求拆分

2.2 服务拆分实现

服务边界分析器

from typing import List, Dict, Set, Tuple
from dataclasses import dataclass
import networkx as nx

@dataclass
class ServiceBoundary:
    """服务边界定义"""
    name: str
    domain: str
    responsibilities: List[str]
    data_entities: Set[str]
    dependencies: Set[str]
    team: str

class ServiceSplitter:
    """服务拆分器"""
    def __init__(self):
        self.coupling_matrix = {}
        self.dependency_graph = nx.DiGraph()
    
    def analyze_coupling(self, modules: List[Dict]) -> Dict[Tuple[str, str], float]:
        """分析模块间耦合度"""
        coupling_scores = {}
        
        for i, module_a in enumerate(modules):
            for j, module_b in enumerate(modules):
                if i != j:
                    score = self._calculate_coupling_score(module_a, module_b)
                    coupling_scores[(module_a['name'], module_b['name'])] = score
                    
                    # 构建依赖图
                    if score > 0.3:  # 高耦合阈值
                        self.dependency_graph.add_edge(
                            module_a['name'], 
                            module_b['name'], 
                            weight=score
                        )
        
        return coupling_scores
    
    def _calculate_coupling_score(self, module_a: Dict, module_b: Dict) -> float:
        """计算耦合度分数"""
        score = 0.0
        
        # 数据耦合 (30%)
        shared_data = set(module_a.get('data_entities', [])) & set(module_b.get('data_entities', []))
        score += len(shared_data) * 0.3
        
        # 功能耦合 (20%)
        shared_functions = set(module_a.get('functions', [])) & set(module_b.get('functions', []))
        score += len(shared_functions) * 0.2
        
        # 接口耦合 (10%)
        interface_calls = module_a.get('calls', {}).get(module_b['name'], 0)
        score += interface_calls * 0.1
        
        # 团队耦合 (40%)
        if module_a.get('team') == module_b.get('team'):
            score += 0.4
        
        return min(score, 1.0)
    
    def suggest_split(self, modules: List[Dict], threshold: float = 0.5) -> List[ServiceBoundary]:
        """建议服务拆分"""
        coupling_scores = self.analyze_coupling(modules)
        
        # 找出高耦合的模块对
        high_coupling_pairs = [
            (pair, score) for pair, score in coupling_scores.items()
            if score > threshold
        ]
        
        # 基于耦合度进行分组
        service_groups = self._group_high_coupling_modules(modules, high_coupling_pairs)
        
        # 生成服务边界
        services = []
        for group in service_groups:
            service = self._create_service_from_group(group)
            services.append(service)
        
        return services
    
    def _group_high_coupling_modules(self, modules: List[Dict], high_coupling_pairs: List) -> List[List[Dict]]:
        """将高耦合模块分组"""
        groups = []
        processed = set()
        
        for module in modules:
            if module['name'] in processed:
                continue
            
            group = [module]
            processed.add(module['name'])
            
            # 找到与当前模块高耦合的其他模块
            for (mod_a, mod_b), score in high_coupling_pairs:
                if mod_a == module['name'] and mod_b not in processed:
                    other_module = next(m for m in modules if m['name'] == mod_b)
                    group.append(other_module)
                    processed.add(mod_b)
                elif mod_b == module['name'] and mod_a not in processed:
                    other_module = next(m for m in modules if m['name'] == mod_a)
                    group.append(other_module)
                    processed.add(mod_a)
            
            groups.append(group)
        
        return groups
    
    def _create_service_from_group(self, group: List[Dict]) -> ServiceBoundary:
        """从模块组创建服务"""
        service_name = f"service_{len(self.coupling_matrix) + 1}"
        domain = group[0].get('domain', 'unknown')
        
        # 合并所有职责和数据实体
        all_responsibilities = []
        all_data_entities = set()
        all_dependencies = set()
        
        for module in group:
            all_responsibilities.extend(module.get('responsibilities', []))
            all_data_entities.update(module.get('data_entities', []))
            all_dependencies.update(module.get('dependencies', []))
        
        return ServiceBoundary(
            name=service_name,
            domain=domain,
            responsibilities=all_responsibilities,
            data_entities=all_data_entities,
            dependencies=all_dependencies,
            team=group[0].get('team', 'unknown')
        )

# 使用示例
modules = [
    {
        'name': 'user_management',
        'domain': 'user',
        'responsibilities': ['user_registration', 'user_authentication'],
        'data_entities': ['users', 'user_profiles'],
        'functions': ['create_user', 'authenticate_user'],
        'team': 'team_a',
        'calls': {'order_management': 5}
    },
    {
        'name': 'order_processing',
        'domain': 'order',
        'responsibilities': ['order_creation', 'order_fulfillment'],
        'data_entities': ['orders', 'order_items'],
        'functions': ['create_order', 'process_payment'],
        'team': 'team_b',
        'calls': {'user_management': 3, 'inventory_management': 8}
    },
    {
        'name': 'inventory_management',
        'domain': 'inventory',
        'responsibilities': ['stock_management', 'inventory_tracking'],
        'data_entities': ['products', 'inventory'],
        'functions': ['update_stock', 'check_availability'],
        'team': 'team_c',
        'calls': {'order_processing': 2}
    }
]

# 执行拆分
splitter = ServiceSplitter()
services = splitter.suggest_split(modules, threshold=0.3)

for service in services:
    print(f"服务: {service.name}")
    print(f"领域: {service.domain}")
    print(f"职责: {service.responsibilities}")
    print(f"数据实体: {list(service.data_entities)}")
    print("---")

熔断器模式实现

import asyncio
import time
from enum import Enum
from typing import Callable, Any
import logging

class CircuitBreakerState(Enum):
    CLOSED = "closed"      # 关闭状态 - 正常调用
    OPEN = "open"          # 打开状态 - 快速失败
    HALF_OPEN = "half_open"  # 半开状态 - 试探调用

class CircuitBreaker:
    """熔断器实现"""
    def __init__(self, 
                 failure_threshold: int = 5,
                 timeout: int = 60,
                 expected_exception: Exception = Exception):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.expected_exception = expected_exception
        
        self.failure_count = 0
        self.last_failure_time = None
        self.state = CircuitBreakerState.CLOSED
        
        self.logger = logging.getLogger(__name__)
    
    async def call(self, func: Callable, *args, **kwargs) -> Any:
        """调用服务"""
        if self.state == CircuitBreakerState.OPEN:
            if time.time() - self.last_failure_time > self.timeout:
                self.state = CircuitBreakerState.HALF_OPEN
                self.logger.info("熔断器进入半开状态")
            else:
                raise Exception("熔断器处于打开状态，服务不可用")
        
        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except self.expected_exception as e:
            self._on_failure()
            raise e
    
    def _on_success(self):
        """成功回调"""
        self.failure_count = 0
        self.state = CircuitBreakerState.CLOSED
        self.logger.info("熔断器重置为关闭状态")
    
    def _on_failure(self):
        """失败回调"""
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitBreakerState.OPEN
            self.logger.warning(f"熔断器打开，失败次数: {self.failure_count}")

# 使用示例
async def unreliable_service_call():
    """模拟不可靠的服务调用"""
    import random
    if random.random() < 0.7:  # 70% 失败率
        raise Exception("服务调用失败")
    return "服务调用成功"

async def main():
    # 创建熔断器
    circuit_breaker = CircuitBreaker(failure_threshold=3, timeout=10)
    
    # 测试熔断器
    for i in range(10):
        try:
            result = await circuit_breaker.call(unreliable_service_call)
            print(f"调用 {i+1}: {result}")
        except Exception as e:
            print(f"调用 {i+1}: {e}")
        
        await asyncio.sleep(1)

if __name__ == "__main__":
    asyncio.run(main())

2.2 拆分策略

水平拆分

• 按功能模块拆分
• 按业务流程拆分
• 按用户群体拆分

核心代码实现

# 微服务拆分策略实现
from abc import ABC, abstractmethod
from typing import List, Dict, Any
import asyncio
import aiohttp

class ServiceBoundary:
    """服务边界定义"""
    def __init__(self, name: str, domain: str, responsibilities: List[str]):
        self.name = name
        self.domain = domain
        self.responsibilities = responsibilities
        self.dependencies = []
        self.data_ownership = []
    
    def add_dependency(self, service_name: str, dependency_type: str):
        """添加服务依赖"""
        self.dependencies.append({
            'service': service_name,
            'type': dependency_type  # 'data', 'function', 'event'
        })
    
    def add_data_ownership(self, data_type: str, access_pattern: str):
        """添加数据所有权"""
        self.data_ownership.append({
            'type': data_type,
            'access_pattern': access_pattern
        })

class ServiceSplitter:
    """服务拆分器"""
    def __init__(self):
        self.services = {}
        self.coupling_matrix = {}
    
    def analyze_coupling(self, modules: List[Dict[str, Any]]):
        """分析模块间耦合度"""
        coupling_scores = {}
        
        for i, module_a in enumerate(modules):
            for j, module_b in enumerate(modules):
                if i != j:
                    score = self._calculate_coupling_score(module_a, module_b)
                    coupling_scores[(module_a['name'], module_b['name'])] = score
        
        return coupling_scores
    
    def _calculate_coupling_score(self, module_a: Dict, module_b: Dict) -> float:
        """计算耦合度分数"""
        score = 0.0
        
        # 数据耦合
        shared_data = set(module_a.get('data_entities', [])) & set(module_b.get('data_entities', []))
        score += len(shared_data) * 0.3
        
        # 功能耦合
        shared_functions = set(module_a.get('functions', [])) & set(module_b.get('functions', []))
        score += len(shared_functions) * 0.2
        
        # 接口耦合
        interface_calls = module_a.get('calls', {}).get(module_b['name'], 0)
        score += interface_calls * 0.1
        
        # 团队耦合
        if module_a.get('team') == module_b.get('team'):
            score += 0.4
        
        return min(score, 1.0)
    
    def suggest_split(self, modules: List[Dict[str, Any]], threshold: float = 0.5):
        """建议服务拆分"""
        coupling_scores = self.analyze_coupling(modules)
        
        # 找出高耦合的模块对
        high_coupling_pairs = [
            (pair, score) for pair, score in coupling_scores.items()
            if score > threshold
        ]
        
        # 基于耦合度进行分组
        service_groups = self._group_high_coupling_modules(modules, high_coupling_pairs)
        
        # 生成服务边界
        services = []
        for group in service_groups:
            service = self._create_service_from_group(group)
            services.append(service)
        
        return services
    
    def _group_high_coupling_modules(self, modules: List[Dict], high_coupling_pairs: List):
        """将高耦合模块分组"""
        groups = []
        processed = set()
        
        for module in modules:
            if module['name'] in processed:
                continue
            
            group = [module]
            processed.add(module['name'])
            
            # 找到与当前模块高耦合的其他模块
            for (mod_a, mod_b), score in high_coupling_pairs:
                if mod_a == module['name'] and mod_b not in processed:
                    other_module = next(m for m in modules if m['name'] == mod_b)
                    group.append(other_module)
                    processed.add(mod_b)
                elif mod_b == module['name'] and mod_a not in processed:
                    other_module = next(m for m in modules if m['name'] == mod_a)
                    group.append(other_module)
                    processed.add(mod_a)
            
            groups.append(group)
        
        return groups
    
    def _create_service_from_group(self, group: List[Dict]) -> ServiceBoundary:
        """从模块组创建服务"""
        service_name = f"service_{len(self.services) + 1}"
        domain = group[0].get('domain', 'unknown')
        
        # 合并所有职责
        all_responsibilities = []
        for module in group:
            all_responsibilities.extend(module.get('responsibilities', []))
        
        service = ServiceBoundary(service_name, domain, all_responsibilities)
        
        # 添加数据所有权
        for module in group:
            for data_entity in module.get('data_entities', []):
                service.add_data_ownership(data_entity, 'read_write')
        
        return service

# 垂直拆分实现
class VerticalSplitter:
    """垂直拆分器 - 按业务领域拆分"""
    def __init__(self):
        self.business_domains = {}
    
    def identify_domains(self, business_requirements: List[Dict[str, Any]]):
        """识别业务领域"""
        domains = {}
        
        for req in business_requirements:
            domain = req.get('domain', 'general')
            if domain not in domains:
                domains[domain] = {
                    'name': domain,
                    'requirements': [],
                    'entities': set(),
                    'use_cases': []
                }
            
            domains[domain]['requirements'].append(req)
            domains[domain]['entities'].update(req.get('entities', []))
            domains[domain]['use_cases'].extend(req.get('use_cases', []))
        
        return domains
    
    def create_services_by_domain(self, domains: Dict[str, Any]):
        """按领域创建服务"""
        services = []
        
        for domain_name, domain_info in domains.items():
            service = ServiceBoundary(
                name=f"{domain_name}_service",
                domain=domain_name,
                responsibilities=domain_info['use_cases']
            )
            
            # 添加数据所有权
            for entity in domain_info['entities']:
                service.add_data_ownership(entity, 'read_write')
            
            services.append(service)
        
        return services

# 混合拆分实现
class HybridSplitter:
    """混合拆分器"""
    def __init__(self):
        self.horizontal_splitter = ServiceSplitter()
        self.vertical_splitter = VerticalSplitter()
    
    def split_services(self, 
                      modules: List[Dict[str, Any]], 
                      business_requirements: List[Dict[str, Any]],
                      strategy: str = 'hybrid'):
        """混合拆分服务"""
        if strategy == 'horizontal':
            return self.horizontal_splitter.suggest_split(modules)
        elif strategy == 'vertical':
            domains = self.vertical_splitter.identify_domains(business_requirements)
            return self.vertical_splitter.create_services_by_domain(domains)
        else:  # hybrid
            # 先按垂直拆分
            domains = self.vertical_splitter.identify_domains(business_requirements)
            vertical_services = self.vertical_splitter.create_services_by_domain(domains)
            
            # 再按水平拆分细化
            refined_services = []
            for service in vertical_services:
                # 将服务内部的模块进一步拆分
                internal_modules = self._extract_internal_modules(service, modules)
                if len(internal_modules) > 1:
                    sub_services = self.horizontal_splitter.suggest_split(internal_modules)
                    refined_services.extend(sub_services)
                else:
                    refined_services.append(service)
            
            return refined_services
    
    def _extract_internal_modules(self, service: ServiceBoundary, modules: List[Dict]):
        """提取服务内部模块"""
        internal_modules = []
        for module in modules:
            if module.get('domain') == service.domain:
                internal_modules.append(module)
        return internal_modules

# 使用示例
modules = [
    {
        'name': 'user_management',
        'domain': 'user',
        'responsibilities': ['user_registration', 'user_authentication'],
        'data_entities': ['users', 'user_profiles'],
        'functions': ['create_user', 'authenticate_user'],
        'team': 'team_a'
    },
    {
        'name': 'order_processing',
        'domain': 'order',
        'responsibilities': ['order_creation', 'order_fulfillment'],
        'data_entities': ['orders', 'order_items'],
        'functions': ['create_order', 'process_payment'],
        'team': 'team_b'
    },
    {
        'name': 'inventory_management',
        'domain': 'inventory',
        'responsibilities': ['stock_management', 'inventory_tracking'],
        'data_entities': ['products', 'inventory'],
        'functions': ['update_stock', 'check_availability'],
        'team': 'team_c'
    }
]

business_requirements = [
    {
        'domain': 'user',
        'entities': ['users', 'user_profiles'],
        'use_cases': ['user_registration', 'user_authentication']
    },
    {
        'domain': 'order',
        'entities': ['orders', 'order_items'],
        'use_cases': ['order_creation', 'order_fulfillment']
    }
]

# 执行拆分
splitter = HybridSplitter()
services = splitter.split_services(modules, business_requirements, strategy='hybrid')

for service in services:
    print(f"服务: {service.name}")
    print(f"领域: {service.domain}")
    print(f"职责: {service.responsibilities}")
    print(f"数据所有权: {[d['type'] for d in service.data_ownership]}")
    print("---")

垂直拆分

• 按业务领域拆分
• 按数据边界拆分
• 按团队边界拆分

混合拆分

• 结合多种拆分策略
• 动态调整拆分粒度
• 渐进式拆分

2.3 拆分实践案例

电商系统拆分示例

3. 服务间通信机制

3.1 通信方式选择

同步通信

• HTTP/REST：简单易用
• gRPC：高性能RPC
• GraphQL：灵活查询

异步通信

• 消息队列：解耦通信
• 事件驱动：事件发布订阅
• 流处理：实时数据流

3.2 API设计原则

RESTful设计

• 资源导向
• 无状态
• 统一接口
• 分层系统

版本管理

• URL版本控制
• Header版本控制
• 向后兼容
• 渐进式升级

错误处理

• 统一错误格式
• HTTP状态码
• 错误详细信息
• 错误处理机制

3.3 服务发现与注册

服务注册中心

• Consul：服务发现和配置
• Eureka：Netflix服务发现
• etcd：分布式键值存储

服务发现模式

4. 负载均衡设计

4.1 负载均衡策略

轮询策略

• 简单轮询
• 加权轮询
• 平滑加权轮询

最少连接策略

• 选择连接数最少的服务
• 动态调整权重
• 实时监控连接数

一致性哈希

• 基于请求特征
• 减少数据迁移
• 支持动态扩容

4.2 负载均衡器类型

硬件负载均衡器

• F5 BIG-IP
• Citrix NetScaler
• 高性能硬件

软件负载均衡器

• Nginx
• HAProxy
• Apache HTTP Server

云负载均衡器

• AWS ELB
• Azure Load Balancer
• 阿里云SLB

4.3 健康检查机制

检查类型

• HTTP检查：发送HTTP请求
• TCP检查：建立TCP连接
• 自定义检查：业务逻辑检查

检查策略

5. 容错与熔断机制

5.1 容错模式

重试机制

• 指数退避
• 最大重试次数
• 重试条件判断

超时机制

• 连接超时
• 读取超时
• 全局超时

降级策略

• 功能降级
• 服务降级
• 数据降级

5.2 熔断器模式

熔断器状态

• 关闭状态：正常调用
• 打开状态：快速失败
• 半开状态：试探调用

熔断器实现

5.3 限流与隔离

限流策略

• 令牌桶：平滑限流
• 漏桶：固定速率
• 滑动窗口：动态限流

隔离策略

• 线程池隔离：资源隔离
• 信号量隔离：并发控制
• 舱壁模式：故障隔离

6. 微服务治理

6.1 配置管理

配置中心

• Apollo：携程配置中心
• Nacos：阿里配置中心
• Consul：服务配置

配置管理策略

6.2 监控与日志

监控指标

• 系统指标：CPU、内存、磁盘
• 应用指标：QPS、响应时间、错误率
• 业务指标：用户数、订单数、收入

日志管理

• 集中日志：ELK Stack
• 结构化日志：JSON格式
• 日志分析：实时分析

6.3 链路追踪

分布式追踪

• Jaeger：Uber开源
• Zipkin：Twitter开源
• SkyWalking：Apache项目

追踪原理

📝 课程总结

微服务架构通过服务拆分、独立部署、故障隔离等机制，提供了更好的可扩展性和可维护性。掌握微服务架构的设计原则，是构建大型分布式系统的重要基础。

关键要点回顾：

1. 微服务架构强调服务的独立性和去中心化
2. 服务拆分需要遵循单一职责和业务边界原则
3. 服务间通信需要考虑同步和异步两种方式
4. 容错和熔断机制是保证系统稳定性的关键