sa-lessons/modules/module4/theory.md

Модуль 4: Масштабируемость и отказоустойчивость

Содержание модуля

1. Основы масштабируемости
2. Паттерны масштабирования
3. Отказоустойчивость
4. Управление данными
5. Кэширование
6. Load Balancing
7. Мониторинг производительности

---

Основы масштабируемости

Что такое масштабируемость

Масштабируемость — способность системы обрабатывать увеличение нагрузки путем добавления ресурсов.

Типы масштабирования

1. Вертикальное масштабирование (Scale Up)

Определение: Увеличение мощности существующих серверов.

Преимущества:

• Простота реализации
• Нет изменений в архитектуре
• Быстрое внедрение

Недостатки:

• Ограниченность ресурсов
• Высокая стоимость
• Точка отказа

Пример:

Сервер: 4 CPU, 8GB RAM → 8 CPU, 16GB RAM

2. Горизонтальное масштабирование (Scale Out)

Определение: Добавление новых серверов в систему.

Преимущества:

• Теоретически неограниченное масштабирование
• Отказоустойчивость
• Экономическая эффективность

Недостатки:

• Сложность архитектуры
• Проблемы с данными
• Сетевые задержки

Пример:

1 сервер → 2 сервера → 4 сервера → 8 серверов

Метрики масштабируемости

1. Throughput (Пропускная способность)

• Запросов в секунду (RPS)
• Транзакций в секунду (TPS)
• Байт в секунду

2. Latency (Задержка)

• Время отклика
• Время обработки
• Время ожидания

3. Resource Utilization (Использование ресурсов)

• CPU utilization
• Memory usage
• Disk I/O
• Network bandwidth

Закон Амдала

Формула: Speedup = 1 / (P + (1-P)/N)

Где:

• P — доля последовательного кода
• N — количество процессоров

Вывод: Параллелизация эффективна только при малой доле последовательного кода.

Принципы проектирования масштабируемых систем

1. Stateless Services (Бессостоятельные сервисы)

Принцип: Сервисы не должны хранить состояние между запросами.

Преимущества:

• Легкое горизонтальное масштабирование
• Отказоустойчивость
• Простота балансировки нагрузки

Реализация:

// Плохо: Stateful сервис
@Service
public class StatefulUserService {
    private Map<String, User> userCache = new HashMap<>();
    
    public User getUser(String id) {
        // Состояние хранится в памяти сервиса
        return userCache.get(id);
    }
}

// Хорошо: Stateless сервис
@Service
public class StatelessUserService {
    @Autowired
    private UserRepository userRepository;
    
    @Autowired
    private RedisTemplate<String, User> redisTemplate;
    
    public User getUser(String id) {
        // Состояние хранится во внешнем хранилище
        User cached = redisTemplate.opsForValue().get("user:" + id);
        if (cached != null) {
            return cached;
        }
        
        User user = userRepository.findById(id);
        redisTemplate.opsForValue().set("user:" + id, user, Duration.ofMinutes(30));
        return user;
    }
}

Rust примеры - Stateless сервисы:

use std::sync::Arc;
use tokio::sync::RwLock;
use redis::Client;

pub struct StatelessUserService {
    user_repository: Arc<dyn UserRepository>,
    redis_client: Arc<Client>,
}

impl StatelessUserService {
    pub fn new(user_repository: Arc<dyn UserRepository>, redis_client: Arc<Client>) -> Self {
        Self {
            user_repository,
            redis_client,
        }
    }
    
    pub async fn get_user(&self, id: String) -> Result<Option<User>, ServiceError> {
        // Проверка кэша
        if let Ok(user) = self.get_user_from_cache(&id).await {
            return Ok(Some(user));
        }
        
        // Получение из базы данных
        let user = self.user_repository.find_by_id(&id).await?;
        
        if let Some(ref user) = user {
            // Сохранение в кэш
            self.save_user_to_cache(user).await?;
        }
        
        Ok(user)
    }
    
    async fn get_user_from_cache(&self, id: &str) -> Result<User, ServiceError> {
        let mut conn = self.redis_client.get_async_connection().await?;
        let key = format!("user:{}", id);
        
        let user_json: String = redis::cmd("GET")
            .arg(&key)
            .query_async(&mut conn)
            .await?;
        
        let user: User = serde_json::from_str(&user_json)?;
        Ok(user)
    }
    
    async fn save_user_to_cache(&self, user: &User) -> Result<(), ServiceError> {
        let mut conn = self.redis_client.get_async_connection().await?;
        let key = format!("user:{}", user.id);
        let user_json = serde_json::to_string(user)?;
        
        redis::cmd("SETEX")
            .arg(&key)
            .arg(1800) // 30 минут
            .arg(&user_json)
            .query_async(&mut conn)
            .await?;
        
        Ok(())
    }
}

Python примеры - Stateless сервисы:

import asyncio
from typing import Optional
import redis.asyncio as redis
from dataclasses import dataclass
import json

@dataclass
class User:
    id: str
    name: str
    email: str

class StatelessUserService:
    def __init__(self, user_repository, redis_client: redis.Redis):
        self.user_repository = user_repository
        self.redis_client = redis_client
    
    async def get_user(self, user_id: str) -> Optional[User]:
        # Проверка кэша
        cached_user = await self._get_user_from_cache(user_id)
        if cached_user:
            return cached_user
        
        # Получение из базы данных
        user = await self.user_repository.find_by_id(user_id)
        
        if user:
            # Сохранение в кэш
            await self._save_user_to_cache(user)
        
        return user
    
    async def _get_user_from_cache(self, user_id: str) -> Optional[User]:
        try:
            key = f"user:{user_id}"
            user_json = await self.redis_client.get(key)
            
            if user_json:
                user_data = json.loads(user_json)
                return User(**user_data)
        except Exception as e:
            print(f"Cache error: {e}")
        
        return None
    
    async def _save_user_to_cache(self, user: User) -> None:
        try:
            key = f"user:{user.id}"
            user_json = json.dumps({
                'id': user.id,
                'name': user.name,
                'email': user.email
            })
            
            await self.redis_client.setex(key, 1800, user_json)  # 30 минут
        except Exception as e:
            print(f"Cache save error: {e}")

Node.js примеры - Stateless сервисы:

import { Redis } from 'ioredis';
import { UserRepository } from './user-repository';

interface User {
  id: string;
  name: string;
  email: string;
}

export class StatelessUserService {
  constructor(
    private userRepository: UserRepository,
    private redisClient: Redis
  ) {}
  
  async getUser(userId: string): Promise<User | null> {
    // Проверка кэша
    const cachedUser = await this.getUserFromCache(userId);
    if (cachedUser) {
      return cachedUser;
    }
    
    // Получение из базы данных
    const user = await this.userRepository.findById(userId);
    
    if (user) {
      // Сохранение в кэш
      await this.saveUserToCache(user);
    }
    
    return user;
  }
  
  private async getUserFromCache(userId: string): Promise<User | null> {
    try {
      const key = `user:${userId}`;
      const userJson = await this.redisClient.get(key);
      
      if (userJson) {
        return JSON.parse(userJson);
      }
    } catch (error) {
      console.error('Cache error:', error);
    }
    
    return null;
  }
  
  private async saveUserToCache(user: User): Promise<void> {
    try {
      const key = `user:${user.id}`;
      const userJson = JSON.stringify(user);
      
      await this.redisClient.setex(key, 1800, userJson); // 30 минут
    } catch (error) {
      console.error('Cache save error:', error);
    }
  }
}

2. Database Sharding (Шардирование базы данных)

Принцип: Разделение данных на несколько независимых баз данных.

Стратегии шардирования:

// Стратегия шардирования по пользователю
@Service
public class ShardedUserService {
    
    @Autowired
    private List<DataSource> shardDataSources;
    
    public DataSource getShardDataSource(String userId) {
        int shardIndex = Math.abs(userId.hashCode()) % shardDataSources.size();
        return shardDataSources.get(shardIndex);
    }
    
    public User saveUser(User user) {
        DataSource shard = getShardDataSource(user.getId());
        JdbcTemplate template = new JdbcTemplate(shard);
        
        String sql = "INSERT INTO users (id, name, email) VALUES (?, ?, ?)";
        template.update(sql, user.getId(), user.getName(), user.getEmail());
        
        return user;
    }
    
    public User getUser(String userId) {
        DataSource shard = getShardDataSource(userId);
        JdbcTemplate template = new JdbcTemplate(shard);
        
        String sql = "SELECT * FROM users WHERE id = ?";
        return template.queryForObject(sql, new Object[]{userId}, new UserRowMapper());
    }
}

// Стратегия шардирования по диапазону
@Service
public class RangeShardedUserService {
    
    private final Map<String, DataSource> shardMap = Map.of(
        "shard1", createDataSource("db_shard1"),
        "shard2", createDataSource("db_shard2"),
        "shard3", createDataSource("db_shard3")
    );
    
    public String getShardName(String userId) {
        int hash = userId.hashCode();
        if (hash < -1000000000) return "shard1";
        if (hash < 0) return "shard2";
        return "shard3";
    }
    
    public DataSource getShardDataSource(String userId) {
        String shardName = getShardName(userId);
        return shardMap.get(shardName);
    }
}

3. Read Replicas (Реплики для чтения)

Принцип: Использование отдельных реплик для операций чтения.

@Configuration
public class DatabaseConfig {
    
    @Bean
    @Primary
    public DataSource masterDataSource() {
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl("jdbc:postgresql://master-db:5432/mydb");
        config.setUsername("master_user");
        config.setPassword("master_password");
        return new HikariDataSource(config);
    }
    
    @Bean
    public DataSource readOnlyDataSource() {
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl("jdbc:postgresql://replica-db:5432/mydb");
        config.setUsername("readonly_user");
        config.setPassword("readonly_password");
        return new HikariDataSource(config);
    }
}

@Service
public class ReadWriteUserService {
    
    @Autowired
    @Qualifier("masterDataSource")
    private DataSource masterDataSource;
    
    @Autowired
    @Qualifier("readOnlyDataSource")
    private DataSource readOnlyDataSource;
    
    public User saveUser(User user) {
        // Запись в master
        JdbcTemplate masterTemplate = new JdbcTemplate(masterDataSource);
        String sql = "INSERT INTO users (id, name, email) VALUES (?, ?, ?)";
        masterTemplate.update(sql, user.getId(), user.getName(), user.getEmail());
        return user;
    }
    
    public User getUser(String userId) {
        // Чтение из replica
        JdbcTemplate replicaTemplate = new JdbcTemplate(readOnlyDataSource);
        String sql = "SELECT * FROM users WHERE id = ?";
        return replicaTemplate.queryForObject(sql, new Object[]{userId}, new UserRowMapper());
    }
    
    public List<User> getAllUsers() {
        // Чтение из replica
        JdbcTemplate replicaTemplate = new JdbcTemplate(readOnlyDataSource);
        String sql = "SELECT * FROM users";
        return replicaTemplate.query(sql, new UserRowMapper());
    }
}

4. Caching Strategies (Стратегии кэширования)

Cache-Aside Pattern:

@Service
public class CacheAsideUserService {
    
    @Autowired
    private UserRepository userRepository;
    
    @Autowired
    private RedisTemplate<String, User> redisTemplate;
    
    public User getUser(String userId) {
        // 1. Проверка кэша
        User cached = redisTemplate.opsForValue().get("user:" + userId);
        if (cached != null) {
            return cached;
        }
        
        // 2. Загрузка из БД
        User user = userRepository.findById(userId);
        
        // 3. Сохранение в кэш
        if (user != null) {
            redisTemplate.opsForValue().set("user:" + userId, user, Duration.ofMinutes(30));
        }
        
        return user;
    }
    
    public User updateUser(User user) {
        // 1. Обновление в БД
        User updated = userRepository.save(user);
        
        // 2. Обновление кэша
        redisTemplate.opsForValue().set("user:" + user.getId(), updated, Duration.ofMinutes(30));
        
        return updated;
    }
    
    public void deleteUser(String userId) {
        // 1. Удаление из БД
        userRepository.deleteById(userId);
        
        // 2. Удаление из кэша
        redisTemplate.delete("user:" + userId);
    }
}

Write-Through Pattern:

@Service
public class WriteThroughUserService {
    
    @Autowired
    private UserRepository userRepository;
    
    @Autowired
    private RedisTemplate<String, User> redisTemplate;
    
    public User saveUser(User user) {
        // 1. Запись в БД
        User saved = userRepository.save(user);
        
        // 2. Запись в кэш
        redisTemplate.opsForValue().set("user:" + saved.getId(), saved, Duration.ofMinutes(30));
        
        return saved;
    }
    
    public User getUser(String userId) {
        // Чтение из кэша (данные всегда актуальны)
        return redisTemplate.opsForValue().get("user:" + userId);
    }
}

Write-Behind Pattern:

@Service
public class WriteBehindUserService {
    
    @Autowired
    private UserRepository userRepository;
    
    @Autowired
    private RedisTemplate<String, User> redisTemplate;
    
    @Async
    public CompletableFuture<User> saveUser(User user) {
        // 1. Быстрая запись в кэш
        redisTemplate.opsForValue().set("user:" + user.getId(), user, Duration.ofMinutes(30));
        
        // 2. Асинхронная запись в БД
        return CompletableFuture.supplyAsync(() -> {
            return userRepository.save(user);
        });
    }
    
    public User getUser(String userId) {
        // Чтение из кэша
        return redisTemplate.opsForValue().get("user:" + userId);
    }
    
    @Scheduled(fixedDelay = 30000) // Каждые 30 секунд
    public void flushCacheToDatabase() {
        // Периодическая синхронизация кэша с БД
        Set<String> keys = redisTemplate.keys("user:*");
        for (String key : keys) {
            User user = redisTemplate.opsForValue().get(key);
            if (user != null) {
                userRepository.save(user);
            }
        }
    }
}

Архитектурные принципы масштабируемости

1. Microservices Architecture

Принципы:

• Разделение по бизнес-доменам
• Независимое развертывание
• Слабая связанность
• Высокая когезия

// Пример микросервиса пользователей
@RestController
@RequestMapping("/api/users")
public class UserController {
    
    @Autowired
    private UserService userService;
    
    @GetMapping("/{id}")
    public ResponseEntity<User> getUser(@PathVariable String id) {
        User user = userService.getUser(id);
        return ResponseEntity.ok(user);
    }
    
    @PostMapping
    public ResponseEntity<User> createUser(@RequestBody CreateUserRequest request) {
        User user = userService.createUser(request);
        return ResponseEntity.status(HttpStatus.CREATED).body(user);
    }
}

@Service
public class UserService {
    
    @Autowired
    private UserRepository userRepository;
    
    @Autowired
    private NotificationService notificationService;
    
    public User createUser(CreateUserRequest request) {
        User user = new User(request.getName(), request.getEmail());
        User saved = userRepository.save(user);
        
        // Асинхронная отправка уведомления
        notificationService.sendWelcomeEmail(saved);
        
        return saved;
    }
}

2. Event-Driven Architecture

Принципы:

• События как основной способ коммуникации
• Асинхронная обработка
• Слабая связанность сервисов

// Event Publisher
@Component
public class UserEventPublisher {
    
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    
    public void publishUserCreated(User user) {
        UserCreatedEvent event = new UserCreatedEvent(user);
        eventPublisher.publishEvent(event);
    }
    
    public void publishUserUpdated(User user) {
        UserUpdatedEvent event = new UserUpdatedEvent(user);
        eventPublisher.publishEvent(event);
    }
}

// Event Listener
@Component
public class UserEventListener {
    
    @Autowired
    private NotificationService notificationService;
    
    @Autowired
    private AuditService auditService;
    
    @EventListener
    @Async
    public void handleUserCreated(UserCreatedEvent event) {
        User user = event.getUser();
        
        // Отправка приветственного email
        notificationService.sendWelcomeEmail(user);
        
        // Аудит
        auditService.logUserCreation(user);
    }
    
    @EventListener
    @Async
    public void handleUserUpdated(UserUpdatedEvent event) {
        User user = event.getUser();
        
        // Аудит изменений
        auditService.logUserUpdate(user);
    }
}

3. API Gateway Pattern

Функции API Gateway:

• Маршрутизация запросов
• Аутентификация и авторизация
• Rate limiting
• Мониторинг и логирование
• Кэширование ответов

// Spring Cloud Gateway конфигурация
@Configuration
public class GatewayConfig {
    
    @Bean
    public RouteLocator customRouteLocator(RouteLocatorBuilder builder) {
        return builder.routes()
            .route("user-service", r -> r.path("/api/users/**")
                .filters(f -> f
                    .stripPrefix(1)
                    .addRequestHeader("X-Service", "user-service")
                    .circuitBreaker(config -> config
                        .setName("user-service-cb")
                        .setFallbackUri("forward:/fallback/user-service")))
                .uri("lb://user-service"))
            
            .route("order-service", r -> r.path("/api/orders/**")
                .filters(f -> f
                    .stripPrefix(1)
                    .addRequestHeader("X-Service", "order-service")
                    .circuitBreaker(config -> config
                        .setName("order-service-cb")
                        .setFallbackUri("forward:/fallback/order-service")))
                .uri("lb://order-service"))
            
            .route("notification-service", r -> r.path("/api/notifications/**")
                .filters(f -> f
                    .stripPrefix(1)
                    .addRequestHeader("X-Service", "notification-service"))
                .uri("lb://notification-service"))
            
            .build();
    }
    
    @Bean
    public GlobalFilter customGlobalFilter() {
        return (exchange, chain) -> {
            ServerHttpRequest request = exchange.getRequest();
            
            // Добавление correlation ID
            String correlationId = UUID.randomUUID().toString();
            ServerHttpRequest mutatedRequest = request.mutate()
                .header("X-Correlation-ID", correlationId)
                .build();
            
            ServerHttpResponse response = exchange.getResponse();
            response.getHeaders().add("X-Correlation-ID", correlationId);
            
            return chain.filter(exchange.mutate().request(mutatedRequest).build());
        };
    }
}

Метрики и мониторинг масштабируемости

1. Ключевые метрики

@Component
public class ScalabilityMetrics {
    
    private final MeterRegistry meterRegistry;
    private final Counter requestCounter;
    private final Timer responseTimer;
    private final Gauge activeConnections;
    
    public ScalabilityMetrics(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        this.requestCounter = Counter.builder("requests.total")
            .description("Total number of requests")
            .register(meterRegistry);
        
        this.responseTimer = Timer.builder("response.time")
            .description("Response time")
            .register(meterRegistry);
        
        this.activeConnections = Gauge.builder("connections.active")
            .description("Active connections")
            .register(meterRegistry, this, ScalabilityMetrics::getActiveConnections);
    }
    
    public void recordRequest(String service, String method) {
        requestCounter.increment(
            Tags.of("service", service, "method", method)
        );
    }
    
    public void recordResponseTime(String service, Duration duration) {
        responseTimer.record(duration, TimeUnit.MILLISECONDS);
    }
    
    private double getActiveConnections() {
        // Логика подсчета активных соединений
        return 0.0;
    }
}

2. Автоматическое масштабирование

# Kubernetes HPA конфигурация
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: user-service-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: user-service
  minReplicas: 2
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80
  - type: Pods
    pods:
      metric:
        name: requests_per_second
      target:
        type: AverageValue
        averageValue: "100"

---

Паттерны масштабирования

1. Stateless Services (Безгосударственные сервисы)

Принцип: Сервисы не хранят состояние между запросами.

Технический пример - Stateless Service:

@RestController
@RequestMapping("/api/users")
public class UserController {
    @Autowired
    private UserService userService;
    
    @GetMapping("/{id}")
    public ResponseEntity<User> getUser(@PathVariable Long id) {
        // Stateless - нет сохранения состояния между запросами
        User user = userService.getUser(id);
        return ResponseEntity.ok(user);
    }
    
    @PostMapping
    public ResponseEntity<User> createUser(@RequestBody CreateUserRequest request) {
        // Stateless - каждый запрос независим
        User user = userService.createUser(request);
        return ResponseEntity.ok(user);
    }
}

@Service
@Transactional
public class UserService {
    @Autowired
    private UserRepository userRepository;
    
    public User getUser(Long id) {
        // Stateless - данные берутся из внешнего хранилища
        return userRepository.findById(id)
            .orElseThrow(() -> new UserNotFoundException("User not found"));
    }
    
    public User createUser(CreateUserRequest request) {
        // Stateless - состояние сохраняется в БД, не в памяти
        User user = new User();
        user.setName(request.getName());
        user.setEmail(request.getEmail());
        return userRepository.save(user);
    }
}

Технический пример - Database Sharding:

// Sharding Strategy
@Component
public class UserShardingStrategy {
    
    private static final int SHARD_COUNT = 4;
    
    public String getShardKey(Long userId) {
        // Простая стратегия sharding по остатку от деления
        int shardIndex = (int) (userId % SHARD_COUNT);
        return "user_shard_" + shardIndex;
    }
    
    public String getShardUrl(Long userId) {
        String shardKey = getShardKey(userId);
        return getShardUrlByKey(shardKey);
    }
    
    private String getShardUrlByKey(String shardKey) {
        Map<String, String> shardUrls = Map.of(
            "user_shard_0", "jdbc:postgresql://shard0:5432/users",
            "user_shard_1", "jdbc:postgresql://shard1:5432/users",
            "user_shard_2", "jdbc:postgresql://shard2:5432/users",
            "user_shard_3", "jdbc:postgresql://shard3:5432/users"
        );
        return shardUrls.get(shardKey);
    }
}

// Sharded Repository
@Repository
public class ShardedUserRepository {
    
    @Autowired
    private UserShardingStrategy shardingStrategy;
    
    @Autowired
    private Map<String, JdbcTemplate> shardTemplates;
    
    public User findById(Long userId) {
        String shardKey = shardingStrategy.getShardKey(userId);
        JdbcTemplate template = shardTemplates.get(shardKey);
        
        String sql = "SELECT id, name, email FROM users WHERE id = ?";
        return template.queryForObject(sql, new Object[]{userId}, (rs, rowNum) -> {
            User user = new User();
            user.setId(rs.getLong("id"));
            user.setName(rs.getString("name"));
            user.setEmail(rs.getString("email"));
            return user;
        });
    }
    
    public void save(User user) {
        String shardKey = shardingStrategy.getShardKey(user.getId());
        JdbcTemplate template = shardTemplates.get(shardKey);
        
        String sql = "INSERT INTO users (id, name, email) VALUES (?, ?, ?)";
        template.update(sql, user.getId(), user.getName(), user.getEmail());
    }
}

// Configuration for multiple datasources
@Configuration
public class ShardingDataSourceConfig {
    
    @Bean
    @Primary
    public DataSource shardingDataSource() {
        Map<String, DataSource> dataSources = new HashMap<>();
        
        // Shard 0
        dataSources.put("user_shard_0", createDataSource(
            "jdbc:postgresql://shard0:5432/users", "user", "password"));
        
        // Shard 1
        dataSources.put("user_shard_1", createDataSource(
            "jdbc:postgresql://shard1:5432/users", "user", "password"));
        
        // Shard 2
        dataSources.put("user_shard_2", createDataSource(
            "jdbc:postgresql://shard2:5432/users", "user", "password"));
        
        // Shard 3
        dataSources.put("user_shard_3", createDataSource(
            "jdbc:postgresql://shard3:5432/users", "user", "password"));
        
        return new ShardingDataSource(dataSources);
    }
    
    @Bean
    public Map<String, JdbcTemplate> shardTemplates() {
        Map<String, JdbcTemplate> templates = new HashMap<>();
        Map<String, DataSource> dataSources = getShardDataSources();
        
        for (Map.Entry<String, DataSource> entry : dataSources.entrySet()) {
            templates.put(entry.getKey(), new JdbcTemplate(entry.getValue()));
        }
        
        return templates;
    }
    
    private DataSource createDataSource(String url, String username, String password) {
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl(url);
        config.setUsername(username);
        config.setPassword(password);
        config.setMaximumPoolSize(10);
        config.setMinimumIdle(5);
        return new HikariDataSource(config);
    }
}

Технический пример - Read Replicas:

// Master-Slave Configuration
@Configuration
public class DatabaseConfig {
    
    @Bean
    @Primary
    @ConfigurationProperties("spring.datasource.master")
    public DataSource masterDataSource() {
        return DataSourceBuilder.create().build();
    }
    
    @Bean
    @ConfigurationProperties("spring.datasource.slave")
    public DataSource slaveDataSource() {
        return DataSourceBuilder.create().build();
    }
    
    @Bean
    public DataSource routingDataSource() {
        RoutingDataSource routingDataSource = new RoutingDataSource();
        
        Map<Object, Object> dataSourceMap = new HashMap<>();
        dataSourceMap.put("master", masterDataSource());
        dataSourceMap.put("slave", slaveDataSource());
        
        routingDataSource.setTargetDataSources(dataSourceMap);
        routingDataSource.setDefaultTargetDataSource(masterDataSource());
        
        return routingDataSource;
    }
}

// Routing DataSource
public class RoutingDataSource extends AbstractRoutingDataSource {
    
    @Override
    protected Object determineCurrentLookupKey() {
        return DatabaseContextHolder.getDatabaseType();
    }
}

// Database Context Holder
public class DatabaseContextHolder {
    
    private static final ThreadLocal<DatabaseType> contextHolder = new ThreadLocal<>();
    
    public static void setDatabaseType(DatabaseType databaseType) {
        contextHolder.set(databaseType);
    }
    
    public static DatabaseType getDatabaseType() {
        return contextHolder.get();
    }
    
    public static void clearDatabaseType() {
        contextHolder.remove();
    }
    
    public enum DatabaseType {
        MASTER, SLAVE
    }
}

// Service with Read/Write separation
@Service
public class UserService {
    
    @Autowired
    private UserRepository userRepository;
    
    @Transactional(readOnly = false)
    public User createUser(CreateUserRequest request) {
        // Write operations go to master
        DatabaseContextHolder.setDatabaseType(DatabaseType.MASTER);
        try {
            User user = new User();
            user.setName(request.getName());
            user.setEmail(request.getEmail());
            return userRepository.save(user);
        } finally {
            DatabaseContextHolder.clearDatabaseType();
        }
    }
    
    @Transactional(readOnly = true)
    public User getUser(Long id) {
        // Read operations can go to slave
        DatabaseContextHolder.setDatabaseType(DatabaseType.SLAVE);
        try {
            return userRepository.findById(id)
                .orElseThrow(() -> new UserNotFoundException("User not found"));
        } finally {
            DatabaseContextHolder.clearDatabaseType();
        }
    }
    
    @Transactional(readOnly = true)
    public List<User> getAllUsers() {
        // Read operations can go to slave
        DatabaseContextHolder.setDatabaseType(DatabaseType.SLAVE);
        try {
            return userRepository.findAll();
        } finally {
            DatabaseContextHolder.clearDatabaseType();
        }
    }
}

// AOP for automatic routing
@Aspect
@Component
public class DatabaseRoutingAspect {
    
    @Before("@annotation(readOnly)")
    public void setReadOnlyDataSource(ReadOnly readOnly) {
        DatabaseContextHolder.setDatabaseType(DatabaseType.SLAVE);
    }
    
    @Before("@annotation(Transactional)")
    public void setMasterDataSource(Transactional transactional) {
        if (!transactional.readOnly()) {
            DatabaseContextHolder.setDatabaseType(DatabaseType.MASTER);
        }
    }
    
    @After("@annotation(readOnly) || @annotation(Transactional)")
    public void clearDataSource() {
        DatabaseContextHolder.clearDatabaseType();
    }
}

// Annotation for read-only operations
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface ReadOnly {
}

public User getUser(@PathVariable Long id) {
    // нет локального состояния
    return userService.findById(id);
}

}

**Преимущества:**
- Легкое масштабирование
- Отказоустойчивость
- Простота развертывания

### 2. Database Sharding (Шардирование БД)

**Определение:** Разделение данных по нескольким базам данных.

#### Стратегии шардирования:

##### 1. Horizontal Sharding
```sql
-- Shard 1: users 1-1000
CREATE TABLE users_1 (
    id INT PRIMARY KEY,
    name VARCHAR(100)
);

-- Shard 2: users 1001-2000
CREATE TABLE users_2 (
    id INT PRIMARY KEY,
    name VARCHAR(100)
);

2. Vertical Sharding

-- Shard 1: User data
CREATE TABLE users (
    id INT PRIMARY KEY,
    name VARCHAR(100),
    email VARCHAR(100)
);

-- Shard 2: User preferences
CREATE TABLE user_preferences (
    user_id INT,
    theme VARCHAR(50),
    language VARCHAR(10)
);

Алгоритмы шардирования:

1. Range-based

public class RangeSharding {
    public String getShard(int userId) {
        if (userId >= 1 && userId <= 1000) {
            return "shard_1";
        } else if (userId >= 1001 && userId <= 2000) {
            return "shard_2";
        }
        return "shard_3";
    }
}

2. Hash-based

public class HashSharding {
    public String getShard(int userId) {
        int hash = userId.hashCode();
        int shardNumber = Math.abs(hash) % 3;
        return "shard_" + (shardNumber + 1);
    }
}

3. Read Replicas (Реплики для чтения)

Принцип: Разделение операций чтения и записи.

# Master-Slave конфигурация
Master DB (Write) → Slave DB 1 (Read)
                 → Slave DB 2 (Read)
                 → Slave DB 3 (Read)

@Service
public class UserService {
    @Autowired
    @Qualifier("masterDataSource")
    private DataSource masterDataSource;
    
    @Autowired
    @Qualifier("slaveDataSource")
    private DataSource slaveDataSource;
    
    public void createUser(User user) {
        // запись в master
        JdbcTemplate masterTemplate = new JdbcTemplate(masterDataSource);
        masterTemplate.update("INSERT INTO users ...", user.getName());
    }
    
    public User findById(Long id) {
        // чтение из slave
        JdbcTemplate slaveTemplate = new JdbcTemplate(slaveDataSource);
        return slaveTemplate.queryForObject("SELECT * FROM users WHERE id = ?", 
                                           new Object[]{id}, User.class);
    }
}

4. Caching Patterns (Паттерны кэширования)

Cache-Aside (Кэш в стороне)

@Service
public class UserService {
    @Autowired
    private RedisTemplate<String, User> redisTemplate;
    
    public User findById(Long id) {
        String key = "user:" + id;
        
        // проверяем кэш
        User user = redisTemplate.opsForValue().get(key);
        if (user != null) {
            return user;
        }
        
        // загружаем из БД
        user = userRepository.findById(id);
        if (user != null) {
            // сохраняем в кэш
            redisTemplate.opsForValue().set(key, user, Duration.ofMinutes(30));
        }
        
        return user;
    }
}

Write-Through (Сквозная запись)

@Service
public class UserService {
    public void updateUser(User user) {
        // обновляем БД
        userRepository.save(user);
        
        // обновляем кэш
        String key = "user:" + user.getId();
        redisTemplate.opsForValue().set(key, user, Duration.ofMinutes(30));
    }
}

Write-Behind (Запись с задержкой)

@Service
public class UserService {
    @Async
    public void updateUser(User user) {
        // обновляем кэш сразу
        String key = "user:" + user.getId();
        redisTemplate.opsForValue().set(key, user, Duration.ofMinutes(30));
        
        // обновляем БД асинхронно
        CompletableFuture.runAsync(() -> {
            userRepository.save(user);
        });
    }
}

Продвинутые паттерны масштабирования

1. Event Sourcing (Событийное программирование)

Принцип: Хранение изменений состояния как последовательности событий.

Преимущества:

• Полная история изменений
• Возможность воспроизведения состояния
• Аудит и отладка
• Временные запросы

Реализация:

// Event Store
@Component
public class EventStore {
    
    @Autowired
    private EventRepository eventRepository;
    
    public void saveEvents(String aggregateId, List<DomainEvent> events, int expectedVersion) {
        List<StoredEvent> storedEvents = events.stream()
            .map(event -> new StoredEvent(aggregateId, event, expectedVersion))
            .collect(Collectors.toList());
        
        eventRepository.saveAll(storedEvents);
    }
    
    public List<DomainEvent> getEvents(String aggregateId) {
        return eventRepository.findByAggregateIdOrderByVersion(aggregateId)
            .stream()
            .map(StoredEvent::getEvent)
            .collect(Collectors.toList());
    }
    
    public List<DomainEvent> getEventsFromVersion(String aggregateId, int fromVersion) {
        return eventRepository.findByAggregateIdAndVersionGreaterThanOrderByVersion(aggregateId, fromVersion)
            .stream()
            .map(StoredEvent::getEvent)
            .collect(Collectors.toList());
    }
}

// Aggregate Root
public abstract class AggregateRoot {
    private String id;
    private int version;
    private List<DomainEvent> uncommittedEvents = new ArrayList<>();
    
    protected void apply(DomainEvent event) {
        uncommittedEvents.add(event);
        handle(event);
    }
    
    protected abstract void handle(DomainEvent event);
    
    public List<DomainEvent> getUncommittedEvents() {
        return new ArrayList<>(uncommittedEvents);
    }
    
    public void markEventsAsCommitted() {
        uncommittedEvents.clear();
    }
}

// User Aggregate
public class User extends AggregateRoot {
    private String name;
    private String email;
    private UserStatus status;
    
    public static User create(String name, String email) {
        User user = new User();
        user.apply(new UserCreatedEvent(name, email));
        return user;
    }
    
    public void changeEmail(String newEmail) {
        if (status == UserStatus.ACTIVE) {
            apply(new UserEmailChangedEvent(newEmail));
        }
    }
    
    public void deactivate() {
        if (status == UserStatus.ACTIVE) {
            apply(new UserDeactivatedEvent());
        }
    }
    
    @Override
    protected void handle(DomainEvent event) {
        if (event instanceof UserCreatedEvent) {
            UserCreatedEvent e = (UserCreatedEvent) event;
            this.name = e.getName();
            this.email = e.getEmail();
            this.status = UserStatus.ACTIVE;
        } else if (event instanceof UserEmailChangedEvent) {
            UserEmailChangedEvent e = (UserEmailChangedEvent) event;
            this.email = e.getNewEmail();
        } else if (event instanceof UserDeactivatedEvent) {
            this.status = UserStatus.INACTIVE;
        }
    }
}

// Event Sourced Repository
@Repository
public class EventSourcedUserRepository {
    
    @Autowired
    private EventStore eventStore;
    
    public User findById(String id) {
        List<DomainEvent> events = eventStore.getEvents(id);
        if (events.isEmpty()) {
            return null;
        }
        
        User user = new User();
        for (DomainEvent event : events) {
            user.handle(event);
        }
        
        return user;
    }
    
    public void save(User user) {
        List<DomainEvent> uncommittedEvents = user.getUncommittedEvents();
        if (!uncommittedEvents.isEmpty()) {
            eventStore.saveEvents(user.getId(), uncommittedEvents, user.getVersion());
            user.markEventsAsCommitted();
        }
    }
}

Rust примеры - Event Sourcing:

use serde::{Deserialize, Serialize};
use std::collections::HashMap;

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum UserEvent {
    Created { name: String, email: String },
    EmailChanged { new_email: String },
    Deactivated,
}

#[derive(Debug, Clone)]
pub struct User {
    id: String,
    name: String,
    email: String,
    status: UserStatus,
    version: u32,
    uncommitted_events: Vec<UserEvent>,
}

#[derive(Debug, Clone)]
pub enum UserStatus {
    Active,
    Inactive,
}

impl User {
    pub fn create(id: String, name: String, email: String) -> Self {
        let mut user = Self {
            id,
            name: String::new(),
            email: String::new(),
            status: UserStatus::Active,
            version: 0,
            uncommitted_events: Vec::new(),
        };
        
        user.apply(UserEvent::Created { name, email });
        user
    }
    
    pub fn change_email(&mut self, new_email: String) {
        if matches!(self.status, UserStatus::Active) {
            self.apply(UserEvent::EmailChanged { new_email });
        }
    }
    
    pub fn deactivate(&mut self) {
        if matches!(self.status, UserStatus::Active) {
            self.apply(UserEvent::Deactivated);
        }
    }
    
    fn apply(&mut self, event: UserEvent) {
        self.uncommitted_events.push(event.clone());
        self.handle(event);
    }
    
    fn handle(&mut self, event: UserEvent) {
        match event {
            UserEvent::Created { name, email } => {
                self.name = name;
                self.email = email;
                self.status = UserStatus::Active;
            }
            UserEvent::EmailChanged { new_email } => {
                self.email = new_email;
            }
            UserEvent::Deactivated => {
                self.status = UserStatus::Inactive;
            }
        }
    }
    
    pub fn get_uncommitted_events(&self) -> Vec<UserEvent> {
        self.uncommitted_events.clone()
    }
    
    pub fn mark_events_as_committed(&mut self) {
        self.uncommitted_events.clear();
        self.version += 1;
    }
}

pub struct EventStore {
    events: HashMap<String, Vec<UserEvent>>,
}

impl EventStore {
    pub fn new() -> Self {
        Self {
            events: HashMap::new(),
        }
    }
    
    pub fn save_events(&mut self, aggregate_id: &str, events: Vec<UserEvent>) {
        let stored_events = self.events.entry(aggregate_id.to_string()).or_insert_with(Vec::new);
        stored_events.extend(events);
    }
    
    pub fn get_events(&self, aggregate_id: &str) -> Vec<UserEvent> {
        self.events.get(aggregate_id).cloned().unwrap_or_default()
    }
}

2. CQRS (Command Query Responsibility Segregation)

Принцип: Разделение операций чтения и записи на разные модели.

Преимущества:

• Оптимизация для чтения и записи
• Независимое масштабирование
• Различные модели данных
• Упрощение сложных доменов

Реализация:

// Command Side
@RestController
@RequestMapping("/api/users/commands")
public class UserCommandController {
    
    @Autowired
    private UserCommandService userCommandService;
    
    @PostMapping
    public ResponseEntity<String> createUser(@RequestBody CreateUserCommand command) {
        String userId = userCommandService.createUser(command);
        return ResponseEntity.ok(userId);
    }
    
    @PutMapping("/{id}/email")
    public ResponseEntity<Void> changeEmail(@PathVariable String id, @RequestBody ChangeEmailCommand command) {
        userCommandService.changeEmail(id, command);
        return ResponseEntity.ok().build();
    }
}

@Service
public class UserCommandService {
    
    @Autowired
    private UserCommandRepository userCommandRepository;
    
    @Autowired
    private UserEventPublisher eventPublisher;
    
    @Transactional
    public String createUser(CreateUserCommand command) {
        User user = new User(command.getName(), command.getEmail());
        user = userCommandRepository.save(user);
        
        // Публикация события для обновления read model
        eventPublisher.publishUserCreated(user);
        
        return user.getId();
    }
    
    @Transactional
    public void changeEmail(String userId, ChangeEmailCommand command) {
        User user = userCommandRepository.findById(userId)
            .orElseThrow(() -> new UserNotFoundException("User not found"));
        
        user.changeEmail(command.getNewEmail());
        userCommandRepository.save(user);
        
        // Публикация события для обновления read model
        eventPublisher.publishUserEmailChanged(user);
    }
}

// Query Side
@RestController
@RequestMapping("/api/users/queries")
public class UserQueryController {
    
    @Autowired
    private UserQueryService userQueryService;
    
    @GetMapping("/{id}")
    public ResponseEntity<UserView> getUser(@PathVariable String id) {
        UserView user = userQueryService.getUser(id);
        return ResponseEntity.ok(user);
    }
    
    @GetMapping
    public ResponseEntity<List<UserView>> getUsers(@RequestParam(required = false) String status) {
        List<UserView> users = userQueryService.getUsers(status);
        return ResponseEntity.ok(users);
    }
}

@Service
public class UserQueryService {
    
    @Autowired
    private UserViewRepository userViewRepository;
    
    public UserView getUser(String id) {
        return userViewRepository.findById(id)
            .orElseThrow(() -> new UserNotFoundException("User not found"));
    }
    
    public List<UserView> getUsers(String status) {
        if (status != null) {
            return userViewRepository.findByStatus(status);
        }
        return userViewRepository.findAll();
    }
}

// Event Handler для синхронизации read model
@Component
public class UserEventHandler {
    
    @Autowired
    private UserViewRepository userViewRepository;
    
    @EventListener
    @Async
    public void handleUserCreated(UserCreatedEvent event) {
        UserView userView = new UserView();
        userView.setId(event.getUserId());
        userView.setName(event.getName());
        userView.setEmail(event.getEmail());
        userView.setStatus("ACTIVE");
        userView.setCreatedAt(event.getTimestamp());
        
        userViewRepository.save(userView);
    }
    
    @EventListener
    @Async
    public void handleUserEmailChanged(UserEmailChangedEvent event) {
        UserView userView = userViewRepository.findById(event.getUserId())
            .orElseThrow(() -> new UserNotFoundException("User not found"));
        
        userView.setEmail(event.getNewEmail());
        userView.setUpdatedAt(event.getTimestamp());
        
        userViewRepository.save(userView);
    }
}

Python примеры - CQRS:

from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime
import asyncio

@dataclass
class CreateUserCommand:
    name: str
    email: str

@dataclass
class ChangeEmailCommand:
    new_email: str

@dataclass
class UserView:
    id: str
    name: str
    email: str
    status: str
    created_at: datetime
    updated_at: Optional[datetime] = None

class UserCommandService:
    def __init__(self, user_command_repository, event_publisher):
        self.user_command_repository = user_command_repository
        self.event_publisher = event_publisher
    
    async def create_user(self, command: CreateUserCommand) -> str:
        user = User(command.name, command.email)
        user = await self.user_command_repository.save(user)
        
        # Публикация события для обновления read model
        await self.event_publisher.publish_user_created(user)
        
        return user.id
    
    async def change_email(self, user_id: str, command: ChangeEmailCommand):
        user = await self.user_command_repository.find_by_id(user_id)
        if not user:
            raise UserNotFoundException("User not found")
        
        user.change_email(command.new_email)
        await self.user_command_repository.save(user)
        
        # Публикация события для обновления read model
        await self.event_publisher.publish_user_email_changed(user)

class UserQueryService:
    def __init__(self, user_view_repository):
        self.user_view_repository = user_view_repository
    
    async def get_user(self, user_id: str) -> UserView:
        user_view = await self.user_view_repository.find_by_id(user_id)
        if not user_view:
            raise UserNotFoundException("User not found")
        return user_view
    
    async def get_users(self, status: Optional[str] = None) -> List[UserView]:
        if status:
            return await self.user_view_repository.find_by_status(status)
        return await self.user_view_repository.find_all()

class UserEventHandler:
    def __init__(self, user_view_repository):
        self.user_view_repository = user_view_repository
    
    async def handle_user_created(self, event: UserCreatedEvent):
        user_view = UserView(
            id=event.user_id,
            name=event.name,
            email=event.email,
            status="ACTIVE",
            created_at=event.timestamp
        )
        
        await self.user_view_repository.save(user_view)
    
    async def handle_user_email_changed(self, event: UserEmailChangedEvent):
        user_view = await self.user_view_repository.find_by_id(event.user_id)
        if not user_view:
            raise UserNotFoundException("User not found")
        
        user_view.email = event.new_email
        user_view.updated_at = event.timestamp
        
        await self.user_view_repository.save(user_view)

3. Saga Pattern (Паттерн Сага)

Принцип: Управление распределенными транзакциями через последовательность локальных транзакций.

Типы Saga:

• Choreography Saga - каждый сервис знает о следующих шагах
• Orchestration Saga - центральный оркестратор управляет процессом

Choreography Saga:

// Order Saga Events
public class OrderCreatedEvent {
    private String orderId;
    private String customerId;
    private BigDecimal amount;
    private List<OrderItem> items;
    // getters/setters
}

public class PaymentProcessedEvent {
    private String orderId;
    private String paymentId;
    private boolean success;
    // getters/setters
}

public class InventoryReservedEvent {
    private String orderId;
    private List<OrderItem> reservedItems;
    private boolean success;
    // getters/setters
}

// Order Service
@Service
public class OrderService {
    
    @Autowired
    private OrderRepository orderRepository;
    
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    
    @Transactional
    public String createOrder(CreateOrderRequest request) {
        Order order = new Order(request.getCustomerId(), request.getItems());
        order = orderRepository.save(order);
        
        // Публикация события для начала саги
        eventPublisher.publishEvent(new OrderCreatedEvent(order.getId(), order.getCustomerId(), order.getAmount(), order.getItems()));
        
        return order.getId();
    }
    
    @EventListener
    @Async
    public void handlePaymentProcessed(PaymentProcessedEvent event) {
        if (event.isSuccess()) {
            // Платеж успешен - резервируем товары
            eventPublisher.publishEvent(new ReserveInventoryCommand(event.getOrderId(), event.getOrderItems()));
        } else {
            // Платеж не прошел - отменяем заказ
            cancelOrder(event.getOrderId());
        }
    }
    
    @EventListener
    @Async
    public void handleInventoryReserved(InventoryReservedEvent event) {
        if (event.isSuccess()) {
            // Товары зарезервированы - подтверждаем заказ
            confirmOrder(event.getOrderId());
        } else {
            // Не удалось зарезервировать - отменяем платеж
            eventPublisher.publishEvent(new CancelPaymentCommand(event.getOrderId()));
        }
    }
}

// Payment Service
@Service
public class PaymentService {
    
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    
    @EventListener
    @Async
    public void handleOrderCreated(OrderCreatedEvent event) {
        try {
            // Обработка платежа
            String paymentId = processPayment(event.getCustomerId(), event.getAmount());
            
            // Публикация успешного результата
            eventPublisher.publishEvent(new PaymentProcessedEvent(event.getOrderId(), paymentId, true));
            
        } catch (PaymentException e) {
            // Публикация неуспешного результата
            eventPublisher.publishEvent(new PaymentProcessedEvent(event.getOrderId(), null, false));
        }
    }
    
    @EventListener
    @Async
    public void handleCancelPayment(CancelPaymentCommand command) {
        // Отмена платежа
        cancelPayment(command.getOrderId());
    }
}

// Inventory Service
@Service
public class InventoryService {
    
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    
    @EventListener
    @Async
    public void handleReserveInventory(ReserveInventoryCommand command) {
        try {
            // Резервирование товаров
            List<OrderItem> reservedItems = reserveItems(command.getOrderItems());
            
            // Публикация успешного результата
            eventPublisher.publishEvent(new InventoryReservedEvent(command.getOrderId(), reservedItems, true));
            
        } catch (InsufficientInventoryException e) {
            // Публикация неуспешного результата
            eventPublisher.publishEvent(new InventoryReservedEvent(command.getOrderId(), null, false));
        }
    }
}

Orchestration Saga:

// Saga Orchestrator
@Component
public class OrderSagaOrchestrator {
    
    @Autowired
    private PaymentService paymentService;
    
    @Autowired
    private InventoryService inventoryService;
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private SagaRepository sagaRepository;
    
    public void startOrderSaga(CreateOrderRequest request) {
        OrderSaga saga = new OrderSaga(request);
        sagaRepository.save(saga);
        
        // Шаг 1: Создание заказа
        String orderId = orderService.createOrder(request);
        saga.setOrderId(orderId);
        saga.setCurrentStep(SagaStep.CREATE_ORDER);
        sagaRepository.save(saga);
        
        // Шаг 2: Обработка платежа
        processPayment(saga);
    }
    
    private void processPayment(OrderSaga saga) {
        try {
            String paymentId = paymentService.processPayment(saga.getCustomerId(), saga.getAmount());
            saga.setPaymentId(paymentId);
            saga.setCurrentStep(SagaStep.PROCESS_PAYMENT);
            sagaRepository.save(saga);
            
            // Шаг 3: Резервирование товаров
            reserveInventory(saga);
            
        } catch (PaymentException e) {
            // Компенсация: отмена заказа
            compensateOrderCreation(saga);
        }
    }
    
    private void reserveInventory(OrderSaga saga) {
        try {
            List<OrderItem> reservedItems = inventoryService.reserveItems(saga.getOrderItems());
            saga.setReservedItems(reservedItems);
            saga.setCurrentStep(SagaStep.RESERVE_INVENTORY);
            sagaRepository.save(saga);
            
            // Шаг 4: Подтверждение заказа
            confirmOrder(saga);
            
        } catch (InsufficientInventoryException e) {
            // Компенсация: отмена платежа
            compensatePayment(saga);
        }
    }
    
    private void confirmOrder(OrderSaga saga) {
        orderService.confirmOrder(saga.getOrderId());
        saga.setCurrentStep(SagaStep.CONFIRM_ORDER);
        saga.setStatus(SagaStatus.COMPLETED);
        sagaRepository.save(saga);
    }
    
    private void compensatePayment(OrderSaga saga) {
        paymentService.cancelPayment(saga.getPaymentId());
        compensateOrderCreation(saga);
    }
    
    private void compensateOrderCreation(OrderSaga saga) {
        orderService.cancelOrder(saga.getOrderId());
        saga.setStatus(SagaStatus.COMPENSATED);
        sagaRepository.save(saga);
    }
}

4. Bulkhead Pattern (Переборка)

Принцип: Изоляция критически важных ресурсов от отказа других ресурсов.

// Thread Pool Isolation
@Configuration
public class ThreadPoolConfig {
    
    @Bean("userServiceExecutor")
    public ThreadPoolTaskExecutor userServiceExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("user-service-");
        executor.initialize();
        return executor;
    }
    
    @Bean("paymentServiceExecutor")
    public ThreadPoolTaskExecutor paymentServiceExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(50);
        executor.setThreadNamePrefix("payment-service-");
        executor.initialize();
        return executor;
    }
    
    @Bean("notificationServiceExecutor")
    public ThreadPoolTaskExecutor notificationServiceExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(3);
        executor.setMaxPoolSize(5);
        executor.setQueueCapacity(25);
        executor.setThreadNamePrefix("notification-service-");
        executor.initialize();
        return executor;
    }
}

// Service с изолированными thread pools
@Service
public class BulkheadUserService {
    
    @Autowired
    @Qualifier("userServiceExecutor")
    private ThreadPoolTaskExecutor userServiceExecutor;
    
    @Autowired
    @Qualifier("paymentServiceExecutor")
    private ThreadPoolTaskExecutor paymentServiceExecutor;
    
    @Autowired
    @Qualifier("notificationServiceExecutor")
    private ThreadPoolTaskExecutor notificationServiceExecutor;
    
    public CompletableFuture<User> createUser(CreateUserRequest request) {
        return CompletableFuture.supplyAsync(() -> {
            // Обработка пользователя в изолированном пуле
            return processUserCreation(request);
        }, userServiceExecutor);
    }
    
    public CompletableFuture<PaymentResult> processPayment(PaymentRequest request) {
        return CompletableFuture.supplyAsync(() -> {
            // Обработка платежа в изолированном пуле
            return processPaymentInternal(request);
        }, paymentServiceExecutor);
    }
    
    public CompletableFuture<Void> sendNotification(NotificationRequest request) {
        return CompletableFuture.runAsync(() -> {
            // Отправка уведомления в изолированном пуле
            sendNotificationInternal(request);
        }, notificationServiceExecutor);
    }
}

5. Database Per Service Pattern

Принцип: Каждый микросервис имеет собственную базу данных.

// User Service Database
@Configuration
@EnableJpaRepositories(basePackages = "com.example.userservice.repository")
public class UserServiceDatabaseConfig {
    
    @Bean
    @Primary
    public DataSource userDataSource() {
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl("jdbc:postgresql://user-db:5432/user_service");
        config.setUsername("user_service_user");
        config.setPassword("user_service_password");
        config.setMaximumPoolSize(20);
        return new HikariDataSource(config);
    }
    
    @Bean
    @Primary
    public LocalContainerEntityManagerFactoryBean userEntityManagerFactory() {
        LocalContainerEntityManagerFactoryBean em = new LocalContainerEntityManagerFactoryBean();
        em.setDataSource(userDataSource());
        em.setPackagesToScan("com.example.userservice.entity");
        
        HibernateJpaVendorAdapter vendorAdapter = new HibernateJpaVendorAdapter();
        em.setJpaVendorAdapter(vendorAdapter);
        
        Properties properties = new Properties();
        properties.setProperty("hibernate.hbm2ddl.auto", "update");
        properties.setProperty("hibernate.dialect", "org.hibernate.dialect.PostgreSQLDialect");
        em.setJpaProperties(properties);
        
        return em;
    }
}

// Order Service Database
@Configuration
@EnableJpaRepositories(basePackages = "com.example.orderservice.repository")
public class OrderServiceDatabaseConfig {
    
    @Bean
    public DataSource orderDataSource() {
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl("jdbc:postgresql://order-db:5432/order_service");
        config.setUsername("order_service_user");
        config.setPassword("order_service_password");
        config.setMaximumPoolSize(15);
        return new HikariDataSource(config);
    }
    
    @Bean
    public LocalContainerEntityManagerFactoryBean orderEntityManagerFactory() {
        LocalContainerEntityManagerFactoryBean em = new LocalContainerEntityManagerFactoryBean();
        em.setDataSource(orderDataSource());
        em.setPackagesToScan("com.example.orderservice.entity");
        
        HibernateJpaVendorAdapter vendorAdapter = new HibernateJpaVendorAdapter();
        em.setJpaVendorAdapter(vendorAdapter);
        
        Properties properties = new Properties();
        properties.setProperty("hibernate.hbm2ddl.auto", "update");
        properties.setProperty("hibernate.dialect", "org.hibernate.dialect.PostgreSQLDialect");
        em.setJpaProperties(properties);
        
        return em;
    }
}

---

Отказоустойчивость

Что такое отказоустойчивость

Отказоустойчивость — способность системы продолжать работу при отказе отдельных компонентов.

Типы отказов

1. Hardware Failures

• Отказ диска
• Отказ памяти
• Отказ сети
• Отказ процессора

2. Software Failures

• Ошибки в коде
• Утечки памяти
• Deadlocks
• Infinite loops

3. Network Failures

• Потеря пакетов
• Высокая задержка
• Разрыв соединения
• DDoS атаки

Паттерны отказоустойчивости

1. Circuit Breaker (Автоматический выключатель)

Технический пример - Circuit Breaker с Spring Cloud:

// Circuit Breaker Configuration
@Configuration
@EnableCircuitBreaker
public class CircuitBreakerConfig {
    
    @Bean
    public CircuitBreakerFactory circuitBreakerFactory() {
        return new DefaultCircuitBreakerFactory();
    }
}

// Service with Circuit Breaker
@Service
public class PaymentService {
    
    @Autowired
    private RestTemplate restTemplate;
    
    @CircuitBreaker(name = "payment-service", fallbackMethod = "processPaymentFallback")
    @Retry(name = "payment-service")
    @TimeLimiter(name = "payment-service")
    public CompletableFuture<PaymentResult> processPayment(PaymentRequest request) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                ResponseEntity<PaymentResult> response = restTemplate.postForEntity(
                    "http://payment-service/api/payments/process",
                    request,
                    PaymentResult.class
                );
                
                if (response.getStatusCode() == HttpStatus.OK) {
                    return response.getBody();
                } else {
                    throw new PaymentServiceException("Payment service returned error");
                }
                
            } catch (Exception e) {
                throw new PaymentServiceException("Payment processing failed", e);
            }
        });
    }
    
    // Fallback method
    public CompletableFuture<PaymentResult> processPaymentFallback(PaymentRequest request, Exception ex) {
        log.warn("Payment service is down, using fallback for request: {}", request, ex);
        
        // Fallback logic - save to queue for later processing
        return CompletableFuture.completedFuture(
            new PaymentResult(
                null,
                PaymentStatus.PENDING,
                "Payment queued for later processing",
                "Service temporarily unavailable"
            )
        );
    }
}

// Advanced Circuit Breaker Implementation
@Component
public class AdvancedCircuitBreaker {
    
    private final Map<String, CircuitBreakerState> circuitStates = new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    
    public <T> T execute(String serviceName, Supplier<T> operation, Supplier<T> fallback) {
        CircuitBreakerState state = circuitStates.computeIfAbsent(serviceName, k -> new CircuitBreakerState());
        
        if (state.isOpen()) {
            if (state.shouldAttemptReset()) {
                state.setState(CircuitState.HALF_OPEN);
            } else {
                return fallback.get();
            }
        }
        
        try {
            T result = operation.get();
            state.onSuccess();
            return result;
        } catch (Exception e) {
            state.onFailure();
            return fallback.get();
        }
    }
    
    private static class CircuitBreakerState {
        private volatile CircuitState state = CircuitState.CLOSED;
        private final AtomicInteger failureCount = new AtomicInteger(0);
        private final AtomicLong lastFailureTime = new AtomicLong(0);
        private final AtomicInteger successCount = new AtomicInteger(0);
        
        private static final int FAILURE_THRESHOLD = 5;
        private static final long TIMEOUT_DURATION = 60000; // 1 minute
        private static final int SUCCESS_THRESHOLD = 3;
        
        public boolean isOpen() {
            return state == CircuitState.OPEN;
        }
        
        public boolean shouldAttemptReset() {
            return System.currentTimeMillis() - lastFailureTime.get() > TIMEOUT_DURATION;
        }
        
        public void onSuccess() {
            if (state == CircuitState.HALF_OPEN) {
                int successes = successCount.incrementAndGet();
                if (successes >= SUCCESS_THRESHOLD) {
                    state = CircuitState.CLOSED;
                    failureCount.set(0);
                    successCount.set(0);
                }
            } else {
                failureCount.set(0);
            }
        }
        
        public void onFailure() {
            int failures = failureCount.incrementAndGet();
            lastFailureTime.set(System.currentTimeMillis());
            
            if (failures >= FAILURE_THRESHOLD) {
                state = CircuitState.OPEN;
                successCount.set(0);
            }
        }
        
        public void setState(CircuitState newState) {
            this.state = newState;
        }
    }
    
    private enum CircuitState {
        CLOSED, OPEN, HALF_OPEN
    }
}

Node.js примеры - Circuit Breaker:

Circuit Breaker с библиотекой opossum:

import CircuitBreaker from 'opossum';
import axios from 'axios';

// Circuit Breaker Configuration
const circuitBreakerOptions = {
  timeout: 3000, // 3 seconds
  errorThresholdPercentage: 50, // 50% error rate
  resetTimeout: 30000, // 30 seconds
  rollingCountTimeout: 10000, // 10 seconds
  rollingCountBuckets: 10,
  name: 'payment-service',
  group: 'payment'
};

// Service with Circuit Breaker
class PaymentService {
  private circuitBreaker: CircuitBreaker;

  constructor() {
    this.circuitBreaker = new CircuitBreaker(this.callPaymentService.bind(this), circuitBreakerOptions);
    
    // Event listeners
    this.circuitBreaker.on('open', () => {
      console.log('Circuit breaker opened for payment service');
    });

    this.circuitBreaker.on('halfOpen', () => {
      console.log('Circuit breaker half-open for payment service');
    });

    this.circuitBreaker.on('close', () => {
      console.log('Circuit breaker closed for payment service');
    });

    this.circuitBreaker.on('fallback', (error: Error) => {
      console.log('Circuit breaker fallback triggered:', error.message);
    });
  }

  async processPayment(request: PaymentRequest): Promise<PaymentResult> {
    try {
      return await this.circuitBreaker.fire(request);
    } catch (error) {
      console.error('Payment processing failed:', error);
      throw error;
    }
  }

  private async callPaymentService(request: PaymentRequest): Promise<PaymentResult> {
    const response = await axios.post('http://payment-service/api/payments/process', request, {
      timeout: 5000
    });

    if (response.status === 200) {
      return response.data;
    } else {
      throw new Error(`Payment service returned error: ${response.status}`);
    }
  }

  // Fallback method
  private async processPaymentFallback(request: PaymentRequest, error: Error): Promise<PaymentResult> {
    console.warn('Payment service is down, using fallback for request:', request, error);
    
    // Fallback logic - save to queue for later processing
    return {
      id: null,
      status: 'PENDING',
      message: 'Payment queued for later processing',
      error: 'Service temporarily unavailable'
    };
  }
}

// Advanced Circuit Breaker Implementation
class AdvancedCircuitBreaker {
  private circuitStates: Map<string, CircuitBreakerState> = new Map();
  private scheduler: NodeJS.Timeout;

  constructor() {
    this.scheduler = setInterval(() => {
      this.checkCircuitStates();
    }, 1000);
  }

  async execute<T>(
    serviceName: string, 
    operation: () => Promise<T>, 
    fallback: () => Promise<T>
  ): Promise<T> {
    const state = this.getOrCreateState(serviceName);
    
    if (state.isOpen()) {
      if (state.shouldAttemptReset()) {
        state.setState(CircuitState.HALF_OPEN);
      } else {
        return await fallback();
      }
    }
    
    try {
      const result = await operation();
      state.onSuccess();
      return result;
    } catch (error) {
      state.onFailure();
      return await fallback();
    }
  }

  private getOrCreateState(serviceName: string): CircuitBreakerState {
    if (!this.circuitStates.has(serviceName)) {
      this.circuitStates.set(serviceName, new CircuitBreakerState());
    }
    return this.circuitStates.get(serviceName)!;
  }

  private checkCircuitStates(): void {
    for (const [serviceName, state] of this.circuitStates) {
      if (state.isOpen() && state.shouldAttemptReset()) {
        state.setState(CircuitState.HALF_OPEN);
        console.log(`Circuit breaker for ${serviceName} moved to HALF_OPEN`);
      }
    }
  }

  destroy(): void {
    if (this.scheduler) {
      clearInterval(this.scheduler);
    }
  }
}

class CircuitBreakerState {
  private state: CircuitState = CircuitState.CLOSED;
  private failureCount: number = 0;
  private lastFailureTime: number = 0;
  private successCount: number = 0;
  
  private static readonly FAILURE_THRESHOLD = 5;
  private static readonly TIMEOUT_DURATION = 60000; // 1 minute
  private static readonly SUCCESS_THRESHOLD = 3;
  
  isOpen(): boolean {
    return this.state === CircuitState.OPEN;
  }
  
  shouldAttemptReset(): boolean {
    return Date.now() - this.lastFailureTime > CircuitBreakerState.TIMEOUT_DURATION;
  }
  
  onSuccess(): void {
    if (this.state === CircuitState.HALF_OPEN) {
      this.successCount++;
      if (this.successCount >= CircuitBreakerState.SUCCESS_THRESHOLD) {
        this.state = CircuitState.CLOSED;
        this.failureCount = 0;
        this.successCount = 0;
        console.log('Circuit breaker moved to CLOSED');
      }
    } else {
      this.failureCount = 0;
    }
  }
  
  onFailure(): void {
    this.failureCount++;
    this.lastFailureTime = Date.now();
    
    if (this.failureCount >= CircuitBreakerState.FAILURE_THRESHOLD) {
      this.state = CircuitState.OPEN;
      this.successCount = 0;
      console.log('Circuit breaker moved to OPEN');
    }
  }
  
  setState(newState: CircuitState): void {
    this.state = newState;
  }
}

enum CircuitState {
  CLOSED = 'CLOSED',
  OPEN = 'OPEN',
  HALF_OPEN = 'HALF_OPEN'
}

// Usage example
const paymentService = new PaymentService();
const advancedCircuitBreaker = new AdvancedCircuitBreaker();

// Using opossum circuit breaker
async function processPaymentWithOpossum(request: PaymentRequest): Promise<PaymentResult> {
  return await paymentService.processPayment(request);
}

// Using custom circuit breaker
async function processPaymentWithCustom(request: PaymentRequest): Promise<PaymentResult> {
  return await advancedCircuitBreaker.execute(
    'payment-service',
    async () => {
      const response = await axios.post('http://payment-service/api/payments/process', request);
      return response.data;
    },
    async () => {
      console.log('Using fallback for payment processing');
      return {
        id: null,
        status: 'PENDING',
        message: 'Payment queued for later processing',
        error: 'Service temporarily unavailable'
      };
    }
  );
}

// Cleanup
process.on('SIGINT', () => {
  advancedCircuitBreaker.destroy();
  process.exit(0);
});

Circuit Breaker с библиотекой hystrix-js:

import HystrixCommand from 'hystrix-js';

class PaymentHystrixCommand extends HystrixCommand {
  constructor(private request: PaymentRequest) {
    super({
      name: 'PaymentService',
      timeout: 3000,
      errorThresholdPercentage: 50,
      requestVolumeThreshold: 10,
      sleepWindowInMilliseconds: 30000
    });
  }

  run(): Promise<PaymentResult> {
    return axios.post('http://payment-service/api/payments/process', this.request)
      .then(response => response.data);
  }

  getFallback(): PaymentResult {
    console.log('Payment service fallback triggered');
    return {
      id: null,
      status: 'PENDING',
      message: 'Payment queued for later processing',
      error: 'Service temporarily unavailable'
    };
  }
}

// Usage
async function processPaymentWithHystrix(request: PaymentRequest): Promise<PaymentResult> {
  const command = new PaymentHystrixCommand(request);
  return await command.execute();
}

Rust примеры - Circuit Breaker:

// Circuit Breaker с tokio
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{Mutex, RwLock};
use tokio::time::sleep;

#[derive(Debug, Clone)]
enum CircuitState {
    Closed,
    Open,
    HalfOpen,
}

#[derive(Debug)]
struct CircuitBreakerConfig {
    failure_threshold: usize,
    timeout_duration: Duration,
    success_threshold: usize,
}

impl Default for CircuitBreakerConfig {
    fn default() -> Self {
        Self {
            failure_threshold: 5,
            timeout_duration: Duration::from_secs(60),
            success_threshold: 3,
        }
    }
}

#[derive(Debug)]
struct CircuitBreakerState {
    state: CircuitState,
    failure_count: usize,
    success_count: usize,
    last_failure_time: Option<Instant>,
}

impl Default for CircuitBreakerState {
    fn default() -> Self {
        Self {
            state: CircuitState::Closed,
            failure_count: 0,
            success_count: 0,
            last_failure_time: None,
        }
    }
}

pub struct CircuitBreaker {
    config: CircuitBreakerConfig,
    state: Arc<RwLock<CircuitBreakerState>>,
}

impl CircuitBreaker {
    pub fn new(config: CircuitBreakerConfig) -> Self {
        Self {
            config,
            state: Arc::new(RwLock::new(CircuitBreakerState::default())),
        }
    }

    pub async fn execute<F, T, E>(&self, operation: F) -> Result<T, CircuitBreakerError>
    where
        F: FnOnce() -> Result<T, E>,
        E: std::fmt::Display,
    {
        let current_state = self.get_state().await;
        
        match current_state {
            CircuitState::Open => {
                if self.should_attempt_reset().await {
                    self.set_state(CircuitState::HalfOpen).await;
                } else {
                    return Err(CircuitBreakerError::CircuitOpen);
                }
            }
            CircuitState::HalfOpen | CircuitState::Closed => {}
        }

        match operation() {
            Ok(result) => {
                self.on_success().await;
                Ok(result)
            }
            Err(error) => {
                self.on_failure().await;
                Err(CircuitBreakerError::OperationFailed(error.to_string()))
            }
        }
    }

    async fn get_state(&self) -> CircuitState {
        let state = self.state.read().await;
        state.state.clone()
    }

    async fn set_state(&self, new_state: CircuitState) {
        let mut state = self.state.write().await;
        state.state = new_state;
    }

    async fn should_attempt_reset(&self) -> bool {
        let state = self.state.read().await;
        if let Some(last_failure) = state.last_failure_time {
            last_failure.elapsed() > self.config.timeout_duration
        } else {
            false
        }
    }

    async fn on_success(&self) {
        let mut state = self.state.write().await;
        
        match state.state {
            CircuitState::HalfOpen => {
                state.success_count += 1;
                if state.success_count >= self.config.success_threshold {
                    state.state = CircuitState::Closed;
                    state.failure_count = 0;
                    state.success_count = 0;
                }
            }
            CircuitState::Closed => {
                state.failure_count = 0;
            }
            CircuitState::Open => {}
        }
    }

    async fn on_failure(&self) {
        let mut state = self.state.write().await;
        state.failure_count += 1;
        state.last_failure_time = Some(Instant::now());
        
        if state.failure_count >= self.config.failure_threshold {
            state.state = CircuitState::Open;
            state.success_count = 0;
        }
    }
}

#[derive(Debug)]
pub enum CircuitBreakerError {
    CircuitOpen,
    OperationFailed(String),
}

impl std::fmt::Display for CircuitBreakerError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            CircuitBreakerError::CircuitOpen => write!(f, "Circuit breaker is open"),
            CircuitBreakerError::OperationFailed(msg) => write!(f, "Operation failed: {}", msg),
        }
    }
}

impl std::error::Error for CircuitBreakerError {}

// Payment Service с Circuit Breaker
pub struct PaymentService {
    circuit_breaker: CircuitBreaker,
    http_client: reqwest::Client,
}

impl PaymentService {
    pub fn new() -> Self {
        let config = CircuitBreakerConfig {
            failure_threshold: 5,
            timeout_duration: Duration::from_secs(30),
            success_threshold: 3,
        };
        
        Self {
            circuit_breaker: CircuitBreaker::new(config),
            http_client: reqwest::Client::new(),
        }
    }

    pub async fn process_payment(&self, request: PaymentRequest) -> Result<PaymentResult, CircuitBreakerError> {
        self.circuit_breaker.execute(|| async {
            let response = self.http_client
                .post("http://payment-service/api/payments/process")
                .json(&request)
                .timeout(Duration::from_secs(5))
                .send()
                .await?;
            
            if response.status().is_success() {
                let result: PaymentResult = response.json().await?;
                Ok(result)
            } else {
                Err(format!("Payment service returned error: {}", response.status()))
            }
        }).await
    }

    pub async fn process_payment_with_fallback(&self, request: PaymentRequest) -> PaymentResult {
        match self.process_payment(request.clone()).await {
            Ok(result) => result,
            Err(_) => {
                // Fallback logic
                PaymentResult {
                    id: None,
                    status: "PENDING".to_string(),
                    message: "Payment queued for later processing".to_string(),
                    error: Some("Service temporarily unavailable".to_string()),
                }
            }
        }
    }
}

#[derive(Debug, serde::Serialize)]
struct PaymentRequest {
    amount: f64,
    currency: String,
    payment_method: String,
    order_id: u64,
    customer_id: u64,
}

#[derive(Debug, serde::Deserialize)]
struct PaymentResult {
    id: Option<u64>,
    status: String,
    message: String,
    error: Option<String>,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let payment_service = PaymentService::new();
    
    let request = PaymentRequest {
        amount: 100.0,
        currency: "USD".to_string(),
        payment_method: "credit_card".to_string(),
        order_id: 12345,
        customer_id: 67890,
    };
    
    let result = payment_service.process_payment_with_fallback(request).await;
    println!("Payment result: {:?}", result);
    
    Ok(())
}

Python примеры - Circuit Breaker:

# Circuit Breaker с asyncio
import asyncio
import time
from enum import Enum
from typing import Callable, Any, Optional
from dataclasses import dataclass
import aiohttp
import json

class CircuitState(Enum):
    CLOSED = "CLOSED"
    OPEN = "OPEN"
    HALF_OPEN = "HALF_OPEN"

@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5
    timeout_duration: float = 60.0  # seconds
    success_threshold: int = 3

class CircuitBreakerError(Exception):
    def __init__(self, message: str):
        self.message = message
        super().__init__(self.message)

class CircuitBreaker:
    def __init__(self, config: CircuitBreakerConfig = None):
        self.config = config or CircuitBreakerConfig()
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time: Optional[float] = None
        self._lock = asyncio.Lock()

    async def execute(self, operation: Callable[[], Any]) -> Any:
        async with self._lock:
            if self.state == CircuitState.OPEN:
                if self._should_attempt_reset():
                    self.state = CircuitState.HALF_OPEN
                else:
                    raise CircuitBreakerError("Circuit breaker is open")

        try:
            result = await operation()
            await self._on_success()
            return result
        except Exception as e:
            await self._on_failure()
            raise CircuitBreakerError(f"Operation failed: {str(e)}")

    def _should_attempt_reset(self) -> bool:
        if self.last_failure_time is None:
            return False
        return time.time() - self.last_failure_time > self.config.timeout_duration

    async def _on_success(self):
        async with self._lock:
            if self.state == CircuitState.HALF_OPEN:
                self.success_count += 1
                if self.success_count >= self.config.success_threshold:
                    self.state = CircuitState.CLOSED
                    self.failure_count = 0
                    self.success_count = 0
            elif self.state == CircuitState.CLOSED:
                self.failure_count = 0

    async def _on_failure(self):
        async with self._lock:
            self.failure_count += 1
            self.last_failure_time = time.time()
            
            if self.failure_count >= self.config.failure_threshold:
                self.state = CircuitState.OPEN
                self.success_count = 0

# Payment Service с Circuit Breaker
class PaymentService:
    def __init__(self):
        config = CircuitBreakerConfig(
            failure_threshold=5,
            timeout_duration=30.0,
            success_threshold=3
        )
        self.circuit_breaker = CircuitBreaker(config)
        self.session = aiohttp.ClientSession()

    async def process_payment(self, request: dict) -> dict:
        async def _make_request():
            async with self.session.post(
                "http://payment-service/api/payments/process",
                json=request,
                timeout=aiohttp.ClientTimeout(total=5)
            ) as response:
                if response.status == 200:
                    return await response.json()
                else:
                    raise Exception(f"Payment service returned error: {response.status}")

        try:
            return await self.circuit_breaker.execute(_make_request)
        except CircuitBreakerError:
            # Fallback logic
            return {
                "id": None,
                "status": "PENDING",
                "message": "Payment queued for later processing",
                "error": "Service temporarily unavailable"
            }

    async def process_payment_with_fallback(self, request: dict) -> dict:
        try:
            return await self.process_payment(request)
        except Exception as e:
            print(f"Payment processing failed: {e}")
            return {
                "id": None,
                "status": "PENDING",
                "message": "Payment queued for later processing",
                "error": "Service temporarily unavailable"
            }

    async def close(self):
        await self.session.close()

# Usage example
async def main():
    payment_service = PaymentService()
    
    request = {
        "amount": 100.0,
        "currency": "USD",
        "payment_method": "credit_card",
        "order_id": 12345,
        "customer_id": 67890
    }
    
    try:
        result = await payment_service.process_payment_with_fallback(request)
        print(f"Payment result: {result}")
    finally:
        await payment_service.close()

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

Технический пример - Bulkhead Pattern:

// Bulkhead Configuration
@Configuration
public class BulkheadConfig {
    
    @Bean
    public ThreadPoolTaskExecutor paymentExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("payment-");
        executor.initialize();
        return executor;
    }
    
    @Bean
    public ThreadPoolTaskExecutor notificationExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(3);
        executor.setMaxPoolSize(5);
        executor.setQueueCapacity(50);
        executor.setThreadNamePrefix("notification-");
        executor.initialize();
        return executor;
    }
}

// Service with Bulkhead Pattern
@Service
public class OrderService {
    
    @Autowired
    @Qualifier("paymentExecutor")
    private ThreadPoolTaskExecutor paymentExecutor;
    
    @Autowired
    @Qualifier("notificationExecutor")
    private ThreadPoolTaskExecutor notificationExecutor;
    
    @Autowired
    private PaymentService paymentService;
    
    @Autowired
    private NotificationService notificationService;
    
    public CompletableFuture<Order> processOrder(Order order) {
        // Payment processing in dedicated thread pool
        CompletableFuture<PaymentResult> paymentFuture = CompletableFuture
            .supplyAsync(() -> paymentService.processPayment(order.getPaymentRequest()), paymentExecutor)
            .exceptionally(ex -> {
                log.error("Payment processing failed for order: {}", order.getId(), ex);
                return new PaymentResult(null, PaymentStatus.FAILED, null, ex.getMessage());
            });
        
        // Notification processing in separate thread pool
        CompletableFuture<Void> notificationFuture = CompletableFuture
            .runAsync(() -> notificationService.sendOrderConfirmation(order), notificationExecutor)
            .exceptionally(ex -> {
                log.error("Notification failed for order: {}", order.getId(), ex);
                return null;
            });
        
        // Combine results
        return paymentFuture.thenCompose(paymentResult -> {
            if (paymentResult.isSuccess()) {
                order.setStatus(OrderStatus.PROCESSED);
                order.setPaymentId(paymentResult.getTransactionId());
            } else {
                order.setStatus(OrderStatus.FAILED);
            }
            
            return CompletableFuture.completedFuture(order);
        });
    }
}

// Timeout Pattern
@Component
public class TimeoutService {
    
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(5);
    
    public <T> CompletableFuture<T> executeWithTimeout(Supplier<T> operation, Duration timeout) {
        CompletableFuture<T> future = CompletableFuture.supplyAsync(operation);
        
        CompletableFuture<T> timeoutFuture = new CompletableFuture<>();
        scheduler.schedule(() -> {
            timeoutFuture.completeExceptionally(new TimeoutException("Operation timed out"));
        }, timeout.toMillis(), TimeUnit.MILLISECONDS);
        
        return CompletableFuture.anyOf(future, timeoutFuture)
            .thenApply(result -> (T) result);
    }
    
    public <T> T executeWithTimeoutSync(Supplier<T> operation, Duration timeout) {
        try {
            return executeWithTimeout(operation, timeout).get();
        } catch (Exception e) {
            throw new RuntimeException("Operation failed or timed out", e);
        }
    }
}

2. Retry Pattern (Паттерн повторов)

@Component
public class RetryService {
    public <T> T executeWithRetry(Supplier<T> operation, int maxAttempts) {
        Exception lastException = null;
        
        for (int attempt = 1; attempt <= maxAttempts; attempt++) {
            try {
                return operation.get();
            } catch (Exception e) {
                lastException = e;
                
                if (attempt < maxAttempts) {
                    long delay = calculateDelay(attempt);
                    try {
                        Thread.sleep(delay);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        throw new RuntimeException(ie);
                    }
                }
            }
        }
        
        throw new RuntimeException("Operation failed after " + maxAttempts + " attempts", lastException);
    }
    
    private long calculateDelay(int attempt) {
        // Exponential backoff
        return (long) Math.pow(2, attempt) * 1000;
    }
}

3. Bulkhead Pattern (Переборка)

@Configuration
public class BulkheadConfig {
    @Bean
    public ThreadPoolExecutor paymentExecutor() {
        return new ThreadPoolExecutor(
            10, 10, 0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(100),
            new ThreadFactoryBuilder().setNameFormat("payment-%d").build()
        );
    }
    
    @Bean
    public ThreadPoolExecutor notificationExecutor() {
        return new ThreadPoolExecutor(
            5, 5, 0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(50),
            new ThreadFactoryBuilder().setNameFormat("notification-%d").build()
        );
    }
}

@Service
public class OrderService {
    @Autowired
    @Qualifier("paymentExecutor")
    private Executor paymentExecutor;
    
    @Autowired
    @Qualifier("notificationExecutor")
    private Executor notificationExecutor;
    
    public void processOrder(Order order) {
        // обработка платежа в отдельном пуле
        paymentExecutor.execute(() -> {
            paymentService.processPayment(order.getPayment());
        });
        
        // отправка уведомления в отдельном пуле
        notificationExecutor.execute(() -> {
            notificationService.sendConfirmation(order.getCustomerId());
        });
    }
}

4. Timeout Pattern (Паттерн таймаута)

@Service
public class ExternalServiceClient {
    private final RestTemplate restTemplate;
    
    public ExternalServiceClient() {
        HttpComponentsClientHttpRequestFactory factory = 
            new HttpComponentsClientHttpRequestFactory();
        factory.setConnectTimeout(5000); // 5 секунд
        factory.setReadTimeout(10000);   // 10 секунд
        
        this.restTemplate = new RestTemplate(factory);
    }
    
    public String callExternalService(String data) {
        try {
            return restTemplate.postForObject(
                "http://external-service/api/process", 
                data, 
                String.class
            );
        } catch (ResourceAccessException e) {
            throw new ServiceTimeoutException("External service timeout", e);
        }
    }
}

Плавная деградация

@Service
public class RecommendationService {
    @Autowired
    private MLRecommendationService mlService;
    
    @Autowired
    private RuleBasedRecommendationService ruleService;
    
    public List<Product> getRecommendations(Long userId) {
        try {
            // пытаемся получить ML рекомендации
            return mlService.getRecommendations(userId);
        } catch (Exception e) {
            log.warn("ML service failed, falling back to rule-based", e);
            
            try {
                // fallback на правило-основанные рекомендации
                return ruleService.getRecommendations(userId);
            } catch (Exception e2) {
                log.error("All recommendation services failed", e2);
                // возвращаем популярные товары
                return getPopularProducts();
            }
        }
    }
    
    private List<Product> getPopularProducts() {
        return Arrays.asList(
            new Product(1L, "Popular Product 1"),
            new Product(2L, "Popular Product 2"),
            new Product(3L, "Popular Product 3")
        );
    }
}

---

Управление данными

Теорема CAP

CAP Theorem утверждает, что распределенная система не может одновременно обеспечить:

• Consistency (Консистентность)
• Availability (Доступность)
• Partition Tolerance (Устойчивость к разделению)

Типы консистентности

1. Strong Consistency

• Все узлы видят одинаковые данные
• Сложно в распределенных системах
• Высокая задержка

2. Eventual Consistency

• Данные в итоге станут консистентными
• Подходит для большинства случаев
• Низкая задержка

3. Weak Consistency

• Нет гарантий консистентности
• Используется редко

Паттерны управления данными

1. Event Sourcing

@Entity
public class OrderEvent {
    private Long id;
    private String aggregateId;
    private String eventType;
    private String eventData;
    private LocalDateTime occurredAt;
}

@Service
public class OrderEventStore {
    public void saveEvent(OrderEvent event) {
        eventRepository.save(event);
    }
    
    public List<OrderEvent> getEvents(String aggregateId) {
        return eventRepository.findByAggregateIdOrderByOccurredAt(aggregateId);
    }
}

@Service
public class OrderProjectionService {
    public OrderView buildOrderView(String orderId) {
        List<OrderEvent> events = eventStore.getEvents(orderId);
        OrderView view = new OrderView();
        
        for (OrderEvent event : events) {
            applyEvent(view, event);
        }
        
        return view;
    }
    
    private void applyEvent(OrderView view, OrderEvent event) {
        switch (event.getEventType()) {
            case "OrderCreated":
                view.setId(event.getAggregateId());
                view.setStatus("CREATED");
                break;
            case "OrderConfirmed":
                view.setStatus("CONFIRMED");
                break;
            case "OrderShipped":
                view.setStatus("SHIPPED");
                break;
        }
    }
}

2. CQRS (Command Query Responsibility Segregation)

// Command Side
@Service
public class OrderCommandService {
    public void createOrder(CreateOrderCommand command) {
        Order order = new Order(command);
        orderRepository.save(order);
        
        // обновляем read model
        orderReadModelService.updateReadModel(order);
    }
}

// Query Side
@Service
public class OrderQueryService {
    public OrderView getOrderView(OrderId orderId) {
        return orderReadModelRepository.findById(orderId);
    }
    
    public List<OrderView> getOrdersByCustomer(CustomerId customerId) {
        return orderReadModelRepository.findByCustomerId(customerId);
    }
}

3. Saga Pattern

@Service
public class OrderSaga {
    public void processOrder(CreateOrderRequest request) {
        Order order = orderService.createOrder(request);
        
        try {
            // шаг 1: резервирование товаров
            inventoryService.reserveItems(order.getId(), request.getItems());
            
            // шаг 2: обработка платежа
            paymentService.processPayment(order.getId(), request.getPayment());
            
            // шаг 3: подтверждение заказа
            orderService.confirmOrder(order.getId());
            
        } catch (Exception e) {
            // компенсирующие транзакции
            compensateOrder(order.getId());
        }
    }
    
    private void compensateOrder(OrderId orderId) {
        try {
            inventoryService.releaseItems(orderId);
            paymentService.refundPayment(orderId);
            orderService.cancelOrder(orderId);
        } catch (Exception e) {
            log.error("Compensation failed for order: " + orderId, e);
        }
    }
}

---

Кэширование

Стратегии кэширования

1. Cache-Aside (Lazy Loading)

Технический пример - Advanced Cache-Aside Pattern:

@Service
public class UserService {
    
    @Autowired
    private RedisTemplate<String, User> redisTemplate;
    
    @Autowired
    private UserRepository userRepository;
    
    private static final String USER_CACHE_KEY_PREFIX = "user:";
    private static final Duration CACHE_TTL = Duration.ofMinutes(30);
    
    public User findById(Long id) {
        String cacheKey = USER_CACHE_KEY_PREFIX + id;
        
        try {
            // Проверяем кэш
            User cachedUser = redisTemplate.opsForValue().get(cacheKey);
            if (cachedUser != null) {
                log.debug("Cache hit for user: {}", id);
                return cachedUser;
            }
            
            log.debug("Cache miss for user: {}", id);
            
            // Загружаем из БД
            User user = userRepository.findById(id)
                .orElseThrow(() -> new UserNotFoundException("User not found: " + id));
            
            // Сохраняем в кэш асинхронно
            CompletableFuture.runAsync(() -> {
                try {
                    redisTemplate.opsForValue().set(cacheKey, user, CACHE_TTL);
                    log.debug("User cached: {}", id);
                } catch (Exception e) {
                    log.error("Failed to cache user: {}", id, e);
                }
            });
            
            return user;
            
        } catch (Exception e) {
            log.error("Error retrieving user: {}", id, e);
            throw new UserServiceException("Failed to retrieve user", e);
        }
    }
    
    public List<User> findByIds(List<Long> ids) {
        List<String> cacheKeys = ids.stream()
            .map(id -> USER_CACHE_KEY_PREFIX + id)
            .collect(Collectors.toList());
        
        // Batch get from cache
        List<User> cachedUsers = redisTemplate.opsForValue().multiGet(cacheKeys);
        
        List<User> result = new ArrayList<>();
        List<Long> missingIds = new ArrayList<>();
        
        for (int i = 0; i < ids.size(); i++) {
            if (cachedUsers.get(i) != null) {
                result.add(cachedUsers.get(i));
            } else {
                missingIds.add(ids.get(i));
            }
        }
        
        // Load missing users from DB
        if (!missingIds.isEmpty()) {
            List<User> dbUsers = userRepository.findByIdIn(missingIds);
            
            // Cache the loaded users
            Map<String, User> usersToCache = new HashMap<>();
            for (User user : dbUsers) {
                usersToCache.put(USER_CACHE_KEY_PREFIX + user.getId(), user);
            }
            
            if (!usersToCache.isEmpty()) {
                redisTemplate.opsForValue().multiSet(usersToCache);
                redisTemplate.expire(usersToCache.keySet(), CACHE_TTL);
            }
            
            result.addAll(dbUsers);
        }
        
        return result;
    }
    
    public void invalidateUserCache(Long userId) {
        String cacheKey = USER_CACHE_KEY_PREFIX + userId;
        redisTemplate.delete(cacheKey);
        log.debug("Cache invalidated for user: {}", userId);
    }
}

2. Write-Through

@Service
public class UserService {
    public void updateUser(User user) {
        // обновляем БД
        userRepository.save(user);
        
        // обновляем кэш
        String key = "user:" + user.getId();
        redisTemplate.opsForValue().set(key, user, Duration.ofMinutes(30));
    }
}

3. Write-Behind (Write-Back)

@Service
public class UserService {
    @Async
    public void updateUser(User user) {
        // обновляем кэш сразу
        String key = "user:" + user.getId();
        redisTemplate.opsForValue().set(key, user, Duration.ofMinutes(30));
        
        // обновляем БД асинхронно
        CompletableFuture.runAsync(() -> {
            userRepository.save(user);
        });
    }
}

Распределенное кэширование

Кластер Redis

@Configuration
public class RedisConfig {
    @Bean
    public LettuceConnectionFactory redisConnectionFactory() {
        RedisClusterConfiguration clusterConfig = new RedisClusterConfiguration();
        clusterConfig.addClusterNode(new RedisNode("127.0.0.1", 7000));
        clusterConfig.addClusterNode(new RedisNode("127.0.0.1", 7001));
        clusterConfig.addClusterNode(new RedisNode("127.0.0.1", 7002));
        
        return new LettuceConnectionFactory(clusterConfig);
    }
}

Memcached

@Service
public class MemcachedService {
    @Autowired
    private MemcachedClient memcachedClient;
    
    public void set(String key, Object value, int expiration) {
        try {
            memcachedClient.set(key, expiration, value);
        } catch (Exception e) {
            log.error("Failed to set cache", e);
        }
    }
    
    public Object get(String key) {
        try {
            return memcachedClient.get(key);
        } catch (Exception e) {
            log.error("Failed to get cache", e);
            return null;
        }
    }
}

---

Load Balancing

Алгоритмы балансировки нагрузки

1. Round Robin

public class RoundRobinLoadBalancer {
    private final List<String> servers;
    private int currentIndex = 0;
    
    public String getServer() {
        String server = servers.get(currentIndex);
        currentIndex = (currentIndex + 1) % servers.size();
        return server;
    }
}

2. Weighted Round Robin

public class WeightedRoundRobinLoadBalancer {
    private final List<Server> servers;
    private int currentWeight = 0;
    private int currentIndex = -1;
    
    public String getServer() {
        int totalWeight = servers.stream().mapToInt(Server::getWeight).sum();
        
        while (true) {
            currentIndex = (currentIndex + 1) % servers.size();
            
            if (currentIndex == 0) {
                currentWeight = currentWeight - 1;
                if (currentWeight <= 0) {
                    currentWeight = totalWeight;
                }
            }
            
            if (servers.get(currentIndex).getWeight() >= currentWeight) {
                return servers.get(currentIndex).getAddress();
            }
        }
    }
}

3. Least Connections

public class LeastConnectionsLoadBalancer {
    private final Map<String, AtomicInteger> connectionCounts = new ConcurrentHashMap<>();
    
    public String getServer() {
        return connectionCounts.entrySet().stream()
            .min(Map.Entry.comparingByValue())
            .map(Map.Entry::getKey)
            .orElse(null);
    }
    
    public void incrementConnections(String server) {
        connectionCounts.computeIfAbsent(server, k -> new AtomicInteger(0)).incrementAndGet();
    }
    
    public void decrementConnections(String server) {
        connectionCounts.computeIfPresent(server, (k, v) -> {
            v.decrementAndGet();
            return v.get() > 0 ? v : null;
        });
    }
}

Проверки здоровья

@Component
public class HealthChecker {
    private final RestTemplate restTemplate;
    
    public HealthChecker() {
        HttpComponentsClientHttpRequestFactory factory = 
            new HttpComponentsClientHttpRequestFactory();
        factory.setConnectTimeout(5000);
        factory.setReadTimeout(5000);
        this.restTemplate = new RestTemplate(factory);
    }
    
    public boolean isHealthy(String serverUrl) {
        try {
            ResponseEntity<String> response = restTemplate.getForEntity(
                serverUrl + "/health", String.class);
            return response.getStatusCode().is2xxSuccessful();
        } catch (Exception e) {
            return false;
        }
    }
    
    @Scheduled(fixedRate = 30000) // каждые 30 секунд
    public void checkHealth() {
        servers.forEach(server -> {
            boolean healthy = isHealthy(server.getUrl());
            server.setHealthy(healthy);
            
            if (!healthy) {
                log.warn("Server {} is unhealthy", server.getUrl());
            }
        });
    }
}

---

Мониторинг производительности

Метрики производительности

1. Application Metrics

@Component
public class MetricsCollector {
    private final MeterRegistry meterRegistry;
    
    public MetricsCollector(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
    }
    
    public void recordRequestDuration(String endpoint, Duration duration) {
        Timer.Sample sample = Timer.start(meterRegistry);
        sample.stop(Timer.builder("http.request.duration")
            .tag("endpoint", endpoint)
            .register(meterRegistry));
    }
    
    public void recordError(String endpoint, String errorType) {
        Counter.builder("http.request.errors")
            .tag("endpoint", endpoint)
            .tag("error_type", errorType)
            .register(meterRegistry)
            .increment();
    }
}

2. Business Metrics

@Service
public class OrderMetrics {
    private final MeterRegistry meterRegistry;
    
    public void recordOrderCreated() {
        Counter.builder("orders.created")
            .register(meterRegistry)
            .increment();
    }
    
    public void recordOrderValue(BigDecimal value) {
        DistributionSummary.builder("orders.value")
            .register(meterRegistry)
            .record(value.doubleValue());
    }
}

APM (Мониторинг производительности приложений)

Пример New Relic

@RestController
public class OrderController {
    @Autowired
    private NewRelic newRelic;
    
    @PostMapping("/orders")
    public ResponseEntity<Order> createOrder(@RequestBody CreateOrderRequest request) {
        newRelic.addCustomParameter("customer_id", request.getCustomerId());
        newRelic.addCustomParameter("order_value", request.getTotalValue());
        
        try {
            Order order = orderService.createOrder(request);
            newRelic.recordMetric("Orders/Created", 1);
            return ResponseEntity.ok(order);
        } catch (Exception e) {
            newRelic.recordMetric("Orders/Errors", 1);
            throw e;
        }
    }
}

---

Практические задания

Задание 1: Circuit Breaker

Реализуйте Circuit Breaker для внешнего сервиса с экспоненциальным backoff.

Задание 2: Кэширование

Настройте многоуровневое кэширование (L1: local, L2: Redis).

Задание 3: Load Balancer

Создайте простой load balancer с health checks.

---

Дополнительные ресурсы

Книги

• "Designing Data-Intensive Applications" by Martin Kleppmann
• "Release It!" by Michael Nygard
• "Site Reliability Engineering" by Google

Инструменты

• Monitoring: Prometheus, Grafana, New Relic
• Caching: Redis, Memcached, Hazelcast
• Load Balancing: HAProxy, NGINX, AWS ALB

---

Контрольные вопросы

1. Какие типы масштабирования вы знаете?
2. Что такое CAP Theorem и как оно влияет на архитектуру?
3. Какие паттерны отказоустойчивости вы знаете?
4. Как работает Circuit Breaker?
5. Какие стратегии кэширования существуют?
6. Как реализовать health checks?
7. Какие метрики важны для мониторинга производительности?
8. Как обеспечить graceful degradation?