# Модуль 4: Масштабируемость и отказоустойчивость ## Содержание модуля 1. [Основы масштабируемости](#основы-масштабируемости) 2. [Паттерны масштабирования](#паттерны-масштабирования) 3. [Отказоустойчивость](#отказоустойчивость) 4. [Управление данными](#управление-данными) 5. [Кэширование](#кэширование) 6. [Load Balancing](#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 (Бессостоятельные сервисы) **Принцип:** Сервисы не должны хранить состояние между запросами. **Преимущества:** - Легкое горизонтальное масштабирование - Отказоустойчивость - Простота балансировки нагрузки **Реализация:** ```java // Плохо: Stateful сервис @Service public class StatefulUserService { private Map userCache = new HashMap<>(); public User getUser(String id) { // Состояние хранится в памяти сервиса return userCache.get(id); } } // Хорошо: Stateless сервис @Service public class StatelessUserService { @Autowired private UserRepository userRepository; @Autowired private RedisTemplate 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 сервисы:** ```rust use std::sync::Arc; use tokio::sync::RwLock; use redis::Client; pub struct StatelessUserService { user_repository: Arc, redis_client: Arc, } impl StatelessUserService { pub fn new(user_repository: Arc, redis_client: Arc) -> Self { Self { user_repository, redis_client, } } pub async fn get_user(&self, id: String) -> Result, 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 { 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 сервисы:** ```python 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 сервисы:** ```typescript 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 { // Проверка кэша 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 { 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 { 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 (Шардирование базы данных) **Принцип:** Разделение данных на несколько независимых баз данных. **Стратегии шардирования:** ```java // Стратегия шардирования по пользователю @Service public class ShardedUserService { @Autowired private List 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 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 (Реплики для чтения) **Принцип:** Использование отдельных реплик для операций чтения. ```java @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 getAllUsers() { // Чтение из replica JdbcTemplate replicaTemplate = new JdbcTemplate(readOnlyDataSource); String sql = "SELECT * FROM users"; return replicaTemplate.query(sql, new UserRowMapper()); } } ``` #### 4. Caching Strategies (Стратегии кэширования) **Cache-Aside Pattern:** ```java @Service public class CacheAsideUserService { @Autowired private UserRepository userRepository; @Autowired private RedisTemplate 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:** ```java @Service public class WriteThroughUserService { @Autowired private UserRepository userRepository; @Autowired private RedisTemplate 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:** ```java @Service public class WriteBehindUserService { @Autowired private UserRepository userRepository; @Autowired private RedisTemplate redisTemplate; @Async public CompletableFuture 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 keys = redisTemplate.keys("user:*"); for (String key : keys) { User user = redisTemplate.opsForValue().get(key); if (user != null) { userRepository.save(user); } } } } ``` ### Архитектурные принципы масштабируемости #### 1. Microservices Architecture **Принципы:** - Разделение по бизнес-доменам - Независимое развертывание - Слабая связанность - Высокая когезия ```java // Пример микросервиса пользователей @RestController @RequestMapping("/api/users") public class UserController { @Autowired private UserService userService; @GetMapping("/{id}") public ResponseEntity getUser(@PathVariable String id) { User user = userService.getUser(id); return ResponseEntity.ok(user); } @PostMapping public ResponseEntity 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 **Принципы:** - События как основной способ коммуникации - Асинхронная обработка - Слабая связанность сервисов ```java // 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 - Мониторинг и логирование - Кэширование ответов ```java // 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. Ключевые метрики ```java @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. Автоматическое масштабирование ```yaml # 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:** ```java @RestController @RequestMapping("/api/users") public class UserController { @Autowired private UserService userService; @GetMapping("/{id}") public ResponseEntity getUser(@PathVariable Long id) { // Stateless - нет сохранения состояния между запросами User user = userService.getUser(id); return ResponseEntity.ok(user); } @PostMapping public ResponseEntity 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:** ```java // 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 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 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 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 shardTemplates() { Map templates = new HashMap<>(); Map dataSources = getShardDataSources(); for (Map.Entry 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:** ```java // 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 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 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 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 ```sql -- 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 ```java 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 ```java 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 (Реплики для чтения) **Принцип:** Разделение операций чтения и записи. ```yaml # Master-Slave конфигурация Master DB (Write) → Slave DB 1 (Read) → Slave DB 2 (Read) → Slave DB 3 (Read) ``` ```java @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 (Кэш в стороне) ```java @Service public class UserService { @Autowired private RedisTemplate 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 (Сквозная запись) ```java @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 (Запись с задержкой) ```java @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 (Событийное программирование) **Принцип:** Хранение изменений состояния как последовательности событий. **Преимущества:** - Полная история изменений - Возможность воспроизведения состояния - Аудит и отладка - Временные запросы **Реализация:** ```java // Event Store @Component public class EventStore { @Autowired private EventRepository eventRepository; public void saveEvents(String aggregateId, List events, int expectedVersion) { List storedEvents = events.stream() .map(event -> new StoredEvent(aggregateId, event, expectedVersion)) .collect(Collectors.toList()); eventRepository.saveAll(storedEvents); } public List getEvents(String aggregateId) { return eventRepository.findByAggregateIdOrderByVersion(aggregateId) .stream() .map(StoredEvent::getEvent) .collect(Collectors.toList()); } public List 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 uncommittedEvents = new ArrayList<>(); protected void apply(DomainEvent event) { uncommittedEvents.add(event); handle(event); } protected abstract void handle(DomainEvent event); public List 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 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 uncommittedEvents = user.getUncommittedEvents(); if (!uncommittedEvents.isEmpty()) { eventStore.saveEvents(user.getId(), uncommittedEvents, user.getVersion()); user.markEventsAsCommitted(); } } } ``` **Rust примеры - Event Sourcing:** ```rust 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, } #[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 { self.uncommitted_events.clone() } pub fn mark_events_as_committed(&mut self) { self.uncommitted_events.clear(); self.version += 1; } } pub struct EventStore { events: HashMap>, } impl EventStore { pub fn new() -> Self { Self { events: HashMap::new(), } } pub fn save_events(&mut self, aggregate_id: &str, events: Vec) { 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 { self.events.get(aggregate_id).cloned().unwrap_or_default() } } ``` #### 2. CQRS (Command Query Responsibility Segregation) **Принцип:** Разделение операций чтения и записи на разные модели. **Преимущества:** - Оптимизация для чтения и записи - Независимое масштабирование - Различные модели данных - Упрощение сложных доменов **Реализация:** ```java // Command Side @RestController @RequestMapping("/api/users/commands") public class UserCommandController { @Autowired private UserCommandService userCommandService; @PostMapping public ResponseEntity createUser(@RequestBody CreateUserCommand command) { String userId = userCommandService.createUser(command); return ResponseEntity.ok(userId); } @PutMapping("/{id}/email") public ResponseEntity 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 getUser(@PathVariable String id) { UserView user = userQueryService.getUser(id); return ResponseEntity.ok(user); } @GetMapping public ResponseEntity> getUsers(@RequestParam(required = false) String status) { List 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 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:** ```python 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:** ```java // Order Saga Events public class OrderCreatedEvent { private String orderId; private String customerId; private BigDecimal amount; private List 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 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 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:** ```java // 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 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 (Переборка) **Принцип:** Изоляция критически важных ресурсов от отказа других ресурсов. ```java // 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 createUser(CreateUserRequest request) { return CompletableFuture.supplyAsync(() -> { // Обработка пользователя в изолированном пуле return processUserCreation(request); }, userServiceExecutor); } public CompletableFuture processPayment(PaymentRequest request) { return CompletableFuture.supplyAsync(() -> { // Обработка платежа в изолированном пуле return processPaymentInternal(request); }, paymentServiceExecutor); } public CompletableFuture sendNotification(NotificationRequest request) { return CompletableFuture.runAsync(() -> { // Отправка уведомления в изолированном пуле sendNotificationInternal(request); }, notificationServiceExecutor); } } ``` #### 5. Database Per Service Pattern **Принцип:** Каждый микросервис имеет собственную базу данных. ```java // 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:** ```java // 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 processPayment(PaymentRequest request) { return CompletableFuture.supplyAsync(() -> { try { ResponseEntity 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 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 circuitStates = new ConcurrentHashMap<>(); private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); public T execute(String serviceName, Supplier operation, Supplier 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:** ```typescript 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 { try { return await this.circuitBreaker.fire(request); } catch (error) { console.error('Payment processing failed:', error); throw error; } } private async callPaymentService(request: PaymentRequest): Promise { 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 { 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 = new Map(); private scheduler: NodeJS.Timeout; constructor() { this.scheduler = setInterval(() => { this.checkCircuitStates(); }, 1000); } async execute( serviceName: string, operation: () => Promise, fallback: () => Promise ): Promise { 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 { return await paymentService.processPayment(request); } // Using custom circuit breaker async function processPaymentWithCustom(request: PaymentRequest): Promise { 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:** ```typescript 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 { 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 { const command = new PaymentHystrixCommand(request); return await command.execute(); } ``` **Rust примеры - Circuit Breaker:** ```rust // 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, } 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>, } impl CircuitBreaker { pub fn new(config: CircuitBreakerConfig) -> Self { Self { config, state: Arc::new(RwLock::new(CircuitBreakerState::default())), } } pub async fn execute(&self, operation: F) -> Result where F: FnOnce() -> Result, 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 { 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, status: String, message: String, error: Option, } #[tokio::main] async fn main() -> Result<(), Box> { 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:** ```python # 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:** ```java // 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 processOrder(Order order) { // Payment processing in dedicated thread pool CompletableFuture 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 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 CompletableFuture executeWithTimeout(Supplier operation, Duration timeout) { CompletableFuture future = CompletableFuture.supplyAsync(operation); CompletableFuture 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 executeWithTimeoutSync(Supplier operation, Duration timeout) { try { return executeWithTimeout(operation, timeout).get(); } catch (Exception e) { throw new RuntimeException("Operation failed or timed out", e); } } } ``` #### 2. Retry Pattern (Паттерн повторов) ```java @Component public class RetryService { public T executeWithRetry(Supplier 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 (Переборка) ```java @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 (Паттерн таймаута) ```java @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); } } } ``` ### Плавная деградация ```java @Service public class RecommendationService { @Autowired private MLRecommendationService mlService; @Autowired private RuleBasedRecommendationService ruleService; public List 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 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 ```java @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 getEvents(String aggregateId) { return eventRepository.findByAggregateIdOrderByOccurredAt(aggregateId); } } @Service public class OrderProjectionService { public OrderView buildOrderView(String orderId) { List 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) ```java // 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 getOrdersByCustomer(CustomerId customerId) { return orderReadModelRepository.findByCustomerId(customerId); } } ``` #### 3. Saga Pattern ```java @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:** ```java @Service public class UserService { @Autowired private RedisTemplate 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 findByIds(List ids) { List cacheKeys = ids.stream() .map(id -> USER_CACHE_KEY_PREFIX + id) .collect(Collectors.toList()); // Batch get from cache List cachedUsers = redisTemplate.opsForValue().multiGet(cacheKeys); List result = new ArrayList<>(); List 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 dbUsers = userRepository.findByIdIn(missingIds); // Cache the loaded users Map 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 ```java @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) ```java @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 ```java @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 ```java @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 ```java public class RoundRobinLoadBalancer { private final List servers; private int currentIndex = 0; public String getServer() { String server = servers.get(currentIndex); currentIndex = (currentIndex + 1) % servers.size(); return server; } } ``` #### 2. Weighted Round Robin ```java public class WeightedRoundRobinLoadBalancer { private final List 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 ```java public class LeastConnectionsLoadBalancer { private final Map 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; }); } } ``` ### Проверки здоровья ```java @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 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 ```java @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 ```java @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 ```java @RestController public class OrderController { @Autowired private NewRelic newRelic; @PostMapping("/orders") public ResponseEntity 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?