Какая была самая сложная задача?

Реализация распределённой системы управления очередью заданий в реальном времени

Контекст и проблема

В одном из проектов, связанных с обработкой мультимедийного контента, возникла задача: необходимо было реализовать систему для параллельной обработки тысяч видеофайлов с гарантированной доставкой и отслеживанием прогресса. Основные сложности:

• Высокая нагрузка: до 10 000 задач в день, каждая — цепочка из 5-7 этапов (транскодирование, анализ, метаданные).
• Отказоустойчивость: система должна переживать падение отдельных воркеров без потери данных.
• Мониторинг: клиенту необходимо видеть прогресс каждой задачи в реальном времени.
• Разнородное железо: задачи выполнялись на кластере из машин разной мощности (CPU/GPU).

Изначальная реализация на чистом Grand Central Dispatch (GCD) и OperationQueue быстро показала свою недостаточность — очереди были в памяти, задачи терялись при перезапуске приложения, не было распределения между несколькими инстансами сервиса.

Архитектурное решение

Было принято решение разработать гибридную систему, сочетающую локальные очереди и внешнее хранилище состояния.

1. Выбор технологий стека

• Локальная очередь: OperationQueue с кастомными AsynchronousOperation для контроля зависимостей между этапами обработки одного файла.
• Хранилище состояния: Redis (в качестве быстрой и надежной key-value базы для статусов и контрольных точек).
• Коммуникация: WebSocket (через URLSessionWebSocketTask) для реального времени и Redis Pub/Sub для межпроцессного взаимодействия между несколькими инстансами обработчика.

2. Ключевые компоненты системы

Асинхронная операция с возможностью паузы и возобновления

class MediaProcessingOperation: AsynchronousOperation {
    let taskId: String
    let stages: [ProcessingStage]
    private var currentStageIndex = 0
    private let redisClient: RedisClient
    private var websocketConnection: WebSocketConnection?

    override func main() {
        guard !isCancelled else { return }
        
        // Восстановление состояния из Redis при старте
        restoreState { [weak self] restoredIndex in
            self?.currentStageIndex = restoredIndex
            self?.processNextStage()
        }
    }
    
    private func processNextStage() {
        guard currentStageIndex < stages.count else {
            finish()
            return
        }
        
        let stage = stages[currentStageIndex]
        updateStatus(.running, stage: stage.name)
        
        stage.execute { [weak self] result in
            guard let self = self else { return }
            
            switch result {
            case .success:
                self.currentStageIndex += 1
                self.saveCheckpoint() // Сохраняем прогресс в Redis
                self.processNextStage()
            case .failure(let error):
                self.handleFailure(error)
            }
        }
    }
    
    private func saveCheckpoint() {
        let key = "task:\(taskId):checkpoint"
        redisClient.set(key, value: "\(currentStageIndex)")
    }
}

Менеджер очередей с балансировкой нагрузки

class DistributedQueueManager {
    private let localQueue = OperationQueue()
    private let redisClient: RedisClient
    private let websocketManager: WebSocketManager
    
    init() {
        localQueue.maxConcurrentOperationCount = ProcessInfo.processInfo.activeProcessorCount
        subscribeToGlobalChannel()
    }
    
    private func subscribeToGlobalChannel() {
        // Подписка на Pub/Channel для получения задач от диспетчера
        redisClient.subscribe(to: "tasks:channel") { [weak self] message in
            self?.decodeAndScheduleTask(message)
        }
    }
    
    private func decodeAndScheduleTask(_ message: String) {
        guard let task = try? JSONDecoder().decode(MediaTask.self, from: message.data(using: .utf8)!) else {
            return
        }
        
        // Определение типа задачи для балансировки (CPU/GPU bound)
        let operation = createOperation(for: task)
        
        // Приоритет на основе нагрузки
        operation.queuePriority = calculatePriority(for: task)
        localQueue.addOperation(operation)
        
        // Отправка подтверждения о принятии задачи
        websocketManager.sendStatusUpdate(for: task.id, status: .queued)
    }
}

Технические сложности и их преодоление

1. Согласованность состояния: Самой болезненной проблемой было обеспечение консистентности данных между памятью операции и Redis. Решили через checkpointing после каждого этапа и механизм retry с экспоненциальной задержкой.

2. Утечки памяти: Длительные операции с обратными вызовами создавали retain cycles. Применили строгую дисциплину с [weak self] и инструменты Instruments (Allocations, Leaks) для отлова.

3. Балансировка между узлами: Реализовали простой алгоритм на основе приоритетов и типа задачи. CPU-intensive задачи шли на машины с многоядерными процессорами, GPU — на машины с видеокартами.

4. Восстановление после сбоя: Каждая операция при инициализации проверяла наличие "зависшего" состояния в Redis и могла продолжить с последнего checkpoint.

Результат и выводы

Система успешно обрабатывала ~15 000 задач ежедневно с uptime 99.8%. Задержка между этапами — менее 50 мс. Ключевые уроки:

• Не изобретать велосипед для всего: сначала рассмотреть готовые решения (RabbitMQ, Apache Kafka), но если они не подходят по специфике — архитектура "локальная очередь + внешнее состояние" мощна.
• Проектировать для отказа с первого дня: checkpointing, idempotent operations, graceful degradation.
• Инструменты мониторинга — обязательны: без графиков в Grafana и логов в ELKStack отладка такой системы невозможна.

Эта задача стала глубоким погружением в конкурентное программирование, распределенные системы и показала, что даже в мобильной разработке иногда приходится решать серверные по своей сложности проблемы.