Разработка консольного менеджера загрузок на Kotlin

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

Чтобы реализовать проект, похожий по логике на ab-download-manager, но работающий исключительно через интерфейс командной строки (CLI), нам потребуется надежный фундамент. Этим фундаментом станет язык программирования Kotlin и интегрированная среда разработки IntelliJ IDEA.

Экосистема: Kotlin и JVM

Прежде чем устанавливать программы, необходимо понять, с чем именно мы будем работать. Kotlin — это современный, статически типизированный язык программирования, разработанный компанией JetBrains.

Главная особенность Kotlin заключается в том, что в своей классической и самой популярной форме он компилируется не напрямую в машинный код процессора (как C или C++), а в байт-код. Этот байт-код затем выполняется специальной программой — JVM (Java Virtual Machine), или виртуальной машиной Java.

Виртуальная машина Java — это программная среда, которая переводит платформонезависимый байт-код в инструкции, понятные конкретной операционной системе (Windows, macOS или Linux).

Oracle Java Documentation

Представьте, что вы написали книгу на эсперанто (байт-код). Чтобы жители разных стран (операционные системы) могли её прочитать, в каждой стране сидит переводчик-синхронист (JVM), который на лету переводит эсперанто на местный язык.

Для системного программирования это означает следующее:

1. Кроссплатформенность: Написанный нами менеджер загрузок будет работать на любой ОС без изменения кода.
2. Управление памятью: Нам не нужно вручную выделять и освобождать память для скачанных фрагментов файлов — за нас это сделает Сборщик мусора (Garbage Collector) внутри JVM.
3. Доступ к потокам ОС: JVM предоставляет удобные абстракции для работы с потоками операционной системы, что критически важно для параллельной загрузки.

Инструментарий: JDK и IntelliJ IDEA

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

Первый — это JDK (Java Development Kit). Это набор инструментов для разработчика, который включает в себя саму JVM (чтобы запускать программы), компилятор (чтобы переводить код в байт-код) и стандартные библиотеки (готовые функции для работы с сетью, файлами и математикой).

Второй — это IDE (Integrated Development Environment), интегрированная среда разработки. Можно писать код в обычном блокноте и компилировать его через командную строку, но это сродни попытке построить дом с помощью одного лишь перочинного ножа. IDE — это ваш экскаватор, подъемный кран и лазерный уровень в одном флаконе.

Мы будем использовать IntelliJ IDEA от JetBrains. Она изначально создавалась с прицелом на Java и Kotlin, поэтому понимает этот код лучше любой другой среды. Она будет подсказывать названия функций, автоматически исправлять ошибки и помогать в отладке многопоточных процессов.

Установка через JetBrains Toolbox

Самый надежный способ установить IntelliJ IDEA — использовать утилиту JetBrains Toolbox App. Это менеджер версий, который позволяет легко обновлять среду разработки и управлять проектами.

1. Скачайте JetBrains Toolbox с официального сайта.
2. Установите и запустите приложение.
3. В списке доступных инструментов найдите IntelliJ IDEA Community Edition (это бесплатная версия, её возможностей более чем достаточно для нашей курсовой работы) и нажмите «Install».

Использование Toolbox избавляет от необходимости вручную настраивать переменные среды и пути к исполняемым файлам в операционной системе.

Создание первого проекта

Запустите IntelliJ IDEA. На стартовом экране нажмите кнопку New Project (Новый проект). Перед вами появится окно конфигурации. От того, как мы настроим проект сейчас, зависит удобство всей дальнейшей разработки.

Заполните поля следующим образом:

• Name: ConsoleDownloadManager (или любое другое имя вашей курсовой).
• Location: Выберите папку, где будут храниться ваши проекты.
• Language: Обязательно выберите Kotlin.
• Build system: Выберите Gradle.
• JDK: Если в выпадающем списке пусто, нажмите Download JDK и выберите версию (рекомендуется версия 17 или 21 от любого вендора, например, Amazon Corretto или Eclipse Temurin).
• Gradle DSL: Выберите Kotlin.

После нажатия кнопки Create среда разработки начнет скачивать необходимые файлы и настраивать структуру папок. Этот процесс называется индексацией и синхронизацией. В первый раз он может занять несколько минут.

Почему именно Gradle?

В настройках мы выбрали Gradle в качестве системы сборки (Build system).

Современные программы редко пишутся с нуля. Для нашего менеджера загрузок нам понадобятся сторонние библиотеки: например, для красивого вывода прогресс-бара в консоль или для выполнения сложных HTTP-запросов.

Gradle — это автоматизированная система управления проектом. Вместо того чтобы вручную скачивать файлы библиотек из интернета и класть их в папку с проектом, мы просто пишем в специальном конфигурационном файле: «Мне нужна библиотека Ktor версии 3.2.0». Gradle сам пойдет в интернет, найдет эту библиотеку, скачает её, проверит совместимость и подключит к нашему коду.

Структура проекта

Когда проект загрузится, обратите внимание на панель слева — Project Tool Window. Там вы увидите древовидную структуру файлов. Для нас сейчас важны два элемента:

1. Файл build.gradle.kts
2. Директория src/main/kotlin

Конфигурация сборки: build.gradle.kts

Откройте файл build.gradle.kts. Расширение .kts означает Kotlin Script. Это значит, что инструкции для сборщика проектов мы пишем на том же языке, что и саму программу.

Внутри вы увидите блок dependencies (зависимости). Именно сюда мы будем добавлять сторонние библиотеки по мере усложнения курсовой работы. Например, когда мы дойдем до работы с сетью, мы добавим сюда клиент Ktor.

Пока что мы не будем ничего менять, но важно понимать, что любое изменение в этом файле требует нажатия иконки слона со стрелочками (Sync Project with Gradle Files), которая появится в правом верхнем углу редактора. Это команда для Gradle: «Я изменил список покупок, сходи в магазин и принеси новые библиотеки».

Исходный код: src/main/kotlin

Разверните папку src, затем main, затем kotlin. Это святая святых нашего проекта — место, где будет жить весь наш исходный код.

Щелкните правой кнопкой мыши по папке kotlin, выберите New -> Kotlin Class/File. Введите имя Main и выберите тип File.

В открывшемся пустом файле напишите следующий код:

fun main(args: Array<String>) {
    println("Console Download Manager инициализирован!")

    if (args.isNotEmpty()) {
        println("Получены аргументы командной строки: ${args.joinToString()}")
    } else {
        println("Запустите программу, указав URL файла для скачивания.")
    }
}

Разберем этот фрагмент подробно, так как он является точкой входа в любую программу на Kotlin.

Ключевое слово fun означает объявление функции.

Функция main — это главная функция. Когда операционная система (через JVM) запускает вашу программу, она ищет именно функцию с именем main и начинает выполнять команды внутри неё сверху вниз.

Параметр args: Array<String> — это массив строк. Поскольку мы пишем консольное приложение, пользователь будет запускать его из терминала, передавая параметры. Например:

java -jar downloader.jar https://example.com/file.iso 4

В этом случае https://example.com/file.iso и 4 (количество потоков) попадут в массив args, и мы сможем прочитать их внутри программы. В нашем коде мы проверяем, не пуст ли массив (args.isNotEmpty()), и выводим переданные аргументы на экран.

Функция println() выводит текст в стандартный поток вывода (stdout) и добавляет перенос строки.

Запуск первой программы

Слева от строки fun main вы увидите зеленую стрелочку (Play). Нажмите на неё и выберите Run 'MainKt'.

В нижней части экрана откроется окно Run (консоль IntelliJ IDEA). Вы увидите процесс компиляции, а затем результат работы вашей программы:

Console Download Manager инициализирован!
Запустите программу, указав URL файла для скачивания.

Поздравляю! Вы только что настроили окружение, написали и успешно выполнили системную консольную программу на Kotlin.

Передача аргументов в IDE

Чтобы протестировать, как программа обрабатывает аргументы командной строки, не выходя из среды разработки, нам нужно настроить конфигурацию запуска.

1. В верхней панели IDEA, рядом с зеленой кнопкой запуска, нажмите на выпадающий список MainKt и выберите Edit Configurations....
2. В поле Program arguments введите тестовые данные, разделенные пробелом: https://test.com/video.mp4 8.
3. Нажмите OK и снова запустите программу.

Теперь в консоли вы увидите:

Console Download Manager инициализирован!
Получены аргументы командной строки: https://test.com/video.mp4, 8

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

Анатомия консольного приложения

Поскольку наша цель — создать именно консольную утилиту, важно понимать, как она взаимодействует с операционной системой. Любой консольный процесс по умолчанию имеет три стандартных потока данных:

1. Standard Input (stdin) — стандартный поток ввода. Обычно это клавиатура. Если мы захотим спросить пользователя «Вы уверены, что хотите прервать загрузку? (y/n)», мы будем читать данные из stdin.
2. Standard Output (stdout) — стандартный поток вывода. Сюда отправляется обычный текст, логи и прогресс-бары. Функция println() пишет именно сюда.
3. Standard Error (stderr) — стандартный поток ошибок. Сюда принято выводить сообщения о сбоях (например, «Потеряно соединение с сервером»). Разделение stdout и stderr позволяет пользователям перенаправлять логи в разные файлы.

В Kotlin доступ к этим потокам осуществляется через объекты System. (например, System.err.println("Ошибка!")).

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

Взгляд в будущее: зачем нам многопоточность?

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

Представьте, что вы скачиваете файл размером 1000 Мегабайт. Скорость вашего интернет-канала позволяет скачивать данные со скоростью 50 Мегабайт в секунду.

Время загрузки рассчитывается по базовой формуле:

$$T = \frac{S}{V}$$

Где:

• $T$ — время загрузки в секундах.
• $S$ — размер файла (в нашем случае 1000 МБ).
• $V$ — скорость передачи данных (50 МБ/с).

В идеальных условиях $T = 1000 / 50 = 20$ секунд.

Однако в реальности сервер, с которого вы качаете файл, может искусственно ограничивать скорость отдачи для одного подключения до 10 МБ/с, чтобы не перегружать свои ресурсы. Тогда ваша загрузка займет $1000 / 10 = 100$ секунд, несмотря на то, что ваш домашний интернет способен на большее.

Здесь на сцену выходит параллельная загрузка. Если сервер поддерживает докачку файлов (запрос конкретных байтов), наш менеджер загрузок сможет открыть не одно, а, например, 5 одновременных подключений к серверу.

Каждое подключение (поток) будет скачивать свой кусок файла (по 200 МБ) со скоростью 10 МБ/с. В результате мы утилизируем всю ширину нашего домашнего канала (5 * 10 = 50 МБ/с) и скачаем файл за исходные 20 секунд.

Реализация этой логики — разделение файла на части, запуск независимых потоков, контроль их выполнения и финальная склейка кусков в единый файл — и станет ядром вашей курсовой работы. А настроенная сегодня IntelliJ IDEA и мощь языка Kotlin помогут сделать этот процесс максимально понятным и контролируемым.

Основы консольного ввода-вывода в Kotlin

В прошлой статье мы подготовили фундамент: установили среду разработки IntelliJ IDEA, настроили систему сборки Gradle и написали первую программу, которая умеет читать аргументы командной строки. Однако настоящий консольный менеджер загрузок не должен быть «глухим» к пользователю после своего запуска. Ему необходимо уметь задавать вопросы, сообщать о прогрессе и предупреждать об ошибках в реальном времени.

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

Анатомия стандартных потоков

Любое консольное приложение в современных операционных системах (будь то Windows, macOS или Linux) при запуске автоматически получает от системы три невидимых канала связи. Эти каналы называются стандартными потоками (standard streams).

Представьте себе заводской станок. У него есть приемный лоток, куда подают сырье, конвейер, откуда выходят готовые детали, и отдельная труба для сброса брака. Программа работает по тому же принципу:

1. Standard Input (stdin) — стандартный поток ввода. По умолчанию он подключен к вашей клавиатуре. Через него программа получает текстовые команды.
2. Standard Output (stdout) — стандартный поток вывода. Сюда программа отправляет результаты своей успешной работы. По умолчанию этот текст печатается на экране терминала.
3. Standard Error (stderr) — стандартный поток ошибок. Это отдельный канал для вывода предупреждений и критических сбоев.

Разделение вывода на stdout и stderr критически важно для системного программирования. Если пользователь решит сохранить логи вашего менеджера загрузок в текстовый файл (используя оператор перенаправления в консоли), он сможет направить успешные сообщения в один файл, а ошибки — в другой, чтобы потом легко найти причину сбоя.

Вывод данных: общение с пользователем

В Kotlin для отправки текста в стандартный поток вывода (stdout) используются две основные функции, унаследованные от стандартной библиотеки.

Функции print() и println()

Функция print() выводит переданный ей текст в консоль, оставляя курсор на той же строке. Следующий вызов print() продолжит печатать текст ровно с того места, где остановился предыдущий.

Функция println() (от английского print line) делает то же самое, но в конце автоматически добавляет невидимый символ перевода строки. Следующий вывод начнется с новой строки.

fun main() {
    print("Загрузка файла: ")
    print("ubuntu-22.04.iso ")
    println("[Ожидание]")
    println("Подключение к серверу...")
}

Результат в консоли:

Загрузка файла: ubuntu-22.04.iso [Ожидание]
Подключение к серверу...

В контексте нашего менеджера загрузок print() будет незаменим для отрисовки прогресс-бара. Мы сможем обновлять проценты на одной строке, не загромождая экран сотнями новых строк.

Вывод ошибок через System.err

Если во время скачивания пропадет интернет, мы должны сообщить об этом пользователю. Использовать обычный println() для этого — плохая практика, так как сообщение уйдет в общий поток stdout.

Для отправки текста в поток stderr в Kotlin используется обращение к базовому классу Java — System.err:

fun main() {
    println("Начинаю загрузку...")
    // Имитация сбоя сети
    System.err.println("Ошибка: Превышено время ожидания ответа от сервера (Timeout).")
}

Если вы запустите этот код в IntelliJ IDEA, вы заметите, что текст ошибки будет подсвечен красным цветом. Это среда разработки визуально разделяет для вас два разных потока.

Ввод данных: чтение строк

Теперь научим программу слушать пользователя. Для чтения данных из стандартного потока ввода (stdin) в Kotlin применяется функция readln().

Когда программа доходит до вызова readln(), она приостанавливает свое выполнение (блокирует текущий поток) и ждет, пока пользователь наберет текст на клавиатуре и нажмет клавишу Enter.

fun main() {
    print("Введите URL файла для скачивания: ")
    val fileUrl = readln()
    println("Добавлено в очередь: $fileUrl")
}

Проблема неожиданного конца файла (EOF)

В системном программировании нужно быть готовым к нестандартному поведению. Что если пользователь не введет текст, а принудительно закроет поток ввода комбинацией клавиш Ctrl+D (в Linux/macOS) или Ctrl+Z (в Windows)? Это называется сигналом EOF (End of File).

Если в этот момент программа ожидает ввода через readln(), она аварийно завершится с ошибкой RuntimeException. Чтобы сделать наш менеджер загрузок отказоустойчивым, лучше использовать безопасную альтернативу — readlnOrNull().

Эта функция возвращает введенную строку, а если поток был неожиданно закрыт — возвращает специальное значение null (пустоту), не ломая программу.

fun main() {
    print("Введите URL: ")
    val input = readlnOrNull()

    if (input == null) {
        System.err.println("\nВвод прерван пользователем. Завершение работы.")
        return
    }

    println("Вы ввели: $input")
}

Безопасность работы с null (Null Safety) — одна из главных особенностей языка Kotlin. Компилятор физически не позволит вам вызвать методы у переменной, которая может оказаться пустой, пока вы явно не проверите её на null.

Продвинутый парсинг: класс Scanner

Функции readln() всегда возвращают текст (тип String). Но что, если нам нужно спросить у пользователя желаемое количество потоков для параллельной загрузки? Нам нужно число (тип Int).

Мы можем прочитать строку и попытаться конвертировать её в число с помощью встроенного метода .toInt(). Но есть инструмент, специально созданный для разбора (парсинга) ввода — класс Scanner из стандартной библиотеки Java.

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

import java.util.Scanner

fun main() {
    val scanner = Scanner(System.`in`)

    print("Введите количество потоков (от 1 до 16): ")

    if (scanner.hasNextInt()) {
        val threads = scanner.nextInt()
        println("Менеджер загрузок будет использовать $threads потоков.")
    } else {
        System.err.println("Ошибка: Вы ввели не число!")
    }
}

Обратите внимание на конструкцию System.in``. Слово in в Kotlin является зарезервированным (используется в циклах), поэтому при обращении к Java-полю System.in мы обязаны брать его в обратные кавычки (backticks).

Метод hasNextInt() предварительно проверяет, является ли следующее введенное слово целым числом. Если пользователь введет «пять» вместо «5», программа не упадет с ошибкой InputMismatchException, а корректно обработает ситуацию.

Математика прогресса загрузки

Поскольку мы учимся выводить данные в консоль, давайте рассмотрим, как рассчитывается информация, которую мы будем показывать пользователю во время скачивания.

Обычно менеджер загрузок отображает три параметра: процент завершения, текущую скорость и оставшееся время.

Формула для расчета оставшегося времени выглядит так:

$$T_{ost} = \frac{S_{total} - S_{current}}{V}$$

Где:

• $T_{ost}$ — оставшееся время в секундах.
• $S_{total}$ — общий размер файла в байтах (мы узнаем его от сервера перед началом скачивания).
• $S_{current}$ — количество уже скачанных байт на данный момент.
• $V$ — текущая скорость скачивания в байтах в секунду.

Пример из жизни: вы качаете образ операционной системы размером 2 000 000 байт ($S_{total}$). Вы уже скачали 500 000 байт ($S_{current}$). Ваша текущая скорость — 100 000 байт в секунду ($V$).

Подставляем в формулу: $T_{ost} = \frac{2000000 - 500000}{100000} = \frac{1500000}{100000} = 15$ секунд.

Именно эти 15 секунд мы будем форматировать в красивую строку 00:00:15 и выводить в консоль с помощью print().

Интерактивный режим vs Аргументы запуска

В прошлой статье мы передавали ссылку на файл через аргументы командной строки (args). Сегодня мы научились запрашивать её интерактивно через readln(). Какой подход правильный?

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

Подход	Как работает	Когда удобнее использовать
Аргументы (args)	java -jar app.jar http://file.zip 4	Для автоматизации. Когда ваш менеджер загрузок запускается другим скриптом без участия человека.
Интерактивный ввод	Программа запускается пустой и спрашивает: «Что качаем?»	Для ручного использования обычным пользователем.

Логика нашего будущего менеджера загрузок будет следующей: при запуске мы проверяем массив args. Если он не пуст — сразу начинаем качать. Если пуст — переходим в интерактивный режим и просим пользователя ввести данные через readln().

Важная настройка: Gradle и поток ввода

Если вы прямо сейчас скопируете код с readln() в свой проект в IntelliJ IDEA и нажмете зеленую кнопку запуска (Run), вы столкнетесь с неприятной проблемой. Программа зависнет, но вы не сможете ничего ввести в консоль.

Это происходит из-за системы сборки Gradle, которую мы выбрали при создании проекта. По умолчанию Gradle запускает вашу программу в изолированном фоновом режиме и отключает передачу стандартного потока ввода от вашей клавиатуры к программе.

Чтобы починить это, нам нужно явно сказать Gradle: «Разреши моей программе читать данные с клавиатуры».

1. Откройте файл build.gradle.kts в корне вашего проекта.
2. Добавьте в самый конец файла следующий блок кода:

tasks.withType<JavaExec> {
    standardInput = System.`in`
}

3. Нажмите появившуюся иконку слона со стрелочками (Sync Project with Gradle Files) в правом верхнем углу редактора.

Теперь Gradle свяжет консоль IntelliJ IDEA с потоком stdin вашего приложения, и функции readln() и Scanner заработают корректно.

Создание интерактивного меню

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

fun main() {
    println("=== Console Download Manager ===")

    while (true) {
        print("\nВведите команду (download, status, exit): ")
        val command = readlnOrNull()?.trim()?.lowercase()

        when (command) {
            "download" -> {
                print("Введите URL: ")
                val url = readln()
                println("Начинаю загрузку $url...")
                // Здесь в будущем будет вызов функции скачивания
            }
            "status" -> {
                println("Активных загрузок нет.")
            }
            "exit" -> {
                println("Завершение работы. До свидания!")
                break // Выход из бесконечного цикла
            }
            null -> {
                System.err.println("\nПоток ввода закрыт. Аварийное завершение.")
                break
            }
            else -> {
                System.err.println("Неизвестная команда. Попробуйте снова.")
            }
        }
    }
}

В этом примере мы используем цикл while (true) для постоянного опроса пользователя. Конструкция when (аналог switch в других языках) элегантно маршрутизирует логику в зависимости от введенного слова.

Методы .trim().lowercase() очищают ввод от случайных пробелов по краям и переводят текст в нижний регистр, чтобы команды EXIT, Exit и exit воспринимались одинаково.

Освоив стандартные потоки ввода и вывода, мы научили нашу программу общаться с внешним миром. Теперь у нас есть полноценный интерфейс командной строки. В следующей статье мы перейдем к самому интересному — работе с сетью, и научим наш менеджер делать реальные HTTP-запросы к серверам для получения информации о файлах.

Работа с файловой системой и потоками данных

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

Сегодня мы разберем, как операционная система управляет файлами, почему нельзя просто скачать фильм в оперативную память и как подготовить фундамент для параллельной загрузки данных, к которой мы стремимся.

Файловая система: взгляд изнутри

Файловая система (ФС) — это подсистема операционной системы, которая организует хранение данных на физических носителях (жестких дисках, SSD, флешках).

На физическом уровне любой накопитель — это просто огромное, неструктурированное пространство для записи нулей и единиц. Без файловой системы операционная система не знала бы, где начинается один файл и заканчивается другой.

Чтобы программам было удобно работать, файловая система предоставляет абстракцию — древовидную структуру. Единицей организации является каталог (папка), который может содержать другие каталоги и файлы. Самый верхний каталог называется корневым.

Абсолютные и относительные пути

Чтобы обратиться к файлу, программе нужно указать его адрес — путь. Существует два способа это сделать:

1. Абсолютный путь — это полный адрес файла, начиная от самого корня файловой системы. В Windows он начинается с буквы диска (например, C:\Downloads\ubuntu.iso), а в Linux и macOS — с корневого слеша (например, /home/user/downloads/ubuntu.iso).
2. Относительный путь — это адрес файла относительно текущего рабочего каталога (места, откуда была запущена ваша программа).

Текущий рабочий каталог — это концепция, позволяющая сократить пути. Если ваша программа запущена из папки /home/user/app/, и вы просите её создать файл log.txt, он будет создан по абсолютному пути /home/user/app/log.txt.

В Kotlin для навигации по относительным путям используются специальные символы:

• . (одна точка) — текущий каталог.
• .. (две точки) — родительский каталог (на один уровень вверх).

Инструменты Kotlin для работы с файлами

Исторически в экосистеме Java (и, следовательно, в Kotlin) существует два основных подхода к работе с файловой системой:

1. Класс java.io.File — старый, базовый подход, появившийся в первой версии Java.
2. Пакет java.nio.file (New I/O) и интерфейс Path — современный подход, который лучше обрабатывает ошибки, поддерживает символические ссылки и работает быстрее.

Для нашего менеджера загрузок мы будем использовать современный подход NIO, но с удобными функциями-расширениями из стандартной библиотеки Kotlin.

Давайте напишем код, который проверяет, существует ли папка для загрузок, и если нет — создает её:

import java.nio.file.Path
import kotlin.io.path.Path
import kotlin.io.path.createDirectories
import kotlin.io.path.exists

fun main() {
    // Создаем объект Path, указывающий на папку downloads в текущем каталоге
    val downloadDir: Path = Path("downloads")

    if (!downloadDir.exists()) {
        println("Папка для загрузок не найдена. Создаю...")
        downloadDir.createDirectories()
        println("Папка успешно создана по пути: ${downloadDir.toAbsolutePath()}")
    } else {
        println("Папка для загрузок уже существует.")
    }
}

Обратите внимание: функция Path("downloads") не создает папку на диске физически. Она лишь создает в оперативной памяти объект, который описывает этот путь. Физическое создание происходит только при вызове метода createDirectories().

Потоки данных: концепция I/O

Когда мы говорим о чтении или записи содержимого файла, мы сталкиваемся с понятием потока данных (Data Stream).

Представьте себе водопроводную трубу. Вы не можете мгновенно телепортировать всю воду из водохранилища в свой стакан. Вода течет непрерывным потоком, капля за каплей. Точно так же работают потоки ввода-вывода (I/O — Input/Output) в программировании.

Существует два базовых типа потоков:

• InputStream (Поток ввода) — труба, по которой байты «втекают» в вашу программу (из файла, из сети, с клавиатуры).
• OutputStream (Поток вывода) — труба, по которой байты «вытекают» из вашей программы (в файл, на экран, в сеть).

Проблема оперативной памяти

Почему мы обязаны использовать потоки? Представьте, что пользователь решил скачать образ игры размером 50 ГБ.

Если бы мы попытались сначала скачать весь файл целиком в оперативную память (RAM), а только потом сохранить его на диск, наша программа немедленно завершилась бы с критической ошибкой OutOfMemoryError. Оперативная память ограничена и стоит дорого.

Потоковый подход решает эту проблему элегантно: мы скачиваем небольшой кусочек файла (например, 8 Килобайт), сразу записываем его на жесткий диск, освобождаем память и скачиваем следующий кусочек. Таким образом, для скачивания файла любого размера нашей программе потребуется всего несколько килобайт оперативной памяти.

Буферизация: оптимизация системных вызовов

Запись данных на физический диск — это очень медленная операция по меркам процессора. Каждый раз, когда ваша программа просит операционную систему записать байт на диск, происходит системный вызов (System Call). Процессор переключает контекст, обращается к ядру ОС, ядро обращается к драйверу диска, драйвер — к контроллеру.

Если мы будем записывать файл по одному байту, программа будет работать невыносимо медленно.

Аналогия из жизни: вам нужно перенести 1000 кирпичей из грузовика на стройку. Вы можете брать по одному кирпичу и ходить туда-сюда 1000 раз (запись по одному байту). А можете взять тачку, положить в нее 50 кирпичей и сделать всего 20 рейсов.

В программировании такая «тачка» называется буфером.

Буфер — это небольшой участок оперативной памяти. Программа быстро пишет данные в этот буфер. Когда буфер заполняется, его содержимое разом сбрасывается (записывается) на жесткий диск одним системным вызовом.

В Kotlin для этого используется класс BufferedOutputStream.

import java.io.BufferedOutputStream
import java.io.FileOutputStream
import java.io.File

fun main() {
    val file = File("downloads/test.txt")

    // Создаем поток вывода в файл и оборачиваем его в буфер
    val outputStream = BufferedOutputStream(FileOutputStream(file))

    val text = "Этот текст будет записан через буфер."
    val bytes = text.toByteArray() // Преобразуем строку в массив байт

    outputStream.write(bytes)

    // ВАЖНО: Принудительно сбрасываем остатки буфера на диск
    outputStream.flush()

    // Закрываем поток, освобождая ресурсы ОС
    outputStream.close()

    println("Данные успешно записаны.")
}

Метод flush() критически важен. Если буфер рассчитан на 8192 байта, а вы записали только 100 байт, они останутся висеть в оперативной памяти. Метод flush() приказывает: «Не жди заполнения, сбрось данные на диск прямо сейчас».

Произвольный доступ: ключ к многопоточности

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

Но наша цель — создать менеджер загрузок с параллельной загрузкой (как ab-download-manager или Internet Download Manager).

Как работает параллельная загрузка?

1. Программа узнает у сервера общий размер файла (например, 100 МБ).
2. Программа делит этот размер на количество потоков (например, 4 части по 25 МБ).
3. Четыре независимых потока одновременно скачивают свои куски.
4. Первый поток пишет данные с начала файла, второй — начиная с 25-го мегабайта, третий — с 50-го, четвертый — с 75-го.

Чтобы реализовать такую архитектуру, нам нужен инструмент, позволяющий писать данные в любую точку файла. В Java/Kotlin для этого используется класс RandomAccessFile (Файл произвольного доступа).

Класс RandomAccessFile и метод seek()

RandomAccessFile позволяет перемещать внутренний «курсор» (указатель) файла в любую позицию перед чтением или записью. Это делается с помощью метода seek(позиция_в_байтах).

Давайте посмотрим, как это работает на практике. Мы создадим файл и запишем слово «Привет» в самое начало, а слово «Мир» — сместив курсор на 100 байт вперед.

import java.io.RandomAccessFile
import java.io.File

fun main() {
    val file = File("downloads/parallel_test.dat")

    // Открываем файл в режиме "rw" (Read/Write - чтение и запись)
    val raf = RandomAccessFile(file, "rw")

    // Пишем в начало (позиция 0)
    raf.write("Привет".toByteArray())

    // Перемещаем курсор на 100-й байт
    raf.seek(100)

    // Пишем второе слово
    raf.write("Мир".toByteArray())

    raf.close()
    println("Файл с произвольным доступом создан. Размер: ${file.length()} байт.")
}

Если вы откроете созданный файл parallel_test.dat в текстовом редакторе, вы увидите слово «Привет», затем много пустого места (нулевых байтов), а затем слово «Мир». Размер файла составит 103 байта (100 байт смещения + 3 байта на слово «Мир» в кодировке UTF-8).

Предварительное выделение места (Pre-allocation)

Профессиональные менеджеры загрузок делают еще одну важную вещь перед началом скачивания: они резервируют место на диске.

Если вы качаете файл на 10 ГБ, программа сразу создает на диске пустой файл размером 10 ГБ. Зачем это нужно?

1. Гарантия места: Если на диске осталось только 5 ГБ, программа выдаст ошибку сразу, а не на середине скачивания, когда вы уже потратили время и трафик.
2. Борьба с фрагментацией: Если файл растет постепенно, операционная система может разбросать его куски по разным физическим секторам диска. Если выделить место сразу, ОС постарается найти непрерывный блок, что ускорит последующее чтение файла.

С помощью RandomAccessFile зарезервировать место очень просто. Достаточно установить длину файла методом setLength():

import java.io.RandomAccessFile
import java.io.File

fun main() {
    val file = File("downloads/movie.mp4")
    val raf = RandomAccessFile(file, "rw")

    // Требуемый размер: 1 Гигабайт
    val requiredSize: Long = 1024L * 1024L * 1024L

    println("Резервирование места на диске...")
    raf.setLength(requiredSize)

    raf.close()
    println("Успешно! Создан пустой файл размером 1 ГБ.")
}

Обратите внимание на суффикс L в числах 1024L. Он указывает компилятору Kotlin, что мы работаем с типом Long (64-битное целое число). Если использовать обычный Int (32-битное число), при умножении больших чисел произойдет переполнение, и мы получим отрицательный или некорректный результат.

Математика смещений при параллельной загрузке

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

Допустим, мы скачиваем файл размером $S = 1000$ байт. Мы хотим использовать $N = 4$ потока.

Размер одной части (чанка), которую должен скачать каждый поток, вычисляется по формуле:

$$C = \frac{S}{N}$$

Подставляем значения: $C = \frac{1000}{4} = 250$ байт.

Теперь нам нужно рассчитать стартовую позицию (смещение) для каждого потока. Формула для стартовой позиции потока с индексом $i$ (где индексы начинаются с 0):

$$P_i = i \times C$$

Считаем для каждого потока:

• Поток 0: $P_0 = 0 \times 250 = 0$ (начнет писать с самого начала)
• Поток 1: $P_1 = 1 \times 250 = 250$ (начнет писать с 250-го байта)
• Поток 2: $P_2 = 2 \times 250 = 500$ (начнет писать с 500-го байта)
• Поток 3: $P_3 = 3 \times 250 = 750$ (начнет писать с 750-го байта)

Именно эти значения $P_i$ мы будем передавать в метод raf.seek(P_i) внутри каждого отдельного потока выполнения.

Продвинутый уровень: FileChannel и NIO

Хотя RandomAccessFile отлично справляется со своей задачей, в современном системном программировании на Kotlin/Java для высоконагруженных приложений (каким является менеджер загрузок) предпочтительнее использовать FileChannel из пакета java.nio.channels.

FileChannel работает быстрее, так как может взаимодействовать с кэшем операционной системы напрямую и поддерживает потокобезопасную запись (несколько потоков могут писать в один канал одновременно, если пишут в разные участки файла).

Пример получения канала из RandomAccessFile:

import java.io.RandomAccessFile
import java.nio.ByteBuffer
import java.io.File

fun main() {
    val file = File("downloads/nio_test.dat")
    val raf = RandomAccessFile(file, "rw")

    // Получаем канал для работы с файлом
    val channel = raf.channel

    // В NIO мы работаем не с массивами байт, а с буферами (ByteBuffer)
    val text = "Запись через FileChannel"
    val buffer = ByteBuffer.wrap(text.toByteArray())

    // Пишем данные по смещению 500 байт, не меняя глобальный курсор файла
    channel.write(buffer, 500L)

    channel.close()
    raf.close()
}

Главное преимущество метода channel.write(buffer, position) заключается в том, что он атомарен относительно позиции. Нам не нужно сначала вызывать seek(), а потом write(). Мы сразу говорим: «Запиши этот буфер по этому смещению». Это избавляет нас от множества проблем при реальном многопоточном программировании.

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

Выполнение базовых HTTP-запросов

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

Теперь настало время дать нашей программе выход в интернет. Главная задача любого менеджера загрузок — запрашивать данные у удаленных серверов и сохранять их локально. Для этого нам необходимо понять, как компьютеры общаются между собой в глобальной сети, и научиться использовать инструменты Kotlin для выполнения сетевых запросов.

Протокол HTTP: язык общения в интернете

Интернет работает на основе архитектуры Клиент-Сервер. Ваша программа (менеджер загрузок) выступает в роли клиента. Она отправляет запрос. Компьютер, на котором хранится нужный файл (например, сервер Ubuntu или GitHub), выступает в роли сервера. Он обрабатывает запрос и возвращает ответ.

Чтобы клиент и сервер понимали друг друга, они должны говорить на одном языке. В веб-технологиях таким языком является HTTP (HyperText Transfer Protocol — протокол передачи гипертекста).

Каждый HTTP-запрос состоит из трех главных частей:

1. Метод — действие, которое мы хотим выполнить (например, получить данные или отправить их).
2. URL — адрес ресурса (ссылка на файл).
3. Заголовки (Headers) — служебная метаинформация (например, какой браузер делает запрос или какой формат данных ожидается).

Методы GET и HEAD

Для разработки менеджера загрузок нам критически важны два HTTP-метода:

• GET — самый популярный метод. Он означает: «Сервер, отдай мне содержимое этого файла». Когда вы вводите адрес в браузере, браузер отправляет именно GET-запрос. В ответ сервер присылает и служебную информацию, и сами байты файла.
• HEAD — это метод-разведчик. Он означает: «Сервер, пришли мне только служебную информацию о файле, но сами данные не отправляй».

Зачем нужен метод HEAD? Представьте, что пользователь хочет скачать файл по ссылке. Прежде чем начать скачивание (GET), профессиональный менеджер загрузок отправляет HEAD-запрос.

Это позволяет мгновенно узнать:

• Существует ли файл вообще (или сервер вернет ошибку 404).
• Каков точный размер файла в байтах (чтобы зарезервировать место на диске, как мы учились ранее).
• Поддерживает ли сервер докачку и параллельную загрузку.

Выбор инструмента: знакомство с Ktor

В экосистеме Java и Kotlin существует множество библиотек для работы с сетью: HttpURLConnection (встроенная, но устаревшая), OkHttp (стандарт индустрии для Android), Retrofit (отлично подходит для REST API).

Однако для нашего проекта мы выберем Ktor.

Ktor — это современный, асинхронный фреймворк, созданный компанией JetBrains (разработчиками языка Kotlin).

Его главные преимущества для системного программирования:

• Асинхронность из коробки: Ktor построен на базе корутин (сопрограмм) Kotlin. Это значит, что сетевые запросы не будут блокировать основной поток выполнения программы.
• Модульность: Вы подключаете только те функции, которые вам нужны. Программа остается легкой и быстрой.
• Кроссплатформенность: Код, написанный на Ktor, можно скомпилировать не только под JVM, но и под Native (исполняемый файл .exe или .elf без необходимости устанавливать Java).

Настройка зависимостей Gradle

Чтобы добавить Ktor в наш проект, необходимо отредактировать файл build.gradle.kts. Откройте его и добавьте следующие строки в блок dependencies:

dependencies {
    // Ядро клиента Ktor
    implementation("io.ktor:ktor-client-core:2.3.7")
    // Движок CIO для выполнения запросов
    implementation("io.ktor:ktor-client-cio:2.3.7")
}

После изменения файла build.gradle.kts не забудьте нажать кнопку Sync Now (Синхронизировать) в правом верхнем углу IntelliJ IDEA, чтобы система сборки скачала нужные библиотеки из интернета.

Обратите внимание на зависимость ktor-client-cio. Ktor спроектирован так, что его ядро ничего не знает о том, как физически отправлять байты по сети. Для этого ему нужен «движок» (Engine).

CIO (Coroutine-based I/O) — это встроенный асинхронный движок Ktor, который идеально подходит для консольных приложений и менеджеров загрузок, так как он отлично справляется с большим количеством одновременных соединений.

Выполнение первого запроса

Давайте напишем код, который обращается к серверу и запрашивает информацию о файле. Для тестов мы будем использовать специальный сервис httpbin.org, который возвращает отправленные ему данные.

import io.ktor.client.*
import io.ktor.client.engine.cio.*
import io.ktor.client.request.*
import io.ktor.client.statement.*
import io.ktor.http.*
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
    // 1. Создаем экземпляр HTTP-клиента с движком CIO
    val client = HttpClient(CIO)

    val fileUrl = "https://httpbin.org/get"

    try {
        println("Отправляем запрос на сервер...")

        // 2. Выполняем GET-запрос.
        // Функция get() является suspend-функцией, поэтому мы используем runBlocking
        val response: HttpResponse = client.get(fileUrl)

        // 3. Анализируем статус ответа
        println("Статус ответа: ${response.status}")

        if (response.status.isSuccess()) {
            println("Успешное подключение!")
        } else {
            println("Ошибка сервера: ${response.status.description}")
        }

    } catch (e: Exception) {
        println("Сетевая ошибка: ${e.message}")
    } finally {
        // 4. Обязательно закрываем клиент, чтобы освободить системные ресурсы
        client.close()
    }
}

Разбор кода: корутины и блокировка

Вы могли заметить конструкцию runBlocking { ... }, оборачивающую тело функции main. Зачем она нужна?

Сетевой запрос — это невероятно медленная операция по меркам процессора. Отправка сигнала на сервер в другую страну и ожидание ответа может занять 100-200 миллисекунд. За это время современный процессор мог бы выполнить миллионы математических операций.

Функция client.get() в Ktor помечена ключевым словом suspend (приостанавливаемая). Это означает, что она не блокирует поток жестко. Она говорит операционной системе: «Я жду ответа от сети. Пока я жду, ты можешь использовать этот поток процессора для других полезных задач».

Однако функция main в Kotlin не может быть suspend по умолчанию в старых версиях, и программа завершилась бы до того, как придет ответ. Блок runBlocking создает мост между обычным (синхронным) кодом и асинхронным миром корутин. Он заставляет главный поток программы дождаться завершения всех сетевых операций внутри блока.

Чтение HTTP-заголовков: разведка перед скачиванием

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

Давайте напишем функцию, которая выполняет HEAD-запрос к реальному файлу и извлекает нужные метаданные.

import io.ktor.client.*
import io.ktor.client.engine.cio.*
import io.ktor.client.request.*
import io.ktor.client.statement.*
import io.ktor.http.*
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
    val client = HttpClient(CIO)
    // Ссылка на тестовый файл (дистрибутив Linux)
    val url = "https://releases.ubuntu.com/22.04.3/ubuntu-22.04.3-desktop-amd64.iso"

    try {
        // Используем метод HEAD вместо GET
        val response: HttpResponse = client.head(url)

        if (response.status.isSuccess()) {
            // Извлекаем заголовок Content-Length (Размер файла)
            val contentLengthStr = response.headers[HttpHeaders.ContentLength]
            val fileSize: Long = contentLengthStr?.toLong() ?: 0L

            // Извлекаем заголовок Accept-Ranges (Поддержка докачки)
            val acceptRanges = response.headers[HttpHeaders.AcceptRanges]
            val supportsResume = acceptRanges == "bytes"

            println("Информация о файле:")
            println("Размер: ${fileSize / (1024 * 1024)} Мегабайт")
            println("Поддержка параллельной загрузки: ${if (supportsResume) "Да" else "Нет"}")
        }
    } finally {
        client.close()
    }
}

Ключевые заголовки для менеджера загрузок

1. Content-Length — указывает точный размер файла в байтах. Если сервер не прислал этот заголовок (такое бывает при динамической генерации файлов), мы не сможем заранее выделить место на диске и не сможем разделить файл на части для параллельной загрузки. Нам придется качать его в один поток до тех пор, пока соединение не закроется.
2. Accept-Ranges: bytes — этот заголовок — зеленый свет для многопоточности. Он означает, что сервер умеет отдавать файл по кускам. Если мы попросим: «Дай мне байты с 5000-го по 10000-й», сервер нас поймет и выполнит просьбу.

Последовательное скачивание файла (Потоковая передача)

Теперь, когда мы умеем получать информацию о файле, давайте его скачаем.

В прошлой статье мы выяснили, что скачивать большие файлы целиком в оперативную память (RAM) нельзя — программа упадет с ошибкой OutOfMemoryError. Нам нужно скачивать данные небольшими порциями (чанками) и сразу записывать их на диск.

В Ktor для потокового чтения тела ответа используется метод bodyAsChannel(). Он возвращает объект ByteReadChannel, из которого можно выкачивать байты порциями.

Давайте объединим наши знания о файловой системе и HTTP-запросах в единый рабочий механизм:

import io.ktor.client.*
import io.ktor.client.engine.cio.*
import io.ktor.client.request.*
import io.ktor.client.statement.*
import io.ktor.http.*
import io.ktor.utils.io.*
import io.ktor.utils.io.core.*
import kotlinx.coroutines.runBlocking
import java.io.File

fun main() = runBlocking {
    val client = HttpClient(CIO)
    // Небольшой тестовый файл (изображение)
    val url = "https://via.placeholder.com/5000"
    val outputFile = File("downloads/image.png")

    // Создаем директорию, если ее нет
    outputFile.parentFile?.mkdirs()

    try {
        println("Начинаем загрузку...")

        // Выполняем GET-запрос.
        // Важно: используем prepareGet и execute для потоковой обработки,
        // чтобы Ktor не пытался загрузить весь ответ в память автоматически.
        client.prepareGet(url).execute { response ->
            if (!response.status.isSuccess()) {
                println("Ошибка: сервер вернул код ${response.status}")
                return@execute
            }

            // Получаем канал для чтения байтов
            val channel: ByteReadChannel = response.bodyAsChannel()

            // Открываем поток для записи в файл
            val outputStream = outputFile.outputStream()

            // Буфер в оперативной памяти (8 Килобайт)
            val bufferSize = 8192
            val buffer = ByteArray(bufferSize)
            var totalBytesDownloaded = 0L

            // Цикл чтения данных
            while (!channel.isClosedForRead) {
                // Читаем порцию данных из сети в буфер
                val bytesRead = channel.readAvailable(buffer, 0, bufferSize)

                if (bytesRead > 0) {
                    // Записываем прочитанные байты на жесткий диск
                    outputStream.write(buffer, 0, bytesRead)
                    totalBytesDownloaded += bytesRead
                    print("\rСкачано: ${totalBytesDownloaded / 1024} KB")
                }
            }

            // Обязательно закрываем файловый поток
            outputStream.flush()
            outputStream.close()

            println("\nЗагрузка успешно завершена!")
        }
    } catch (e: Exception) {
        println("Ошибка при скачивании: ${e.message}")
    } finally {
        client.close()
    }
}

Математика буферизации

Давайте разберем, что происходит в цикле while. Допустим, размер файла $S = 1048576$ байт (1 Мегабайт). Размер нашего буфера $C = 8192$ байта (8 Килобайт).

Количество итераций цикла (сколько раз программа обратится к диску для записи), вычисляется по формуле:

$$I = \lceil \frac{S}{C} \rceil$$

Где $\lceil x \rceil$ — округление в большую сторону. Подставляем значения: $I = \lceil \frac{1048576}{8192} \rceil = 128$ итераций.

Если бы мы не использовали буфер и читали по 1 байту ($C = 1$), программе пришлось бы выполнить более миллиона операций записи на диск, что замедлило бы скачивание в десятки раз из-за накладных расходов операционной системы на системные вызовы.

Метод prepareGet() vs get()

В коде выше мы использовали конструкцию client.prepareGet(url).execute { ... } вместо обычного client.get(url). Это критически важное архитектурное решение для менеджера загрузок.

Обычный метод get() пытается быть удобным: он скачивает весь ответ сервера в оперативную память и только потом возвращает вам объект HttpResponse. Для JSON-ответов весом в пару килобайт это отлично. Но для фильма весом 10 ГБ это приведет к краху программы.

Метод prepareGet() создает HTTP-запрос, отправляет его, получает заголовки ответа и останавливается. Он не скачивает тело ответа (сами байты файла), пока вы явно не начнете читать их из канала ByteReadChannel внутри блока execute. Как только блок execute завершается, сетевое соединение безопасно закрывается.

Мы успешно реализовали базовый сетевой слой нашего приложения. Программа умеет проводить «разведку» с помощью HEAD-запросов, анализировать заголовки для определения размера файла и безопасно скачивать данные потоковым методом, не перегружая оперативную память. Этот последовательный алгоритм — идеальная база. В следующем этапе мы применим концепцию корутин на полную мощность, разделим файл на части и запустим несколько таких потоковых загрузок одновременно.

Введение в многопоточность и конкурентность

Наш консольный менеджер загрузок уже умеет выполнять HTTP-запросы, анализировать заголовки сервера и скачивать файлы последовательно, сохраняя их на диск небольшими порциями (чанками). Это надежный, но медленный подход. Если сервер искусственно ограничивает скорость отдачи данных для одного соединения, скачивание большого файла займет часы.

Чтобы обойти это ограничение и максимально утилизировать пропускную способность вашего интернет-канала, программа должна скачивать разные части одного файла одновременно. Для реализации этой задачи нам необходимо погрузиться в системное программирование и разобраться, как операционная система и процессор управляют выполнением кода.

Процессы и потоки: анатомия выполнения программы

Прежде чем писать многопоточный код на Kotlin, необходимо понять базовые абстракции операционной системы (ОС). Любая запущенная программа в компьютере существует в виде процесса.

Процесс (Process) — это изолированная среда выполнения программы. Когда вы запускаете скомпилированный .jar файл вашего менеджера загрузок, ОС выделяет ему собственное виртуальное адресное пространство (оперативную память), файловые дескрипторы и другие системные ресурсы. Процессы строго изолированы друг от друга: процесс браузера не может напрямую прочитать память процесса вашего менеджера загрузок. Это обеспечивает безопасность и стабильность системы.

Внутри каждого процесса существует как минимум один поток выполнения (Thread), который часто называют главным потоком (Main Thread).

Поток — это наименьшая единица обработки, исполнение которой может быть назначено ядру процессора операционной системой. Потоки живут внутри процесса и делят между собой его ресурсы (например, общую память).

Поток — это отдельный работник, выполняющий задачу в процессе. Иными словами, поток есть то, что он делает.

nuancesprog.ru

Чтобы лучше понять эту концепцию, представим завод по сборке автомобилей. Завод со всеми его станками, складами и электричеством — это процесс. Рабочие на конвейере — это потоки. Если на заводе работает только один человек (однопоточная программа), он должен сам принести детали, собрать двигатель, установить колеса и покрасить кузов. Это работает, но крайне неэффективно. Если мы наймем больше рабочих (создадим новые потоки), они смогут использовать одни и те же инструменты (общую память процесса) для одновременной сборки разных частей автомобиля.

Конкурентность против Параллелизма

В системном программировании термины конкурентность (Concurrency) и параллелизм (Parallelism) часто путают, хотя они описывают принципиально разные концепции.

Конкурентность — это способ структурирования программы, при котором несколько задач выполняются в перекрывающиеся периоды времени. Это не означает, что они выполняются ровно в одну и ту же миллисекунду.

Представьте баристу в кофейне, который делает кофе для двух клиентов. Он запускает эспрессо-машину для первого клиента, и пока кофе варится (операция ожидания), он поворачивается и начинает взбивать молоко для второго клиента. Бариста один (одно ядро процессора), но он жонглирует задачами, не простаивая в ожидании. Это конкурентность.

Параллелизм — это физическое одновременное выполнение нескольких задач.

Если в кофейне работают два баристы за двумя разными кофемашинами, и они одновременно готовят два напитка — это параллелизм. Для параллелизма аппаратно необходимо наличие нескольких вычислительных ядер.

Современные операционные системы используют планировщик задач (Scheduler), который невероятно быстро переключает ядра процессора между тысячами потоков. Это создает иллюзию параллелизма даже на одноядерном процессоре (конкурентность). Когда же потоки физически распределяются по разным ядрам многоядерного процессора, мы получаем истинный параллелизм.

Проблема традиционных потоков ОС

В классическом Java/Kotlin программировании для выполнения фоновой задачи можно создать новый поток операционной системы с помощью класса Thread.

Казалось бы, для параллельной загрузки файла достаточно разделить его на 1000 частей и создать 1000 потоков. Однако этот подход приведет к катастрофическому падению производительности или аварийному завершению программы. У потоков ОС есть два критических недостатка:

1. Высокое потребление памяти. При создании каждого нового потока ОС выделяет для него отдельный блок памяти под стек вызовов (Call Stack). По умолчанию в JVM размер стека одного потока составляет около 1 Мегабайта.
2. Накладные расходы на переключение контекста. Когда планировщик ОС переключает ядро процессора с одного потока на другой, он должен сохранить текущее состояние регистров процессора в память и загрузить состояние нового потока. Это ресурсоемкая операция.

Давайте посчитаем потребление памяти по простой формуле:

$$M = N \times S$$

Где $M$ — общий объем потребляемой оперативной памяти, $N$ — количество созданных потоков, а $S$ — размер стека одного потока. Если мы создадим $N = 1000$ потоков, то $M = 1000 \times 1 \text{ MB} = 1000 \text{ MB}$ (почти 1 Гигабайт) оперативной памяти будет потрачено исключительно на поддержание самих потоков, еще до того, как программа скачает хотя бы один байт полезных данных.

Кроме того, при сетевых запросах (как в нашем менеджере загрузок) потоки большую часть времени ничего не делают. Они отправляют запрос на сервер и блокируются (Blocking), ожидая ответа по сети. Держать тяжеловесный поток ОС заблокированным — это непозволительная трата ресурсов.

Корутины: легковесные потоки Kotlin

Для решения проблемы блокировок и потребления памяти инженеры JetBrains внедрили в Kotlin концепцию корутин (Coroutines, сопрограммы).

Корутины — это абстракция над потоками. Их часто называют «легковесными потоками». Главное отличие корутины от потока ОС заключается в том, что корутина может быть приостановлена (Suspended) без блокировки потока, в котором она выполняется.

Асинхронная операция не требует синхронизации данных и это отложенное выполнение операции в будущем при наступлении определенных условий. Нужно еще сказать про блокирующие и неблокирующие операции, но для простоты будем рассматривать только синхронные блокирующие операции и асинхронные неблокирующие.

habr.com

Когда корутина делает сетевой запрос (например, через Ktor), она не блокирует поток ОС. Вместо этого она говорит: «Я ухожу в режим ожидания ответа от сети. Поток, на котором я выполнялась, теперь свободен». Поток ОС мгновенно берет другую корутину из очереди и начинает выполнять ее. Когда сетевой ответ приходит, первая корутина «просыпается» и возобновляет свою работу на любом свободном потоке.

Благодаря этому механизму, на одном потоке ОС могут конкурентно выполняться десятки тысяч корутин. Создание 100 000 корутин в Kotlin займет всего несколько десятков мегабайт памяти.

Ключевое слово suspend

В Kotlin магия приостановки работает через специальное ключевое слово suspend. Если функция помечена как suspend, это означает, что ее выполнение может быть приостановлено и возобновлено позже.

// Обычная блокирующая функция (заморозит поток)
fun downloadDataBlocking() {
    Thread.sleep(2000) // Имитация долгой загрузки
    println("Данные загружены")
}

// Приостанавливаемая функция (освободит поток для других задач)
suspend fun downloadDataSuspending() {
    delay(2000) // Специальная функция корутин, не блокирующая поток
    println("Данные загружены")
}

Важное правило языка Kotlin: suspend функцию можно вызвать только из другой suspend функции или из строителя корутин (например, launch или async).

Диспетчеры (Dispatchers): управление пулами потоков

Хотя корутины не привязаны жестко к потокам, им все равно нужны реальные потоки ОС для выполнения процессорных инструкций. В Kotlin за распределение корутин по потокам отвечают Диспетчеры (Dispatchers).

Диспетчер — это менеджер, который управляет пулом (группой) потоков ОС и решает, какой поток будет выполнять конкретную корутину в данный момент.

В стандартной библиотеке kotlinx.coroutines есть несколько встроенных диспетчеров, каждый из которых оптимизирован под свой тип задач:

Диспетчер	Назначение	Особенности пула потоков
Dispatchers.Default	Интенсивные вычисления (CPU-bound). Сортировка массивов, парсинг огромных JSON, криптография.	Количество потоков равно количеству ядер процессора. Нет смысла создавать больше потоков, чем есть физических ядер, так как задача требует 100% времени процессора.
Dispatchers.IO	Операции ввода-вывода (I/O-bound). Сетевые запросы, чтение/запись файлов, работа с базами данных.	Пул может динамически расширяться до 64 потоков (или больше). Потоки часто простаивают в ожидании ответа от диска или сети, поэтому их нужно много.
Dispatchers.Main	Обновление пользовательского интерфейса (UI).	Работает только в Android, JavaFX или Swing. Состоит ровно из одного главного потока. В нашем консольном приложении не используется.

Для нашего менеджера загрузок критически важен Dispatchers.IO. Скачивание файла — это классическая операция ввода-вывода. Мы читаем байты из сети (I/O) и записываем их на жесткий диск (I/O). Процессор при этом почти не нагружается.

Структурированная конкурентность

В старых языках программирования (например, в ранних версиях C++ или Java) запуск потока был операцией по принципу «запустил и забыл». Поток уходил в фоновый режим, и если в нем происходила ошибка, или если результат его работы больше не был нужен, остановить его было крайне сложно. Это приводило к утечкам памяти и зависаниям («утечка потоков»).

В парадигме неструктурированной конкурентности потоки запускаются в любом участке кода, без какой-либо конкретики относительно места их начала и конца. Данный код не безопасен, поскольку сбой одного из потоков не отменяет выполнение других задач, и результаты непредсказуемы.

nuancesprog.ru

Kotlin решает эту проблему с помощью парадигмы Структурированной конкурентности (Structured Concurrency).

Ее главный принцип: ни одна корутина не может быть запущена «в пустоте». Каждая корутина должна быть привязана к определенной области видимости (CoroutineScope). Область видимости устанавливает жесткую иерархию (родитель-ребенок) между корутинами.

Как это работает на практике:

1. Ожидание завершения: Родительская корутина не завершится, пока не завершатся все ее дочерние корутины. Вам не нужно вручную писать код для ожидания каждого потока.
2. Каскадная отмена: Если вы отменяете родительскую область видимости (например, пользователь нажал Ctrl+C для отмены загрузки), все дочерние корутины автоматически получают сигнал об отмене и корректно завершают работу, освобождая ресурсы.
3. Обработка ошибок: Если одна из дочерних корутин падает с ошибкой (например, оборвалось сетевое соединение при скачивании одного из чанков), ошибка передается родителю. Родитель автоматически отменяет все остальные дочерние корутины (зачем качать остальные части файла, если одна часть безвозвратно потеряна?) и обрабатывает исключение.

Практический пример: Scope и Launch

Давайте посмотрим, как выглядит запуск корутин с использованием структурированной конкурентности. Для создания области видимости используется функция coroutineScope, а для запуска новых корутин — строитель launch.

import kotlinx.coroutines.*

// suspend функция, имитирующая скачивание части файла
suspend fun downloadChunk(chunkId: Int) {
    println("Начало загрузки чанка $chunkId на потоке ${Thread.currentThread().name}")
    // Имитация сетевой задержки (не блокирует поток!)
    delay(1000)
    println("Чанк $chunkId успешно загружен")
}

fun main() = runBlocking {
    println("Программа запущена")

    // Создаем область видимости.
    // Переключаем контекст на Dispatchers.IO для сетевых операций
    withContext(Dispatchers.IO) {
        // Запускаем 3 корутины конкурентно
        launch { downloadChunk(1) }
        launch { downloadChunk(2) }
        launch { downloadChunk(3) }

        // Блок withContext не завершится, пока не отработают все 3 launch
    }

    println("Все чанки скачаны. Программа завершена.")
}

Если вы запустите этот код, вы увидите, что все три чанка начинают загружаться практически одновременно, и часто на разных потоках из пула DefaultDispatcher-worker. При этом функция main терпеливо ждет их завершения благодаря структурированной конкурентности.

Мы разобрали фундаментальные основы многопоточности. Вы узнали, почему создание тысяч потоков ОС — плохая идея, как корутины решают проблему блокировок с помощью механизма приостановки (suspend), и почему Dispatchers.IO станет основой нашего сетевого слоя. Концепция структурированной конкурентности гарантирует, что наш менеджер загрузок будет надежным: он не оставит «висящих» сетевых соединений в случае ошибки. Опираясь на эту теорию, на следующем этапе мы спроектируем архитектуру параллельной загрузки: научимся запрашивать у сервера поддержку докачки (Accept-Ranges), математически делить файл на равные блоки и скачивать их одновременно с помощью корутин.

Основы корутин в Kotlin

В системном программировании понимание теории — это лишь половина успеха. Ранее мы разобрали, почему потоки операционной системы слишком «тяжелые» для сетевых задач, и познакомились с концепцией легковесных корутин и диспетчеров. Теперь пришло время перенести эти знания в код.

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

Строители корутин: мост между мирами

Главное правило языка Kotlin гласит: приостанавливаемую функцию (suspend function) можно вызвать только из другой приостанавливаемой функции или из корутины.

Обычная функция main(), с которой начинается выполнение любой консольной программы, не является suspend. Она принадлежит к классическому, блокирующему миру. Чтобы войти в асинхронный мир корутин, используются специальные функции, которые называются строителями корутин (Coroutine Builders).

runBlocking: точка входа

Строитель runBlocking создает корутину и жестко блокирует текущий поток операционной системы до тех пор, пока эта корутина (и все ее дочерние задачи) не завершит свою работу.

В реальном Android-приложении или на сервере использование runBlocking считается плохой практикой, так как блокировка главного потока приводит к зависанию интерфейса. Однако в консольных утилитах это абсолютно необходимый инструмент. Если функция main() завершится, виртуальная машина Java (JVM) немедленно остановит программу, даже если в фоне скачиваются гигабайты данных. runBlocking заставляет main() терпеливо ждать.

import kotlinx.coroutines.*

fun main() {
    println("Начало программы")

    // Мост в асинхронный мир
    runBlocking {
        // Внутри этого блока мы находимся в корутине
        delay(1000) // Имитация работы
        println("Корутина завершила работу")
    }

    println("Конец программы")
}

launch: принцип «запустил и забыл»

Если runBlocking блокирует поток, то строитель launch работает иначе. Он запускает новую корутину в фоновом режиме, не блокируя текущий поток, и мгновенно возвращает управление дальше по коду.

Представьте работу шеф-повара в ресторане. Когда поступает заказ на стейк, шеф не стоит у плиты 15 минут, глядя на мясо (runBlocking). Он поручает эту задачу су-шефу (launch) и немедленно переходит к нарезке салата для другого заказа. Су-шеф жарит стейк в фоновом режиме.

В контексте нашего менеджера загрузок launch идеально подходит для фоновых задач, результат которых нам не нужно возвращать в виде переменной. Например, для периодического вывода прогресса скачивания (процентов) в консоль.

Управление жизненным циклом: объект Job

Когда вы вызываете launch, функция не возвращает пустоту. Она возвращает специальный объект типа Job (Задача).

Job — это дескриптор корутины. Это ваш пульт управления фоновой задачей. С его помощью можно узнать статус корутины (активна, завершена, отменена), дождаться ее окончания или принудительно прервать.

Job — это отменяемая сущность с жизненным циклом, который завершается по завершении корутины. Каждая корутина создаёт свой собственный Job.

habr.com

Ожидание завершения: join()

Часто возникает ситуация, когда нам нужно запустить фоновую задачу, поделать другие дела, а затем обязательно дождаться завершения этой фоновой задачи, прежде чем идти дальше. Для этого используется метод join().

import kotlinx.coroutines.*

fun main() = runBlocking {
    val downloadJob: Job = launch(Dispatchers.IO) {
        println("Начинаем скачивание файла...")
        delay(3000) // Имитация долгой загрузки
        println("Файл успешно скачан!")
    }

    println("Пока файл качается, мы можем подготовить место на диске.")

    // Программа дойдет до этой строчки мгновенно,
    // но дальше не пойдет, пока downloadJob не завершится
    downloadJob.join()

    println("Теперь можно разархивировать файл.")
}

Отмена загрузки: cancel()

Одно из главных преимуществ корутин — встроенный механизм кооперативной отмены. Если пользователь нажимает Ctrl+C или скорость интернета падает до нуля, мы должны иметь возможность остановить загрузку, чтобы не тратить ресурсы системы.

Метод cancel() отправляет корутине сигнал об отмене. Важно понимать слово кооперативная: корутина должна сама проверять, не отменили ли ее. Стандартные функции вроде delay() или сетевые вызовы Ktor делают это автоматически.

import kotlinx.coroutines.*

fun main() = runBlocking {
    val job = launch(Dispatchers.IO) {
        for (i in 1..100) {
            delay(100) // Проверка на отмену происходит здесь
            print("\rПрогресс: $i%")
        }
    }

    delay(350) // Ждем немного
    println("\nПользователь отменил загрузку!")
    job.cancel() // Отправляем сигнал отмены
    job.join()   // Ждем корректного завершения
    println("Загрузка прервана.")
}

Параллельное получение результатов: async и Deferred

Строитель launch отлично подходит для задач, которые просто выполняют действие (например, запись байтов на диск). Но что, если корутина должна вычислить какое-то значение и вернуть его нам? Например, мы хотим параллельно запросить размер файла у сервера и проверить свободное место на диске.

Для таких случаев используется строитель async.

async работает точно так же, как launch: он запускает корутину в фоне. Но вместо объекта Job он возвращает объект Deferred<T> (Отложенный результат), где T — тип возвращаемых данных.

Deferred — это обещание (promise), что в будущем, когда корутина закончит работу, внутри этого объекта появится результат. Чтобы получить этот результат, нужно вызвать метод await(). Если результат еще не готов, await() приостановит текущую корутину до его появления.

Математика параллелизма

Давайте посчитаем выгоду от использования async при скачивании файла, разбитого на чанки (части).

Предположим, мы скачиваем файл, состоящий из трех чанков. Время загрузки каждого чанка зависит от сети и составляет $t_1 = 4$ сек, $t_2 = 6$ сек, $t_3 = 3$ сек.

Если мы скачиваем их последовательно (в цикле без корутин), общее время $T_{seq}$ вычисляется как сумма:

$$T_{seq} = t_1 + t_2 + t_3$$

$T_{seq} = 4 + 6 + 3 = 13$ секунд.

Если мы используем async для одновременного скачивания всех трех чанков, они стартуют в одну и ту же миллисекунду. Общее время $T_{par}$ будет равно времени загрузки самого медленного чанка:

$$T_{par} = \max(t_1, t_2, t_3)$$

$T_{par} = \max(4, 6, 3) = 6$ секунд.

Мы ускорили процесс более чем в два раза. Именно этот принцип лежит в основе современных менеджеров загрузок.

Практический пример: awaitAll

В Kotlin есть удобная функция расширения awaitAll(), которая принимает список объектов Deferred и дожидается завершения их всех, возвращая список результатов.

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis

// Имитация скачивания чанка, возвращает количество скачанных байт
suspend fun downloadChunk(id: Int, time: Long): Int {
    println("Чанк $id: начало загрузки...")
    delay(time)
    println("Чанк $id: загружен!")
    return 1024 * 1024 // Вернули 1 Мегабайт
}

fun main() = runBlocking {
    val time = measureTimeMillis {
        // Создаем список отложенных задач
        val deferredChunks: List<Deferred<Int>> = listOf(
            async(Dispatchers.IO) { downloadChunk(1, 4000) },
            async(Dispatchers.IO) { downloadChunk(2, 6000) },
            async(Dispatchers.IO) { downloadChunk(3, 3000) }
        )

        // Ждем завершения всех задач параллельно
        val results: List<Int> = deferredChunks.awaitAll()

        val totalBytes = results.sum()
        println("Всего скачано: $totalBytes байт")
    }
    println("Общее время выполнения: ${time / 1000} сек.")
}

Запустив этот код, вы увидите, что общее время выполнения составит ровно 6 секунд, как мы и рассчитали по формуле.

Контекст корутины и withContext

Каждая корутина выполняется в определенном Контексте (CoroutineContext). Контекст — это набор элементов, определяющих поведение корутины.

Два самых важных элемента контекста, с которыми мы уже столкнулись:

1. Job — управляет жизненным циклом.
2. Dispatcher — определяет, на каком потоке (или пуле потоков) будет выполняться код.

Дочерняя корутина (корутина, запущенная из другой корутины) наследует контекст корутины-родителя (кроме Job).

habr.com

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

Например, мы скачали кусок данных из сети. Сетевой запрос должен выполняться на Dispatchers.IO. Затем нам нужно вычислить сложную контрольную сумму (хэш SHA-256) этого куска, чтобы убедиться, что файл не поврежден. Вычисление хэша — это интенсивная математическая операция, которая нагружает процессор. Ее правильнее выполнять на Dispatchers.Default.

Для безопасного переключения контекста «на лету» используется функция withContext.

Функция withContext приостанавливает выполнение текущей корутины, переносит ее выполнение на указанный диспетчер, выполняет блок кода, а затем автоматически возвращает корутину на исходный диспетчер.

import kotlinx.coroutines.*

suspend fun processDownloadedData(data: ByteArray) {
    // Допустим, мы находимся в Dispatchers.IO после скачивания
    println("1. Сеть: поток ${Thread.currentThread().name}")

    // Переключаемся на пул для вычислений
    val hash = withContext(Dispatchers.Default) {
        println("2. Вычисление хэша: поток ${Thread.currentThread().name}")
        // Имитация сложной математики
        delay(500)
        "a1b2c3d4"
    }

    // Мы автоматически вернулись в Dispatchers.IO!
    println("3. Запись на диск: поток ${Thread.currentThread().name}. Хэш: $hash")
}

fun main() = runBlocking {
    launch(Dispatchers.IO) {
        processDownloadedData(ByteArray(100))
    }.join()
}

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

Мы изучили базовый инструментарий корутин Kotlin. Теперь вы знаете, как удерживать консольную программу от преждевременного завершения с помощью runBlocking, как запускать фоновые процессы через launch и управлять ими через объект Job. Вы поняли математическую выгоду параллелизма и научились реализовывать ее с помощью async и Deferred. Наконец, мы разобрали, как жонглировать потоками с помощью withContext для оптимизации нагрузки на процессор и сеть.

В следующей статье мы объединим эти знания с HTTP-клиентом Ktor и файловой системой NIO, чтобы написать первую рабочую версию параллельного загрузчика файлов, который будет делить файл на чанки и скачивать их одновременно.

Диспетчеры и контекст корутин

Любая программа нуждается в среде выполнения. В классическом программировании такой средой выступает процесс операционной системы и его потоки. Легковесные корутины Kotlin, хотя и не привязаны жестко к системным потокам, также не могут существовать в вакууме. Им требуются четкие правила: где выполняться, как реагировать на ошибки и кто управляет их жизненным циклом.

Ранее мы научились переключать пулы потоков «на лету» с помощью withContext. Теперь пришло время заглянуть под капот этого механизма и разобраться, как именно Kotlin управляет тысячами одновременных загрузок, не перегружая процессор.

Анатомия корутины: CoroutineContext

Каждая корутина в Kotlin выполняется в определенном Контексте корутины (CoroutineContext). Технически это коллекция элементов, похожая на словарь (Map) или множество (Set), где каждый элемент имеет уникальный ключ.

Контекст хранит всю метаинформацию, необходимую для работы корутины. Основными элементами контекста являются:

1. Job — управляет жизненным циклом (запуск, отмена, завершение).
2. CoroutineDispatcher (Диспетчер) — определяет, на каком потоке или пуле потоков будет выполняться код.
3. CoroutineName — имя корутины (полезно для отладки и логирования).
4. CoroutineExceptionHandler — глобальный перехватчик необработанных исключений.

Элементы контекста можно комбинировать с помощью оператора +. Это похоже на сборку конфигурации перед запуском задачи.

import kotlinx.coroutines.*

fun main() = runBlocking {
    // Комбинируем диспетчер и имя корутины
    val customContext = Dispatchers.IO + CoroutineName("ChunkDownloader-1")

    launch(customContext) {
        println("Выполняемся в потоке: ${Thread.currentThread().name}")
        // Вывод: Выполняемся в потоке: DefaultDispatcher-worker-1 @ChunkDownloader-1#2
    }
}

CoroutineScope: Границы дозволенного

Контекст — это просто набор настроек. Чтобы эти настройки применить и запустить корутину, нужна Область видимости (CoroutineScope).

CoroutineScope — это интерфейс, который имеет всего одно свойство: val coroutineContext: CoroutineContext. Его главная задача — ограничить время жизни корутин и реализовать парадигму структурированной конкурентности.

Билдеры корутин, такие как launch и async, не являются самостоятельными функциями. Это функции-расширения для CoroutineScope. Именно поэтому вы не можете просто написать launch { ... } в пустом месте кода — компилятор выдаст ошибку. Вам нужен Scope.

Когда мы используем runBlocking, эта функция создает корневой CoroutineScope и блокирует текущий поток, пока все корутины внутри этого Scope не завершатся. Любая дочерняя корутина автоматически наследует контекст своего родителя (кроме объекта Job, который создается заново для каждой новой корутины, формируя иерархию).

Диспетчеры: Распределение рабочей силы

Диспетчер (Dispatcher) — это элемент контекста, который решает, какому системному потоку отдать код корутины на выполнение. Если корутина — это задача, то диспетчер — это менеджер, распределяющий задачи между рабочими (потоками).

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

Диспетчер	Назначение	Размер пула потоков	Пример использования в менеджере загрузок
Dispatchers.Default	Интенсивные вычисления (CPU-bound)	Равен количеству ядер процессора (минимум 2)	Вычисление хэш-сумм (MD5, SHA-256), парсинг больших JSON-ответов.
Dispatchers.IO	Ввод-вывод (I/O-bound), сетевые запросы	64 потока или количество ядер (что больше)	Скачивание чанков файла по сети, чтение/запись на жесткий диск.
Dispatchers.Unconfined	Выполнение без привязки к пулу	Выполняется в текущем потоке до первой приостановки	Специфичное тестирование (в реальных проектах используется редко).

Математика пула потоков

Почему для Dispatchers.Default количество потоков равно количеству ядер процессора $N_{cores}$? Если у вас 8-ядерный процессор, он может физически выполнять только 8 потоков одновременно. Если создать 100 потоков для тяжелых математических вычислений, операционная система начнет постоянно переключать контекст между ними. На это переключение тратятся ресурсы процессора, и общая производительность падает.

Для сетевых операций (Dispatchers.IO) ситуация иная. При скачивании файла поток большую часть времени ничего не вычисляет — он просто ждет ответа от сервера или готовности жесткого диска. Поэтому лимит потоков для ввода-вывода $T_{IO}$ рассчитывается по формуле:

$$T_{IO} = \max(64, N_{cores})$$

Это позволяет запустить 64 одновременных сетевых запроса, даже если у вас всего 4 ядра, так как эти потоки будут «спать» в ожидании данных, не нагружая процессор.

Под капотом: Как диспетчер меняет потоки

Код корутины выполняется в потоках, но корутины не привязаны жестко к конкретным потокам. Как происходит магия переключения?

Когда компилятор Kotlin обрабатывает suspend функцию, он преобразует ее в конечный автомат (State Machine). Состояние этого автомата хранится в специальном объекте Continuation (Продолжение).

Когда корутина приостанавливается (например, ждет байты из сети), она освобождает системный поток. Когда данные приходят, корутину нужно возобновить. Для этого объект Continuation оборачивается в DispatchedContinuation.

Вместо того чтобы напрямую продолжить выполнение, вызывается метод диспетчера dispatch(). Диспетчер берет свободный поток из своего пула и уже внутри этого потока вызывает метод resumeWith() у оригинального Continuation.

Именно поэтому корутина может начать работу в Thread-1, приостановиться для сетевого запроса, а после получения ответа продолжить работу уже в Thread-5.

Практика: Ограничение параллелизма

При разработке менеджера загрузок мы сталкиваемся с практической проблемой. Допустим, мы хотим скачать файл, разбив его на 100 чанков. Если мы запустим 100 корутин в Dispatchers.IO, они попытаются открыть 100 одновременных TCP-соединений с сервером.

Большинство серверов расценят это как DDoS-атаку и заблокируют ваш IP-адрес (вернут HTTP статус 429 Too Many Requests или разорвут соединение). Нам нужно ограничить количество одновременных загрузок, например, до 5.

Вместо создания собственных сложных очередей, Kotlin предоставляет элегантный метод limitedParallelism.

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis

// Имитация скачивания чанка
suspend fun downloadChunk(id: Int) {
    println("Чанк $id: начало загрузки в потоке ${Thread.currentThread().name}")
    delay(1000) // Имитация работы сети
    println("Чанк $id: завершен")
}

fun main() = runBlocking {
    // Создаем кастомный диспетчер на базе IO, но не более 3 потоков одновременно
    val downloadDispatcher = Dispatchers.IO.limitedParallelism(3)

    val time = measureTimeMillis {
        // Запускаем 10 задач
        val jobs = List(10) { id ->
            launch(downloadDispatcher) {
                downloadChunk(id)
            }
        }
        jobs.joinAll() // Ждем завершения всех чанков
    }

    println("Все чанки скачаны за ${time / 1000} сек.")
}

В этом примере, несмотря на то, что мы мгновенно создали 10 корутин, диспетчер downloadDispatcher пропустит на выполнение только 3 из них. Остальные 7 будут находиться в состоянии ожидания (Suspended), пока не освободится «слот». Это идеальный паттерн для контроля нагрузки на сеть и диск в нашем менеджере загрузок.

Обработка ошибок: CoroutineExceptionHandler

В многопоточной среде обработка ошибок усложняется. Если одна из корутин, скачивающих чанк, упадет с сетевой ошибкой (например, java.net.SocketTimeoutException), по правилам структурированной конкурентности эта ошибка передастся родительской корутине. Родительская корутина отменит все остальные дочерние задачи (остальные чанки перестанут качаться) и программа завершится с ошибкой.

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

import kotlinx.coroutines.*

fun main() = runBlocking {
    // Создаем обработчик ошибок
    val errorHandler = CoroutineExceptionHandler { context, exception ->
        println("Перехвачена фатальная ошибка: ${exception.message}")
        // Здесь можно записать ошибку в лог или попытаться перезапустить загрузку
    }

    // Добавляем обработчик в контекст корневой корутины
    val job = GlobalScope.launch(Dispatchers.IO + errorHandler) {
        println("Начинаем скачивание...")

        launch {
            delay(500)
            throw RuntimeException("Сервер разорвал соединение!")
        }

        launch {
            delay(1000)
            println("Этот код не выполнится, так как соседняя корутина упала")
        }
    }

    job.join()
    println("Программа продолжает работу, краха не произошло.")
}

Примечание: В примере выше использован GlobalScope исключительно для демонстрации работы перехватчика на верхнем уровне. В реальном коде менеджера загрузок мы будем использовать кастомные CoroutineScope с SupervisorJob, о которых поговорим позже.

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

Синхронизация и управление общим состоянием

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

Чтобы вывести в консоль красивый прогресс-бар, нам нужно знать общее количество скачанных байт. Каждая из десятков корутин, скачивающих свой чанк, должна постоянно сообщать: «Я скачала еще 8 килобайт». Все эти данные должны суммироваться в одной общей переменной. И именно здесь, на стыке параллельного выполнения и общих данных, скрывается одна из самых коварных проблем системного программирования.

Проблема общего состояния: Гонка данных

Представьте, что вы пишете консольный менеджер загрузок. У вас есть переменная downloadedBytes, которая хранит количество загруженных байт. Каждая корутина после получения порции данных из сети делает простое действие: downloadedBytes += 8192.

Кажется, что это одна простая операция. Но для процессора компьютера это не так. Операция сложения с присваиванием состоит из трех независимых шагов (цикл Read-Modify-Write):

1. Чтение: Процессор читает текущее значение downloadedBytes из оперативной памяти в свой регистр.
2. Модификация: Процессор прибавляет к этому значению 8192.
3. Запись: Процессор записывает новое значение обратно в оперативную память.

Если у нас работает только один поток, эти шаги выполняются строго друг за другом. Но в многопоточной среде (каковой является пул Dispatchers.IO) несколько потоков могут попытаться выполнить эти шаги одновременно. Это приводит к явлению, которое называется Гонка данных (Race Condition).

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

Давайте рассмотрим бытовой пример. Представьте совместный банковский счет, на котором лежит 10 000 руб. Вы и ваш партнер одновременно подходите к двум разным банкоматам и пытаетесь снять по 5 000 руб.

Банкомат А (ваш) читает баланс: 10 000 руб. Банкомат Б (партнера) читает баланс: 10 000 руб. Банкомат А вычитает 5 000 и записывает новый баланс: 5 000 руб. Банкомат Б вычитает 5 000 и записывает свой новый баланс: 5 000 руб.

В итоге вы сняли суммарно 10 000 руб., но на счету осталось 5 000 руб. Банк понес убытки из-за гонки данных. Точно так же ваш менеджер загрузок «потеряет» скачанные байты, и прогресс-бар никогда не достигнет 100%.

Чтобы убедиться в этом на практике, напишем тестовый код на Kotlin:

import kotlinx.coroutines.*

var downloadedBytes = 0

fun main() = runBlocking {
    // Запускаем 1000 корутин
    val jobs = List(1000) {
        launch(Dispatchers.Default) {
            // Каждая корутина прибавляет 1000
            downloadedBytes += 1000
        }
    }

    jobs.joinAll() // Ждем завершения всех корутин
    println("Всего скачано: $downloadedBytes")
}

Если вы запустите этот код несколько раз, вы почти никогда не увидите ожидаемый результат 1000000. Вы будете получать случайные числа: 984000, 991000, 973000. Часть обновлений была перезаписана и потеряна.

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

Атомарные операции: Быстро и надежно

Самый простой и быстрый способ защитить одиночную переменную (например, счетчик байт) — использовать Атомарные переменные (Atomic Variables).

Слово «атомарный» происходит от греческого atomos (неделимый). Атомарная операция выполняется целиком и мгновенно с точки зрения других потоков. Она не может быть прервана на середине.

В Kotlin (через платформу JVM) доступны классы из пакета java.util.concurrent.atomic, такие как AtomicInteger, AtomicLong, AtomicBoolean.

Как они работают под капотом? Они не блокируют потоки. Вместо этого они используют специальную инструкцию процессора, которая называется CAS (Compare-And-Swap, Сравни-и-Замени).

Логика CAS выглядит так:

1. Процессор читает текущее значение переменной.
2. Вычисляет новое значение.
3. Перед записью проверяет: «Изменилось ли значение в памяти с момента моего чтения?».
4. Если не изменилось — записывает новое значение.
5. Если изменилось (другой поток успел вклиниться) — операция отменяется, и процессор повторяет все шаги с пункта 1.

Перепишем наш пример с использованием AtomicLong (так как размер файлов часто превышает лимит обычного Int в 2 ГБ):

import kotlinx.coroutines.*
import java.util.concurrent.atomic.AtomicLong

// Используем атомарную переменную
val downloadedBytes = AtomicLong(0)

fun main() = runBlocking {
    val jobs = List(1000) {
        launch(Dispatchers.Default) {
            // Атомарное прибавление
            downloadedBytes.addAndGet(1000)
        }
    }

    jobs.joinAll()
    println("Всего скачано: ${downloadedBytes.get()}") // Всегда 1000000
}

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

Формула расчета процента загрузки $P$ выглядит так:

$$P = \frac{D}{T} \times 100$$

Где $D$ — количество скачанных байт (наш AtomicLong), а $T$ — общий размер файла в байтах (полученный из заголовка Content-Length).

Мьютексы: Защита сложных структур данных

Атомарные переменные прекрасно справляются с числами. Но что, если нам нужно обновить сложную структуру данных?

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

Для защиты критических секций кода (участков, где происходит работа с общими ресурсами) используется Мьютекс (Mutex — от английского Mutual Exclusion, взаимное исключение).

Мьютекс можно представить как ключ от примерочной в магазине. В примерочную (критическую секцию) может зайти только один человек (корутина). Он берет ключ, заходит и запирает дверь. Если подходит второй человек, он видит, что дверь заперта, и ждет снаружи. Когда первый выходит, он возвращает ключ, и следующий может зайти.

В классическом Java-программировании для этого используются блоки synchronized или ReentrantLock. Но в мире корутин Kotlin их использовать категорически нельзя.

Почему? Классические блокировки усыпляют (блокируют) системный поток ОС. Если корутина заблокирует поток из пула Dispatchers.IO, этот поток не сможет выполнять другие корутины. Если заблокировать все потоки пула, программа зависнет.

Вместо этого Kotlin предоставляет специальный корутинный мьютекс — kotlinx.coroutines.sync.Mutex.

Его главная особенность: если мьютекс занят, корутина не блокирует поток. Она приостанавливается (suspends), освобождая системный поток для других задач. Когда мьютекс освобождается, корутина возобновляет работу.

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock

// Общий ресурс - список завершенных чанков
val completedChunks = mutableListOf<Int>()
// Создаем мьютекс для защиты списка
val mutex = Mutex()

fun main() = runBlocking {
    val jobs = List(100) { chunkId ->
        launch(Dispatchers.IO) {
            // Имитация скачивания
            delay(50)

            // Критическая секция защищена мьютексом
            mutex.withLock {
                completedChunks.add(chunkId)
                println("Чанк $chunkId сохранен. Всего готово: ${completedChunks.size}")
            }
        }
    }

    jobs.joinAll()
    println("Загрузка завершена. Чанков скачано: ${completedChunks.size}")
}

Функция withLock — это безопасная обертка. Она автоматически захватывает мьютекс перед выполнением блока кода и гарантированно освобождает его после, даже если внутри блока произойдет ошибка (исключение).

Изоляция состояния: Однопоточный диспетчер

Мьютексы мощные, но у них есть недостаток: если логика становится слишком сложной, легко забыть обернуть какой-то участок кода в withLock, что снова приведет к гонке данных. Кроме того, чрезмерное использование мьютексов снижает производительность, так как корутины выстраиваются в очередь.

Существует альтернативный, очень элегантный архитектурный подход, который идеально подходит для Kotlin: Изоляция состояния (Thread Confinement).

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

Если поток всего один, гонка данных физически невозможна. Нам не нужны ни атомарные переменные, ни мьютексы.

В Kotlin это реализуется с помощью создания диспетчера, ограниченного одним потоком:

import kotlinx.coroutines.*

// Создаем диспетчер, который использует ровно 1 поток
val stateDispatcher = Dispatchers.Default.limitedParallelism(1)

// Обычные, не потокобезопасные переменные!
var totalBytes = 0L
val finishedChunks = mutableListOf<Int>()

fun main() = runBlocking {
    val jobs = List(100) { chunkId ->
        launch(Dispatchers.IO) {
            // Скачиваем данные в многопоточном пуле IO
            val bytesDownloaded = 8192L
            delay(10) // Имитация сети

            // Переключаемся в однопоточный контекст ТОЛЬКО для обновления состояния
            withContext(stateDispatcher) {
                totalBytes += bytesDownloaded
                finishedChunks.add(chunkId)
                // Здесь безопасно обновлять любые данные, гонки нет
            }
        }
    }

    jobs.joinAll()
    println("Успешно скачано байт: $totalBytes")
}

Этот паттерн часто используется в современных архитектурах (например, в Android при работе с UI-потоком или в MVI/Redux архитектурах).

Что выбрать для менеджера загрузок?

В системном программировании нет серебряной пули. Выбор инструмента зависит от задачи:

Инструмент	Когда использовать в менеджере загрузок	Плюсы	Минусы
AtomicLong	Подсчет скачанных байт, расчет скорости.	Максимальная скорость, не приостанавливает корутины.	Подходит только для простых чисел.
Mutex	Запись метаданных о чанках в файл конфигурации.	Позволяет защитить любую логику и структуры данных.	Корутины ждут в очереди, риск взаимных блокировок (Deadlocks).
Single-thread Dispatcher	Сложное управление состоянием (списки чанков, статусы, ошибки).	Код пишется как обычный синхронный, нет риска забыть блокировку.	Накладные расходы на переключение контекста (withContext).

В нашем проекте консольного менеджера загрузок мы применим комбинированный подход. Для высокочастотных обновлений (каждый килобайт из сети) мы будем использовать AtomicLong, чтобы не тормозить загрузку. А для редких, но сложных операций (например, отметка чанка как полностью завершенного и сохранение прогресса на диск) мы будем использовать Mutex.

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

Архитектура консольного менеджера загрузок

К этому моменту мы изучили все базовые строительные блоки системного программирования на Kotlin. Мы умеем работать с файловой системой, выполнять HTTP-запросы, запускать легковесные корутины для параллельной работы и безопасно управлять общим состоянием с помощью атомарных переменных и мьютексов.

Если собрать весь этот код в одну функцию main, программа, скорее всего, будет работать. Однако для курсовой работы и реальной инженерной практики этого недостаточно. Код, написанный сплошным полотном, невозможно тестировать, сложно читать, а добавление новой функции (например, паузы и возобновления загрузки) приведет к лавине ошибок.

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

Что такое архитектура и зачем она нужна?

В программной инженерии архитектура — это фундаментальная организация системы, воплощенная в ее компонентах, их связях друг с другом и с окружающей средой.

Представьте, что вы строите сарай. Вам не нужны чертежи: достаточно взять доски, гвозди и молоток. Но если вы строите многоэтажный дом (а многопоточный менеджер загрузок — это именно многоэтажный дом в мире консольных утилит), без детального плана здание рухнет под собственным весом.

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

Блог Роберта Мартина (Дяди Боба)

Для консольного приложения мы применим классическую Слоистую архитектуру (Layered Architecture). Суть этого подхода заключается в разделении кода на независимые уровни, каждый из которых решает только свою специфическую задачу.

Три главных принципа проектирования

Прежде чем мы перейдем к конкретным классам, зафиксируем три правила, которые сделают вашу курсовую работу образцовой:

1. Единая ответственность (Single Responsibility). Один класс должен выполнять только одну задачу. Класс, который скачивает данные из сети, не должен заниматься отрисовкой прогресс-бара в консоли.
2. Слабая связность (Low Coupling). Компоненты должны как можно меньше знать о внутреннем устройстве друг друга.
3. Внедрение зависимостей (Dependency Injection). Классы не должны сами создавать объекты, от которых они зависят (например, сетевой клиент). Эти объекты должны передаваться им извне, обычно через конструктор. Это критически важно для написания автоматических тестов.

Разделение на слои

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

1. Слой данных (Data Layer)

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

• HttpClient — обертка над сетевой библиотекой (например, Ktor). Умеет делать GET и HEAD запросы.
• FileWriter — класс, инкапсулирующий работу с RandomAccessFile. Умеет записывать массив байт по определенному смещению (offset) на жесткий диск.

2. Слой бизнес-логики (Domain Layer)

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

• DownloadManager — главный оркестратор. Он принимает URL, запрашивает размер файла, вычисляет размеры чанков и запускает процесс.
• ChunkDownloader — рабочий класс. Отвечает за скачивание конкретного диапазона байт и передачу их в FileWriter.
• ProgressTracker — хранитель состояния. Именно здесь инкапсулированы AtomicLong и Mutex, которые мы изучали в прошлой статье. Он считает скачанные байты и вычисляет скорость.

3. Слой представления (Presentation / UI Layer)

В нашем случае UI — это консоль (терминал). Этот слой отвечает за взаимодействие с пользователем.

• ConsoleApp — парсит аргументы командной строки (URL, количество потоков, путь сохранения).
• ProgressBarRenderer — в отдельной корутине регулярно опрашивает ProgressTracker и красиво выводит в stdout заполняющуюся полосу загрузки, проценты и оставшееся время.

Характеристика	Монолитный код (всё в main)	Слоистая архитектура
Тестируемость	Невозможно протестировать логику без реального скачивания файлов.	Можно протестировать ChunkDownloader, подсунув ему фейковый HttpClient.
Масштабируемость	Добавление графического интерфейса потребует переписывания всего кода.	Достаточно заменить ConsoleApp на окно с кнопками, логика не изменится.
Поиск ошибок	Ошибка может быть где угодно в 500 строках кода.	Ошибка записи на диск локализована строго в классе FileWriter.

Проектирование ядра: Распределение чанков

Остановимся подробнее на слое бизнес-логики. Главная математическая задача DownloadManager — правильно разделить файл на части (чанки) для параллельной загрузки.

Допустим, сервер вернул нам заголовок Content-Length, и мы знаем общий размер файла. Нам нужно вычислить размер одного чанка. Формула расчета $C$ выглядит следующим образом:

$$C = \lceil \frac{S}{N} \rceil$$

Где:

• $C$ — размер одного чанка в байтах.
• $S$ — общий размер файла в байтах.
• $N$ — количество потоков (корутин), заданное пользователем.
• $\lceil x \rceil$ — математическая операция округления вверх. Это необходимо, чтобы не «потерять» байты, если размер файла не делится нацело на количество потоков.

Например, размер файла $S = 1000$ байт, и мы хотим скачать его в $N = 3$ потока. $1000 / 3 = 333.33$. Округляем вверх, получаем $C = 334$ байта.

Теперь мы можем определить диапазоны (Ranges) для каждого потока:

• Поток 1: от 0 до 333 (скачает 334 байта)
• Поток 2: от 334 до 667 (скачает 334 байта)
• Поток 3: от 668 до 999 (скачает оставшиеся 332 байта)

Обратите внимание на последний поток: он скачивает меньше байт, так как файл закончился. Логика DownloadManager должна обязательно учитывать этот граничный случай (edge case), чтобы не запросить у сервера данные за пределами файла.

Практическая реализация: Скелет приложения

Давайте посмотрим, как эти концепции выражаются в коде на Kotlin. Мы не будем писать реализацию каждого метода (вы сделаете это в курсовой), но создадим правильные интерфейсы и классы.

Использование интерфейсов (ключевое слово interface) — это отличный тон в системном программировании. Интерфейс описывает что делает компонент, скрывая то, как он это делает.

// 1. Слой данных (Data Layer)
interface NetworkClient {
    suspend fun getFileSize(url: String): Long
    suspend fun downloadRange(url: String, start: Long, end: Long): ByteArray
}

interface Storage {
    fun writeChunk(offset: Long, data: ByteArray)
    fun allocateSpace(size: Long)
}

// 2. Слой бизнес-логики (Domain Layer)
class ProgressTracker(private val totalBytes: Long) {
    private val downloadedBytes = java.util.concurrent.atomic.AtomicLong(0)

    fun addBytes(count: Int) {
        downloadedBytes.addAndGet(count.toLong())
    }

    fun getProgressPercentage(): Int {
        if (totalBytes == 0L) return 0
        return ((downloadedBytes.get().toDouble() / totalBytes) * 100).toInt()
    }
}

// Внедрение зависимостей через конструктор
class DownloadManager(
    private val networkClient: NetworkClient,
    private val storage: Storage,
    private val threadCount: Int
) {
    suspend fun startDownload(url: String) {
        // 1. Получаем размер
        val fileSize = networkClient.getFileSize(url)

        // 2. Выделяем место на диске
        storage.allocateSpace(fileSize)

        // 3. Создаем трекер прогресса
        val tracker = ProgressTracker(fileSize)

        // 4. Вычисляем чанки и запускаем корутины (логика распределения)
        // ...
    }
}

Посмотрите на класс DownloadManager. Он не создает NetworkClient внутри себя через val client = MyKtorClient(). Вместо этого он требует передать ему готовый клиент при создании.

Зачем это нужно? Если на защите курсовой работы преподаватель попросит вас доказать, что логика распределения чанков работает правильно при обрыве сети, вам не придется отключать интернет на компьютере. Вы просто создадите FakeNetworkClient, который всегда выбрасывает ошибку на 50-м проценте загрузки, и передадите его в DownloadManager.

Управление потоком выполнения (Data Flow)

В многопоточной среде важно понимать, как компоненты общаются друг с другом. В нашей архитектуре мы используем паттерн Наблюдатель (Observer) или его современные аналоги в Kotlin — Flow (потоки данных).

Процесс выглядит так:

1. Главная функция main инициализирует все компоненты.
2. Запускается корутина DownloadManager.startDownload().
3. Параллельно запускается корутина ProgressBarRenderer.
4. ProgressBarRenderer в бесконечном цикле (с задержкой delay(100)) обращается к ProgressTracker.getProgressPercentage().
5. Как только загрузка завершается, DownloadManager подает сигнал, и цикл отрисовки прерывается.

Такой подход гарантирует, что тяжелые операции ввода-вывода (скачивание и запись на диск) полностью изолированы от логики отрисовки интерфейса. Если консоль «затормозит» при выводе текста, это никак не повлияет на скорость скачивания файла.

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

Архитектура должна предусматривать, что всё пойдет не так. Сеть может оборваться, на диске может закончиться место, сервер может вернуть ошибку 500.

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

• Если HttpClient получает ошибку таймаута (сервер долго не отвечает), он не должен завершать всю программу (вызывать exitProcess). Он должен выбросить исключение (Exception) наверх.
• ChunkDownloader перехватывает это исключение. Он знает, что скачивал конкретный чанк. Его решение: попробовать скачать этот же чанк еще раз (сделать 3 попытки).
• Если все 3 попытки провалились, ChunkDownloader передает ошибку в DownloadManager.
• DownloadManager понимает, что файл скачать невозможно. Он отменяет все остальные работающие корутины (благодаря механизму Structured Concurrency, который мы изучали), корректно закрывает файл на диске и передает сообщение об ошибке в слой UI.
• Слой UI выводит пользователю красивое сообщение красным цветом: «Ошибка загрузки: сервер недоступен», вместо того чтобы вываливать в консоль страшный StackTrace на 50 строк.

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

Получение метаданных файла по HTTP

В прошлой статье мы спроектировали архитектуру нашего консольного менеджера загрузок. Мы выделили слой данных и создали интерфейс NetworkClient, который должен отвечать за общение с сетью. Прежде чем мы начнем скачивать гигабайты данных и разбивать их на параллельные потоки, нам необходимо ответить на два критически важных вопроса: каков размер файла и поддерживает ли сервер скачивание по частям?

В этой статье мы реализуем первый шаг работы нашего менеджера — получение метаданных файла с помощью библиотеки Ktor.

Что такое метаданные и зачем они нужны?

Метаданные — это «данные о данных». В контексте загрузки файлов это служебная информация, которую сервер сообщает клиенту перед тем, как отправить само содержимое файла (тело ответа).

Представьте, что вы нанимаете грузовик для переезда. Прежде чем машина приедет, диспетчеру нужно знать объем ваших вещей, чтобы прислать подходящий транспорт. Если вы не сообщите объем, может приехать маленькая «Газель», в которую не поместится диван.

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

1. Выделение памяти на диске. Как мы обсуждали в статье про файловую систему, эффективнее заранее зарезервировать место на жестком диске с помощью RandomAccessFile.setLength(), чтобы избежать фрагментации.
2. Расчет чанков. Чтобы разделить работу между корутинами, нам нужно поделить общий размер файла на количество потоков.

Метод HTTP HEAD: Спрашиваем, но не скачиваем

В архитектуре Клиент-Сервер для получения файлов обычно используется метод GET. Однако, если мы отправим GET-запрос для файла размером 10 ГБ только ради того, чтобы узнать его размер, сервер начнет передавать нам все 10 ГБ. Это бессмысленная трата сетевого трафика и оперативной памяти.

Для решения этой проблемы в протоколе HTTP существует метод HEAD.

Метод HEAD запрашивает у сервера точно такой же ответ, как и метод GET, но без тела ответа (самого файла).

Спецификация RFC 7231

Используя HEAD, мы получаем только HTTP-заголовки — текстовые пары «ключ-значение», содержащие всю необходимую служебную информацию. Запрос HEAD выполняется за миллисекунды и потребляет всего несколько сотен байт трафика.

Ключевые HTTP-заголовки для менеджера загрузок

Из всего множества заголовков, которые вернет сервер, нас интересуют ровно два:

Заголовок	Значение	Описание для разработчика
Content-Length	Число (в байтах)	Указывает точный размер файла. Например, Content-Length: 1048576 означает, что файл весит ровно 1 Мегабайт. Если этого заголовка нет, сервер передает данные потоком неизвестной длины, и параллельная загрузка невозможна.
Accept-Ranges	bytes или none	Сообщает, поддерживает ли сервер докачку и скачивание файла по частям. Если сервер возвращает Accept-Ranges: bytes, мы можем смело запускать наши корутины. Если заголовка нет или он равен none, файл придется качать в один поток от начала до конца.

Реализация клиента на Ktor

Теперь перейдем к практике. В экосистеме Kotlin стандартом де-факто для сетевых запросов является асинхронный клиент Ktor.

Для создания HTTP-клиента в Ktor используется класс HttpClient. В качестве параметра он принимает сетевой движок (Engine). Мы будем использовать движок CIO (Coroutine-based I/O), так как он идеально подходит для асинхронной работы в консольных JVM-приложениях.

Важное правило работы с HttpClient

Создание объекта HttpClient — это ресурсоемкая операция. Под капотом он инициализирует пулы потоков, диспетчеры корутин и механизмы управления соединениями.

Типичная ошибка новичков — создавать новый клиент для каждого запроса. В правильной архитектуре HttpClient создается один раз при старте приложения и передается в другие классы через конструктор (внедрение зависимостей), а по завершении работы программы закрывается методом close() для освобождения ресурсов.

Пишем код получения метаданных

Давайте создадим структуру данных для хранения результата и напишем функцию, которая выполняет HEAD-запрос и парсит заголовки.

import io.ktor.client.*
import io.ktor.client.engine.cio.*
import io.ktor.client.request.*
import io.ktor.client.statement.*
import io.ktor.http.*

// 1. Класс для хранения полученных метаданных
data class FileMetadata(
    val sizeBytes: Long,
    val supportsRanges: Boolean
)

// 2. Наш сетевой клиент (реализация слоя данных)
class KtorNetworkClient(private val client: HttpClient) {

    suspend fun getFileMetadata(url: String): FileMetadata {
        // Выполняем HTTP HEAD запрос
        val response: HttpResponse = client.head(url)

        // Проверяем, что сервер ответил успешно (код 200 OK)
        if (!response.status.isSuccess()) {
            throw IllegalStateException("Ошибка сервера: ${response.status}")
        }

        // Извлекаем заголовок Content-Length
        val contentLengthStr = response.headers[HttpHeaders.ContentLength]
        val size = contentLengthStr?.toLongOrNull() ?: 0L

        // Извлекаем заголовок Accept-Ranges
        val acceptRanges = response.headers[HttpHeaders.AcceptRanges]
        val supportsRanges = acceptRanges == "bytes"

        // Если размер неизвестен, мы не сможем качать параллельно
        if (size <= 0L) {
            println("Внимание: Сервер не сообщил размер файла. Параллельная загрузка невозможна.")
            return FileMetadata(sizeBytes = 0L, supportsRanges = false)
        }

        return FileMetadata(sizeBytes = size, supportsRanges = supportsRanges)
    }
}

Разбор кода

1. client.head(url): Это suspend-функция Ktor, которая асинхронно отправляет HEAD-запрос. Она не блокирует основной поток выполнения, позволяя программе заниматься другими делами (например, отрисовкой интерфейса), пока мы ждем ответа от сервера.
2. response.headers[...]: Объект ответа содержит коллекцию заголовков. Мы используем встроенные константы Ktor (HttpHeaders.ContentLength), чтобы не ошибиться в написании строк.
3. Безопасное извлечение: Заголовки всегда приходят в виде строк. Мы используем функцию toLongOrNull(), которая безопасно попытается преобразовать строку в число. Если сервер прислал некорректные данные (или вообще не прислал заголовок), мы получим null, который благодаря оператору Элвиса ?: превратится в безопасный 0L.

Обработка граничных случаев (Edge Cases)

В системном программировании вы не можете доверять сети. Серверы настроены по-разному, и ваш менеджер загрузок должен быть готов к нестандартным ситуациям.

Что делать, если supportsRanges равно false? В этом случае слой бизнес-логики (DownloadManager) должен изменить свое поведение. Вместо того чтобы делить файл на чанки и запускать несколько корутин, он должен запустить ровно одну корутину и скачать файл целиком с помощью обычного GET-запроса. Если вы попытаетесь отправить запрос на скачивание куска файла серверу, который это не поддерживает, он либо вернет ошибку, либо начнет отдавать весь файл с самого начала, что приведет к повреждению данных на диске.

В следующей статье мы перейдем к самому интересному этапу — формированию запросов с заголовком Range и параллельному скачиванию байт в несколько корутин.

Разделение файла на части для загрузки

В предыдущем материале мы научились получать метаданные файла с помощью HTTP-запроса HEAD. Теперь наша программа знает точный размер файла в байтах и уверена, что сервер поддерживает докачку (заголовок Accept-Ranges: bytes). Имея эти данные, мы стоим перед следующей инженерной задачей: как именно разделить монолитный файл на независимые фрагменты, чтобы загружать их параллельно?

В системном программировании этот процесс называется чанкингом (от английского chunk — кусок, фрагмент). Правильное разделение файла — это фундамент, на котором строится вся логика многопоточного менеджера загрузок. Ошибка на один байт в расчетах приведет к тому, что итоговый файл окажется поврежденным («битым»), и операционная система не сможет его открыть.

Концепция параллельного скачивания

Чтобы понять суть процесса, представим задачу из реальной жизни. Вам нужно перевести книгу объемом 1000 страниц за минимальное время. Если вы отдадите всю книгу одному переводчику, он будет работать 100 дней. Это аналог однопоточной загрузки.

Чтобы ускорить процесс, вы нанимаете 10 переводчиков. Но вы не можете просто сказать им: «Переводите книгу». Вам необходимо четко распределить работу, чтобы они не переводили одни и те же страницы и не пропустили ни одной. Вы даете первому переводчику страницы с 1 по 100, второму — с 101 по 200, и так далее. Когда все закончат, вы просто сошьете их переводы по порядку.

В нашем консольном менеджере загрузок «книга» — это файл, «страницы» — это байты, а «переводчики» — это корутины Kotlin. Наша задача как архитекторов — написать алгоритм, который выдаст каждой корутине точный диапазон байт (от и до).

Протокол HTTP: Заголовок Range

Для того чтобы запросить у сервера не весь файл, а только конкретный его фрагмент, в протоколе HTTP существует специальный заголовок Range.

Спецификация HTTP определяет строгий формат передачи этого заголовка. Мы должны указать единицу измерения (всегда bytes) и диапазон индексов. Важно помнить, что индексация байтов начинается с нуля, как и в массивах в большинстве языков программирования.

Формат заголовка выглядит так: Range: bytes=НачальныйБайт-КонечныйБайт

Рассмотрим конкретные примеры:

• Range: bytes=0-499 — запрашивает первые 500 байт файла (с нулевого по 499-й включительно).
• Range: bytes=500-999 — запрашивает следующие 500 байт.
• Range: bytes=1000- — запрашивает все данные, начиная с 1000-го байта и до самого конца файла.

Статус 206 Partial Content

Когда мы отправляем обычный GET-запрос, успешный ответ сервера сопровождается статус-кодом 200 OK. Но когда мы используем заголовок Range, поведение сервера меняется.

Если сервер успешно обработал наш запрос и готов отдать запрошенный кусок, он возвращает статус-код 206 Partial Content (Частичное содержимое). Вместе с этим статусом сервер присылает заголовок Content-Range, который подтверждает, какой именно кусок он сейчас передает.

Статус 206 Partial Content — это главный индикатор того, что сервер понял наш запрос на скачивание части файла. Если в ответ на запрос с заголовком Range сервер возвращает 200 OK, это означает, что он проигнорировал наш заголовок и начал передавать весь файл целиком.

Спецификация RFC 7233

Математика разделения на чанки

Перейдем к математическим расчетам. У нас есть общий размер файла в байтах и желаемое количество потоков (корутин). Нам нужно вычислить размер одного стандартного чанка.

Базовая формула расчета размера чанка опирается на целочисленное деление:

$C = \lfloor \frac{S}{N} \rfloor$

Где:

• $C$ — базовый размер чанка в байтах.
• $S$ — общий размер файла в байтах.
• $N$ — количество потоков (корутин).
• $\lfloor x \rfloor$ — математическая операция округления вниз (в программировании это стандартное поведение при делении целых чисел).

Проблема остатка

Файлы редко делятся на количество потоков без остатка. Представим, что мы скачиваем файл размером 100 байт в 3 потока.

Применяем формулу: $100 / 3 = 33$ байта (остаток 1). Если мы создадим 3 чанка по 33 байта, мы скачаем только $33 \times 3 = 99$ байт. Один байт будет потерян, и файл окажется нерабочим.

Существует два элегантных решения этой проблемы:

1. Распределение остатка по одному байту первым чанкам. Это сложно в реализации и не дает ощутимого прироста производительности.
2. Добавление всего остатка к последнему чанку. Это стандартный подход в системном программировании. Последний поток скачает чуть больше данных, чем остальные, но разница составит максимум несколько байт, что абсолютно незаметно при современных скоростях интернета.

Проектирование доменной модели

Прежде чем писать алгоритм, нам нужно создать структуру данных для хранения информации о чанке. В Kotlin для этого идеально подходят data class.

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

data class DownloadChunk(
    val id: Int,
    val startByte: Long,
    val endByte: Long,
    var downloadedBytes: Long = 0L,
    var isCompleted: Boolean = false
)

Разберем поля нашего класса:

• id — порядковый номер чанка (полезно для логирования и вывода прогресса в консоль).
• startByte — индекс байта, с которого начинается загрузка этого фрагмента.
• endByte — индекс байта, на котором загрузка заканчивается.
• downloadedBytes — изменяемое поле (var), счетчик уже скачанных байт в рамках этого чанка. Необходимо для отображения прогресс-бара и реализации паузы/докачки.
• isCompleted — флаг успешного завершения загрузки фрагмента.

Реализация алгоритма на Kotlin

Теперь напишем функцию, которая принимает размер файла и количество потоков, а возвращает список готовых объектов DownloadChunk.

fun calculateChunks(fileSize: Long, threads: Int): List<DownloadChunk> {
    // Если файл пустой, возвращаем пустой список
    if (fileSize <= 0) return emptyList()

    val chunks = mutableListOf<DownloadChunk>()
    val chunkSize = fileSize / threads

    for (i in 0 until threads) {
        val startByte = i * chunkSize

        // Для последнего чанка конечный байт всегда равен концу файла
        val endByte = if (i == threads - 1) {
            fileSize - 1
        } else {
            startByte + chunkSize - 1
        }

        chunks.add(
            DownloadChunk(
                id = i,
                startByte = startByte,
                endByte = endByte
            )
        )
    }

    return chunks
}

Пошаговый разбор кода

1. Мы вычисляем базовый размер чанка chunkSize с помощью обычного целочисленного деления.
2. Запускаем цикл for от 0 до threads - 1.
3. startByte вычисляется просто: номер итерации умножается на размер чанка.
4. Самая важная часть — вычисление endByte. Мы используем конструкцию if как выражение. Если это последний чанк (i == threads - 1), мы жестко привязываем его конец к последнему байту файла (fileSize - 1), тем самым забирая весь возможный остаток от деления. Для всех остальных чанков конечный байт — это начальный байт плюс размер чанка минус один.

Проверим наш алгоритм на примере 100 байт и 3 потоков:

• Итерация 0: start = 0, end = 32 (размер 33 байта).
• Итерация 1: start = 33, end = 65 (размер 33 байта).
• Итерация 2: start = 66, end = 99 (размер 34 байта — остаток учтен!).

Алгоритм работает безупречно.

Граничные случаи (Edge Cases)

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

Случай 1: Файл слишком мал

Что произойдет, если пользователь попытается скачать текстовый файл размером 5 КБ (5120 байт), указав в настройках 32 потока?

Наш алгоритм честно разделит 5120 на 32 и создаст 32 чанка по 160 байт. Программа отправит 32 одновременных HTTP-запроса к серверу. Накладные расходы на установку TCP-соединения и TLS-рукопожатия (Handshake) для каждого запроса займут больше времени и трафика, чем скачивание самих 160 байт. Это антипаттерн.

Решение: Ввести константу минимального размера чанка (например, 1 Мегабайт). Перед расчетом необходимо скорректировать количество потоков.

val MIN_CHUNK_SIZE = 1024L * 1024L // 1 Мегабайт

fun optimizeThreadCount(fileSize: Long, requestedThreads: Int): Int {
    val maxPossibleThreads = (fileSize / MIN_CHUNK_SIZE).toInt()
    // Если файл меньше 1 МБ, maxPossibleThreads будет 0, используем минимум 1 поток
    val safeMaxThreads = if (maxPossibleThreads > 0) maxPossibleThreads else 1

    // Выбираем наименьшее из запрошенного пользователем и максимально разумного
    return minOf(requestedThreads, safeMaxThreads)
}

Случай 2: Сервер игнорирует заголовок Range

Как мы обсуждали ранее, некоторые серверы могут заявить о поддержке докачки в HEAD-запросе, но по факту проигнорировать заголовок Range при GET-запросе, вернув статус 200 OK вместо 206 Partial Content.

Если это произойдет, каждая из наших корутин начнет скачивать весь файл целиком с самого начала. Если мы качаем файл на 10 ГБ в 8 потоков, мы скачаем 80 ГБ данных, и все они запишутся в файл вперемешку, уничтожив данные.

Решение: В слое сетевого клиента необходимо строго проверять статус-код ответа.

// Пример интеграции с Ktor (подготовка к следующей статье)
val response = client.get(url) {
    header(HttpHeaders.Range, "bytes=${chunk.startByte}-${chunk.endByte}")
}

if (response.status == HttpStatusCode.OK) {
    // Сервер проигнорировал Range!
    // Необходимо отменить все остальные корутины и качать файл в один поток.
    throw ServerIgnoredRangeHeaderException()
} else if (response.status != HttpStatusCode.PartialContent) {
    throw IllegalStateException("Неожиданный статус: ${response.status}")
}

Подготовка к многопоточной загрузке

Разделив файл на чанки, мы подготовили идеальную структуру данных для параллельной работы. Теперь у нас есть список объектов DownloadChunk, каждый из которых содержит четкие инструкции: откуда качать и сколько качать.

В архитектуре нашего приложения этот список будет передан в DownloadManager. Там мы создадим пул корутин (используя диспетчер Dispatchers.IO), и каждая корутина возьмет себе один чанк из списка, сформирует HTTP-запрос с заголовком Range и начнет потоковую запись полученных байт на диск.

Именно процесс безопасной записи этих фрагментов в один общий файл на жестком диске без состояния гонки (Race Condition) станет темой нашего следующего занятия.

Параллельная загрузка фрагментов файла

Имея на руках список объектов DownloadChunk, каждый из которых содержит точные байтовые границы (от и до), мы подошли к кульминации нашего проекта. Наша задача — заставить программу скачивать эти фрагменты одновременно и сохранять их на жесткий диск.

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

Физика конкурентной записи на диск

Представьте, что вы с тремя коллегами пишете одну большую книгу. Если вы будете писать по очереди, проблем не возникнет. Но если вы начнете писать одновременно на одном листе бумаги, ваши слова наложатся друг на друга, превратив текст в нечитаемую грязь.

В операционных системах эта проблема известна как состояние гонки (Race Condition) при операциях ввода-вывода. Когда стандартный поток записи (FileOutputStream) открывает файл, он поддерживает внутренний указатель (курсор). Каждая операция записи сдвигает этот курсор вперед. Если две корутины попытаются записать данные через один и тот же поток, операционная система смешает их байты.

Чтобы решить эту проблему, мы должны заранее выделить место под всю «книгу» и жестко закрепить за каждым «писателем» его собственные страницы.

Инструмент RandomAccessFile

В экосистеме Java (и, следовательно, Kotlin) для произвольного доступа к файлу используется класс RandomAccessFile. В отличие от обычных потоков, он позволяет перемещать внутренний курсор в любую точку файла перед чтением или записью.

Ключевой метод этого класса — seek(Long pos). Он приказывает операционной системе переместить указатель записи на конкретный байт от начала файла.

val file = RandomAccessFile("downloaded_video.mp4", "rw")
// Перемещаем курсор на 1024-й байт
file.seek(1024L)
// Записываем данные (они перезапишут то, что было начиная с 1024 байта)
file.write(byteArrayOf(1, 2, 3))
file.close()

Флаг "rw" означает Read/Write (чтение и запись). Если файла не существует, он будет создан.

Изоляция ресурсов: Главная ловушка новичков

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

Даже если вы используете seek() перед каждой записью, использование одного объекта RandomAccessFile несколькими потоками приведет к повреждению данных.

Почему это происходит? Рассмотрим сценарий:

1. Корутина А вызывает seek(0).
2. Планировщик ОС приостанавливает Корутину А и передает управление Корутине Б.
3. Корутина Б вызывает seek(1000).
4. Планировщик возвращает управление Корутине А.
5. Корутина А вызывает write(). Но курсор уже находится на позиции 1000!

Архитектурное решение: Каждая корутина должна открывать свой собственный экземпляр RandomAccessFile, указывающий на один и тот же физический файл на диске. Операционные системы (Windows, Linux, macOS) прекрасно справляются с конкурентной записью в один файл из разных дескрипторов, если их области записи не пересекаются.

Потоковая передача данных (Streaming)

Мы не можем просто сказать HTTP-клиенту: «Скачай этот чанк размером 100 МБ и верни мне его в виде массива байт». Если мы запустим 10 таких корутин, приложение мгновенно потребит 1 ГБ оперативной памяти и упадет с ошибкой OutOfMemoryError.

Данные из сети нужно читать небольшими порциями — буферами — и сразу же сбрасывать их на диск. В библиотеке Ktor для работы с асинхронными потоками данных используется интерфейс ByteReadChannel.

Математика буферизации

Размер буфера — это компромисс между потреблением памяти и нагрузкой на процессор.

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

$$T = \frac{S}{B} \times L$$

Где:

• $T$ — общее время накладных расходов на операции ввода-вывода.
• $S$ — общий размер чанка в байтах.
• $B$ — размер буфера в байтах.
• $L$ — задержка (latency) одного системного вызова (чтение из сокета + запись на диск).

Если размер буфера $B$ равен 1 байту, для скачивания 1 Мегабайта программа сделает 1 048 576 системных вызовов. Переключение контекста между пользовательским режимом (User Space) и ядром ОС (Kernel Space) убьет всю производительность.

Если $B$ равен 100 МБ, мы исчерпаем оперативную память.

Стандартом в системном программировании считается буфер размером 8 КБ (8192 байта). Это значение кратно размеру страницы памяти в большинстве архитектур процессоров и размеру кластера файловой системы, что обеспечивает оптимальную скорость без перерасхода RAM.

Реализация функции загрузки чанка

Теперь объединим сеть, буферизацию и запись на диск в единую функцию. Эта функция будет выполняться изолированно в рамках одной корутины.

import io.ktor.client.*
import io.ktor.client.request.*
import io.ktor.client.statement.*
import io.ktor.http.*
import io.ktor.utils.io.*
import io.ktor.utils.io.core.*
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext
import java.io.File
import java.io.RandomAccessFile

suspend fun downloadChunk(
    client: HttpClient,
    url: String,
    file: File,
    chunk: DownloadChunk
) = withContext(Dispatchers.IO) { // Переключаемся на пул потоков для ввода-вывода

    // 1. Открываем независимый дескриптор файла для этой корутины
    val randomAccessFile = RandomAccessFile(file, "rw")

    try {
        // 2. Выполняем HTTP GET запрос с заголовком Range
        val response: HttpResponse = client.prepareGet(url) {
            header(HttpHeaders.Range, "bytes=${chunk.startByte}-${chunk.endByte}")
        }.execute()

        // Проверяем, что сервер поддерживает частичную загрузку
        if (response.status != HttpStatusCode.PartialContent && response.status != HttpStatusCode.OK) {
            throw IllegalStateException("Неожиданный статус сервера: ${response.status}")
        }

        // 3. Получаем канал для потокового чтения
        val channel: ByteReadChannel = response.bodyAsChannel()

        // 4. Создаем буфер на 8 КБ
        val buffer = ByteArray(8 * 1024)

        // 5. Устанавливаем курсор файла на начальную позицию чанка
        var currentPosition = chunk.startByte
        randomAccessFile.seek(currentPosition)

        // 6. Главный цикл чтения/записи
        while (!channel.isClosedForRead) {
            // Читаем доступные байты из сети в буфер
            val bytesRead = channel.readAvailable(buffer, 0, buffer.size)

            if (bytesRead > 0) {
                // Записываем ровно столько байт, сколько прочитали
                randomAccessFile.write(buffer, 0, bytesRead)

                // Обновляем состояние чанка (для прогресс-бара)
                chunk.downloadedBytes += bytesRead
                currentPosition += bytesRead
            }
        }

        // Отмечаем чанк как завершенный
        chunk.isCompleted = true

    } finally {
        // 7. Гарантированно закрываем файл, даже если произошла ошибка
        randomAccessFile.close()
    }
}

Разбор критических участков кода

Обратите внимание на использование client.prepareGet(url).execute(). В отличие от обычного client.get(), который скачивает весь ответ в память перед возвратом, prepareGet устанавливает соединение, получает заголовки и останавливается. Тело ответа остается в сетевом сокете ОС, ожидая, пока мы начнем вытягивать его через ByteReadChannel.

Внутри цикла while метод readAvailable возвращает количество реально прочитанных байт. Это критически важно. В конце чанка в сети может остаться, например, всего 500 байт. Если мы вызовем randomAccessFile.write(buffer) без указания длины, мы запишем 500 полезных байт и 7692 байта старого мусора, оставшегося в буфере от предыдущей итерации. Конструкция write(buffer, 0, bytesRead) гарантирует запись только свежих данных.

Блок finally обязателен. Если во время скачивания пропадет интернет, Ktor выбросит исключение. Если мы не закроем RandomAccessFile, операционная система не освободит блокировку файла (File Lock), и мы не сможем ни удалить его, ни возобновить загрузку.

Оркестрация: Запуск роя корутин

У нас есть функция, которая скачивает один чанк. Теперь нам нужно запустить её для всех чанков одновременно и дождаться завершения.

В парадигме структурированной конкурентности Kotlin для параллельного выполнения задач с ожиданием результата используется связка coroutineScope, async и awaitAll().

import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.coroutineScope

suspend fun downloadFileParallel(
    client: HttpClient,
    url: String,
    destination: File,
    chunks: List<DownloadChunk>
) = coroutineScope { // Создаем область видимости для группы корутин

    println("Начинаем загрузку в ${chunks.size} потоков...")

    // Запускаем все чанки параллельно
    val deferredDownloads = chunks.map { chunk ->
        async {
            downloadChunk(client, url, destination, chunk)
        }
    }

    // Приостанавливаем текущую корутину, пока ВСЕ чанки не скачаются
    deferredDownloads.awaitAll()

    println("Загрузка успешно завершена!")
}

Как работает awaitAll() и обработка ошибок

Функция async не блокирует поток. Она мгновенно запускает фоновую задачу и возвращает объект Deferred (обещание результата в будущем). Метод map пробегается по списку из 8 чанков за долю миллисекунды, создавая 8 параллельных сетевых запросов.

Магия происходит на строке awaitAll(). Эта функция приостанавливает выполнение downloadFileParallel, пока все 8 задач не завершатся успешно.

Но что произойдет, если сервер разорвет соединение на 3-м чанке?

Здесь проявляется мощь структурированной конкурентности Kotlin. Если одна из корутин внутри coroutineScope выбрасывает исключение (например, java.net.SocketException), область видимости немедленно отменяет все остальные активные корутины. Загрузка 1, 2, 4, 5, 6, 7 и 8 чанков будет прервана, сетевые соединения закрыты, а исключение пробросится выше по стеку вызовов.

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

Синхронизация прогресса

В коде downloadChunk мы обновляем поле chunk.downloadedBytes += bytesRead. Поскольку каждая корутина работает только со своим собственным объектом DownloadChunk, здесь нет состояния гонки.

Однако, чтобы отобразить общий прогресс-бар в консоли, нам потребуется отдельная корутина, которая будет периодически (например, раз в 500 миллисекунд) пробегаться по списку чанков, суммировать их downloadedBytes и выводить результат на экран.

О том, как спроектировать красивый и не мерцающий консольный интерфейс для отслеживания скорости и прогресса загрузки, мы поговорим в следующей статье.

Объединение скачанных фрагментов в единый файл

В системном программировании существует два глобальных архитектурных подхода к параллельной загрузке файлов. Первый подход — предварительное выделение памяти (Pre-allocation), при котором мы сразу создаем пустой файл нужного размера и пишем данные в разные его участки с помощью RandomAccessFile. Этот метод мы подробно разобрали ранее.

Однако в реальной практике часто применяется второй подход — загрузка во временные файлы с их последующим слиянием. В этом сценарии каждый поток скачивает свой фрагмент в отдельный независимый файл на диске (например, part1.tmp, part2.tmp), а после успешного завершения всех загрузок программа склеивает их в финальный файл.

Зачем нужен подход с временными файлами?

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

1. Отсутствие блокировок (Lock Contention): Операционной системе не нужно синхронизировать доступ нескольких потоков к одному дескриптору файла. Каждый поток работает со своим изолированным файлом, что на некоторых файловых системах дает прирост скорости записи.
2. Безопасность при сбоях: Если во время загрузки отключат электричество, единый файл может оказаться поврежденным (битым). Временные файлы остаются целыми, и при перезапуске менеджер загрузок точно знает, какие фрагменты скачаны полностью, а какие нужно запросить заново.
3. Поддержка файловых систем без Sparse Files: Не все файловые системы эффективно поддерживают создание огромных пустых файлов. На старых системах (например, FAT32) предварительное выделение 4 ГБ может занять несколько минут, в течение которых программа будет «висеть».

Главный недостаток этого метода — двойной расход дискового пространства в момент слияния. Математически это выражается формулой:

$$S_{total} = S_{file} + \sum_{i=1}^{n} S_{chunk_i}$$

Где:

• $S_{total}$ — пиковое потребление места на диске.
• $S_{file}$ — размер итогового файла.
• $S_{chunk_i}$ — размер $i$-го временного фрагмента.
• $n$ — количество потоков.

Поскольку сумма размеров всех фрагментов равна размеру итогового файла, в пиковый момент (до удаления временных файлов) программе потребуется ровно в два раза больше места, чем весит сам файл. Если вы качаете игру на 100 ГБ, на диске должно быть свободно 200 ГБ.

Проблема классического ввода-вывода

Допустим, у нас есть 8 скачанных файлов-фрагментов, и нам нужно собрать их по порядку. Самое наивное решение новичка — прочитать содержимое каждого файла в массив байт (ByteArray) и записать в новый файл.

Это приведет к катастрофе. Если фрагмент весит 500 МБ, программа попытается выделить 500 МБ в оперативной памяти. Сборщик мусора (Garbage Collector) JVM сойдет с ума, а приложение упадет с ошибкой OutOfMemoryError.

Чуть более продвинутый подход — использовать буферизированные потоки (BufferedInputStream и BufferedOutputStream), перекладывая данные порциями по 8 КБ. Это спасет оперативную память, но создаст огромную нагрузку на процессор из-за архитектуры современных операционных систем.

При использовании стандартных потоков данные проходят следующий путь:

1. Жесткий диск читает данные в буфер ядра ОС (Kernel Space).
2. ОС копирует данные из ядра в память вашего Kotlin-приложения (User Space).
3. Ваше приложение передает эти же данные обратно в ядро ОС для записи.
4. Ядро ОС отправляет данные на жесткий диск.

Для простого копирования файла из одного места в другое мы делаем два лишних копирования в оперативной памяти и постоянно переключаем контекст процессора.

Магия Zero-Copy и FileChannel

В системном программировании для решения этой проблемы используется концепция Zero-Copy (нулевое копирование). Суть в том, чтобы приказать операционной системе передать данные напрямую из одного файла в другой на уровне ядра, вообще не поднимая их в память приложения.

В экосистеме Java/Kotlin доступ к этой системной возможности предоставляет пакет java.nio (New I/O) и класс FileChannel.

Метод transferTo() позволяет перекачать байты из канала чтения напрямую в канал записи. На Linux и macOS этот метод под капотом вызывает системный вызов sendfile(), который работает с фантастической скоростью, ограниченной только физической пропускной способностью вашего SSD.

Реализация слияния на Kotlin

Напишем функцию, которая принимает список временных файлов (строго отсортированных по порядку!) и объединяет их в финальный файл.

import java.io.File
import java.io.FileInputStream
import java.io.FileOutputStream
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext

suspend fun mergeChunksZeroCopy(chunks: List<File>, outputFile: File) = withContext(Dispatchers.IO) {
    println("Начинаем слияние файлов...")

    // Открываем канал для записи в итоговый файл (флаг true означает append - дописывание в конец)
    FileOutputStream(outputFile, true).channel.use { outputChannel ->

        for (chunk in chunks) {
            // Открываем канал для чтения текущего фрагмента
            FileInputStream(chunk).channel.use { inputChannel ->
                val size = inputChannel.size()
                var position = 0L

                // transferTo может не перенести все данные за один раз,
                // поэтому используем цикл while
                while (position < size) {
                    position += inputChannel.transferTo(position, size - position, outputChannel)
                }
            }

            // После успешного переноса данных удаляем временный файл
            chunk.delete()
            println("Фрагмент ${chunk.name} успешно объединен и удален.")
        }
    }

    println("Слияние завершено! Файл сохранен как ${outputFile.name}")
}

Разбор критических участков кода

Функция use { ... } в Kotlin — это аналог try-with-resources из Java. Она гарантирует, что файловый канал будет безопасно закрыт после выполнения блока кода, даже если внутри произойдет исключение (например, закончится место на диске). Это предотвращает утечки файловых дескрипторов.

Обратите внимание на цикл while (position < size). В документации к методу transferTo() указано, что он не гарантирует перенос всех запрошенных байт за один вызов. В зависимости от ОС и состояния буферов, он может перенести только часть данных и вернуть количество реально скопированных байт. Поэтому мы обязаны сдвигать позицию и повторять вызов, пока не перенесем весь фрагмент целиком.

Проверка целостности (Хэширование)

После того как файл собран воедино (или скачан полностью через RandomAccessFile), профессиональный менеджер загрузок должен убедиться, что файл не поврежден. При передаче по сети пакеты могут теряться или искажаться.

Для этого серверы часто передают в HTTP-заголовках (например, ETag или кастомных X-Checksum-MD5) хэш-сумму оригинального файла. Хэш-сумма — это уникальная цифровая подпись файла фиксированной длины. Если изменить в файле размером 1 ГБ хотя бы один бит, его хэш изменится до неузнаваемости.

В Kotlin мы можем вычислить хэш (например, SHA-256) с помощью встроенного класса MessageDigest. Чтобы не загружать весь файл в память, мы читаем его потоком:

import java.io.File
import java.security.MessageDigest

fun calculateSHA256(file: File): String {
    val digest = MessageDigest.getInstance("SHA-256")

    // Читаем файл блоками по 8 КБ и скармливаем их алгоритму
    file.inputStream().use { fis ->
        val buffer = ByteArray(8192)
        var bytesRead:
        while (fis.read(buffer).also { bytesRead = it } != -1) {
            digest.update(buffer, 0, bytesRead)
        }
    }

    // Преобразуем массив байт в красивую шестнадцатеричную строку
    return digest.digest().joinToString("") { "%02x".format(it) }
}

Если вычисленный хэш совпадает с тем, что отдал сервер — загрузка абсолютно успешна. Вы создали надежный, многопоточный и отказоустойчивый системный инструмент.

Обработка сетевых ошибок и таймаутов

В предыдущих статьях мы спроектировали архитектуру нашего менеджера загрузок, научились делить файл на фрагменты, скачивать их параллельно и объединять на диске с помощью технологии Zero-Copy. Кажется, программа готова. Но если вы запустите её для скачивания файла размером 10 ГБ через нестабильный мобильный интернет, она почти гарантированно зависнет или упадет с ошибкой.

В системном и сетевом программировании существует фундаментальное правило: сеть всегда ненадежна. Пакеты теряются, маршрутизаторы перезагружаются, серверы обрывают соединения, а у пользователя может на секунду пропасть Wi-Fi. Профессиональный менеджер загрузок отличается от студенческой поделки тем, как он реагирует на эти сбои.

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

Иллюзия надежной сети

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

Представьте, что вы заказали доставку пиццы (отправили HTTP-запрос). Возможны три сценария сбоя:

1. Ошибка на вашей стороне: вы забыли включить телефон (нет подключения к интернету, ошибка DNS).
2. Ошибка на стороне пиццерии: у них отключили свет или сломалась печь (сервер вернул ошибку 500 Internal Server Error).
3. Проблема в пути: курьер заблудился или попал в пробку. Вы ждете час, два, три, но ни пиццы, ни отказа нет.

Третий сценарий — самый опасный для программиста. Если сервер просто «молчит», системный сокет может оставаться открытым часами. В контексте нашего приложения это означает, что корутина, скачивающая чанк, зависнет навсегда. Она будет удерживать оперативную память и файловый дескриптор, а прогресс-бар остановится на 99%.

Таймауты: защита от бесконечного ожидания

Чтобы предотвратить зависание, мы обязаны использовать таймауты — жесткие временные лимиты на выполнение сетевых операций. Если операция не завершается за отведенное время, она принудительно прерывается с ошибкой.

В HTTP-клиенте Ktor для этого существует специальный плагин HttpTimeout. Он позволяет настроить три вида лимитов:

• requestTimeoutMillis — максимальное время на весь запрос целиком (от отправки первого байта до получения последнего байта ответа).
• connectTimeoutMillis — время, выделенное на установку TCP-соединения с сервером.
• socketTimeoutMillis — максимальное время ожидания между получением двух соседних пакетов данных.

Для менеджера загрузок наиболее критичен socketTimeoutMillis. Если мы качаем большой файл, весь запрос может занять часы, поэтому requestTimeoutMillis нам не подходит. А вот если данные перестали поступать более чем на 15 секунд — это явный признак обрыва связи.

Настроим наш клиент Ktor:

import io.ktor.client.*
import io.ktor.client.engine.cio.*
import io.ktor.client.plugins.*

val httpClient = HttpClient(CIO) {
    install(HttpTimeout) {
        connectTimeoutMillis = 10_000L // 10 секунд на подключение
        socketTimeoutMillis = 15_000L  // 15 секунд между пакетами
    }
}

Если сервер перестанет присылать байты, Ktor выбросит исключение HttpRequestTimeoutException (или SocketTimeoutException). Наша корутина не зависнет, а завершится с ошибкой. Теперь эту ошибку нужно правильно обработать.

Стратегия повторных попыток (Exponential Backoff)

Если загрузка одного фрагмента прервалась из-за таймаута, глупо отменять скачивание всего файла. Мы должны попробовать скачать этот фрагмент заново (с того места, где произошел обрыв).

Самое наивное решение — использовать цикл while и долбить сервер запросами без остановки, пока он не отдаст файл. Это называется DDoS-атакой. Если сервер упал от перегрузки, а тысячи клиентов начнут агрессивно переподключаться каждую миллисекунду, сервер не поднимется никогда.

В индустрии стандартом де-факто является алгоритм Exponential Backoff (экспоненциальная задержка). Его суть проста: с каждой неудачной попыткой время ожидания перед следующим запросом увеличивается в геометрической прогрессии.

Математически это описывается следующей формулой:

$$T_{wait} = T_{base} \times 2^n$$

Где:

• $T_{wait}$ — время, которое программа должна подождать перед следующей попыткой.
• $T_{base}$ — базовое (начальное) время задержки (например, 1000 миллисекунд).
• $n$ — порядковый номер неудачной попытки (начиная с 0).

При $T_{base} = 1000$:

• После 1-й ошибки ($n=0$) мы ждем 1 секунду.
• После 2-й ошибки ($n=1$) мы ждем 2 секунды.
• После 3-й ошибки ($n=2$) мы ждем 4 секунды.
• После 4-й ошибки ($n=3$) мы ждем 8 секунд.

Такой подход дает серверу «перевести дух» и восстановиться после сбоя, а также экономит ресурсы процессора на клиенте.

Реализация Exponential Backoff на Kotlin

В Kotlin мы можем написать универсальную функцию высшего порядка (Higher-Order Function), которая будет принимать любой блок кода и выполнять его с экспоненциальной задержкой в случае ошибок.

import kotlinx.coroutines.delay
import kotlin.math.pow

suspend fun <T> withRetry(
    maxRetries: Int = 5,
    baseDelayMs: Long = 1000L,
    block: suspend () -> T
): T {
    var currentAttempt = 0

    while (true) {
        try {
            // Пытаемся выполнить переданный блок кода
            return block()
        } catch (e: Exception) {
            currentAttempt++

            // Если исчерпали лимит попыток - пробрасываем ошибку дальше
            if (currentAttempt >= maxRetries) {
                println("Превышен лимит попыток. Последняя ошибка: ${e.message}")
                throw e
            }

            // Вычисляем задержку по формуле: baseDelay * 2^n
            val delayTime = (baseDelayMs * 2.0.pow(currentAttempt - 1)).toLong()
            println("Ошибка сети. Повторная попытка через ${delayTime} мс...")

            // Приостанавливаем корутину (не блокируя системный поток!)
            delay(delayTime)
        }
    }
}

Теперь мы можем обернуть нашу функцию скачивания чанка в withRetry. Если произойдет HttpRequestTimeoutException, корутина просто подождет и попробует снова.

Изоляция ошибок через класс Result

Вспомним концепцию структурированной конкурентности (Structured Concurrency) из прошлых статей. Если мы запускаем 8 корутин через async внутри coroutineScope, и хотя бы одна из них выбрасывает необработанное исключение (например, закончились попытки в withRetry), родительский coroutineScope немедленно отменяет все остальные 7 успешно работающих корутин.

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

В Kotlin для элегантной обработки ошибок без выброса исключений наружу используется встроенный класс Result<T>. Он инкапсулирует либо успешный результат (тип T), либо исключение (Throwable).

Для удобной работы с Result применяется функция runCatching:

// Функция теперь возвращает Result<File>, а не просто File
suspend fun downloadChunkSafe(url: String, chunkId: Int): Result<File> {
    return runCatching {
        withRetry(maxRetries = 5) {
            // Здесь вызывается Ktor клиент и происходит запись на диск
            // Если после 5 попыток будет ошибка, withRetry выбросит Exception,
            // но runCatching перехватит его и упакует в Result.failure()
            downloadChunk(url, chunkId)
        }
    }
}

Теперь в главном потоке, где мы оркестрируем загрузку, мы можем безопасно дождаться всех результатов и проверить их:

val results: List<Result<File>> = chunks.map { chunk ->
    async {
        downloadChunkSafe(url, chunk.id)
    }
}.awaitAll()

// Проверяем, есть ли ошибки
val failedChunks = results.filter { it.isFailure }

if (failedChunks.isNotEmpty()) {
    println("Загрузка завершена с ошибками. Не удалось скачать ${failedChunks.size} фрагментов.")
    // Здесь можно реализовать логику сохранения состояния для докачки при следующем запуске
} else {
    println("Все фрагменты успешно скачаны!")
    // Переходим к слиянию файлов (Zero-Copy)
}

Использование runCatching и Result — это паттерн функционального программирования. Мы перестаем использовать исключения для управления потоком выполнения программы (Control Flow) и начинаем работать с ошибками как с обычными данными. Это делает архитектуру консольного приложения предсказуемой и легко тестируемой.

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

Отображение прогресса загрузки в консоли

В предыдущих этапах разработки мы создали мощный движок: программа умеет получать метаданные, делить файл на фрагменты, скачивать их параллельно с учетом таймаутов и объединять на диске с помощью Zero-Copy. Однако сейчас наш менеджер загрузок работает абсолютно «молча». Вы запускаете команду, консоль замирает на несколько минут, и затем файл появляется на диске.

Для пользователя (и для вас как разработчика) это выглядит как зависание. Непонятно, идет ли загрузка, с какой скоростью, и сколько времени осталось ждать. Профессиональное консольное приложение (CLI) обязано предоставлять качественную обратную связь.

Анатомия консольного вывода: магия возврата каретки

Исторически терминалы произошли от электромеханических печатных машинок (телетайпов). Когда вы печатаете текст, печатающая головка движется слева направо. Чтобы начать новую строку, нужно выполнить два физических действия:

1. Опустить барабан на одну линию вниз (Line Feed, \n).
2. Вернуть печатающую головку в самое начало строки (Carriage Return, \r).

В современных операционных системах функция println() автоматически добавляет символ новой строки (в Linux/macOS это \n, в Windows — \r\n). Это заставляет курсор перепрыгнуть ниже. Если мы будем выводить прогресс через println(), консоль быстро заполнится сотнями строк, что сделает интерфейс нечитаемым.

Секрет создания динамического интерфейса в консоли заключается в использовании символа возврата каретки \r без перехода на новую строку.

Символ \r приказывает терминалу: «перемести курсор в нулевую позицию текущей строки, но не спускайся вниз». Следующий вызов функции print() начнет печатать текст поверх старого, затирая его.

Проблема «призрачных символов»

При перезаписи строк возникает классическая проблема консольного программирования. Представьте, что вы вывели строку: Загрузка: 1000 КБ

Затем скорость упала, и вы хотите вывести новую строку поверх старой: Ошибка: 0 КБ

Если вы просто используете \r и напечатаете новую строку, в консоли останется: Ошибка: 0 КБ0 КБ

Почему это произошло? Вторая строка короче первой. Она перезаписала только первые 12 символов, а хвост от предыдущей строки (0 КБ) остался нетронутым. Терминал не очищает строку автоматически при возврате каретки.

Решение: всегда добивать выводимую строку пробелами до фиксированной ширины терминала (обычно 80 или 100 символов), либо использовать ANSI-escape последовательность \u001B[K, которая принудительно очищает строку от курсора до конца экрана.

Математика прогресс-бара

Визуальный прогресс-бар — это строка фиксированной длины, состоящая из двух типов символов: «заполненных» (отражающих скачанную часть) и «пустых» (отражающих оставшуюся часть).

Для расчета количества заполненных символов используется пропорция:

$$C_{filled} = \lfloor \frac{B_{current}}{B_{total}} \times W \rfloor$$

Где:

• $C_{filled}$ — количество символов, которые нужно закрасить.
• $B_{current}$ — количество скачанных байт на данный момент.
• $B_{total}$ — общий размер файла в байтах.
• $W$ — ширина прогресс-бара в символах (например, 50).
• $\lfloor x \rfloor$ — математическая операция округления вниз до целого числа.

Пример: мы скачали 250 МБ из 1000 МБ. Ширина бара — 50 символов. $C_{filled} = (250 / 1000) \times 50 = 0.25 \times 50 = 12.5$. Округляем вниз — получаем 12 заполненных символов и 38 пустых.

Реализация генератора строки на Kotlin

Для создания строки прогресс-бара в Kotlin эффективнее всего использовать класс StringBuilder или конструктор класса String, принимающий массив символов. Конкатенация строк через оператор + в цикле будет создавать множество лишних объектов в памяти, нагружая сборщик мусора (Garbage Collector).

Для современного внешнего вида мы будем использовать Unicode-символы: █ (U+2588) для заполненной части и ░ (U+2591) для пустой.

fun buildProgressBar(downloaded: Long, total: Long, width: Int = 50): String {
    // Защита от деления на ноль, если размер файла неизвестен
    if (total <= 0) return "[ Размер неизвестен ]"

    // Вычисляем процент завершения (от 0.0 до 1.0)
    val progressFraction = downloaded.toDouble() / total.toDouble()

    // Вычисляем количество заполненных символов
    val filledCount = (progressFraction * width).toInt().coerceIn(0, width)
    val emptyCount = width - filledCount

    // Вычисляем процент для текстового отображения
    val percent = (progressFraction * 100).toInt()

    // Формируем строку: [██████░░░░] 60%
    val filledStr = "█".repeat(filledCount)
    val emptyStr = "░".repeat(emptyCount)

    return "[$filledStr$emptyStr] $percent%"
}

Архитектура UI-цикла: изоляция потоков

Самая частая ошибка новичков в многопоточном программировании — заставить рабочие потоки (корутины, которые качают чанки) напрямую выводить свой прогресс в консоль.

Представьте: 8 корутин одновременно скачивают данные. Если каждая из них после получения очередного пакета байт вызовет print("\rПрогресс..."), произойдет состояние гонки (Race Condition) на уровне стандартного потока вывода (stdout). Строки перемешаются, консоль начнет мерцать, а курсор будет хаотично прыгать. Кроме того, операции ввода-вывода в консоль (I/O) достаточно медленные и будут тормозить саму загрузку сети.

Правильный архитектурный паттерн — выделенный UI-поток (UI Loop).

Рабочие корутины не знают о существовании консоли. Их единственная задача — качать байты и обновлять глобальный счетчик скачанных данных в оперативной памяти.

Отдельная корутина (UI-корутина) просыпается через равные промежутки времени (например, каждые 100 миллисекунд), считывает значение этого глобального счетчика, формирует красивую строку и выводит её в консоль. Это классическое разделение ответственности (Separation of Concerns).

Безопасный глобальный счетчик

Поскольку 8 рабочих корутин будут одновременно прибавлять скачанные байты к одной переменной, нам нужен потокобезопасный механизм. Как мы обсуждали в статье про синхронизацию, для простых числовых счетчиков идеально подходит AtomicLong из пакета java.util.concurrent.atomic.

Он использует процессорные инструкции CAS (Compare-And-Swap), что делает инкремент невероятно быстрым и не блокирует системные потоки.

import java.util.concurrent.atomic.AtomicLong

// Глобальное состояние загрузки
val totalBytesDownloaded = AtomicLong(0)

// Внутри рабочей корутины, скачивающей чанк:
suspend fun downloadChunk(url: String, start: Long, end: Long) {
    // ... настройка HTTP клиента ...
    val channel = response.bodyAsChannel()
    val buffer = ByteArray(8 * 1024) // 8 KB буфер

    while (!channel.isClosedForRead) {
        val bytesRead = channel.readAvailable(buffer)
        if (bytesRead > 0) {
            // Записываем на диск...
            fileChannel.write(ByteBuffer.wrap(buffer, 0, bytesRead), currentPosition)

            // БЕЗОПАСНО обновляем глобальный счетчик
            totalBytesDownloaded.addAndGet(bytesRead.toLong())
        }
    }
}

Расчет скорости и оставшегося времени (ETA)

Просто полоски прогресса недостаточно. Пользователь хочет знать скорость загрузки и Estimated Time of Arrival (ETA) — расчетное время до окончания.

Вычисление скорости

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

$$Speed = \frac{B_{current} - B_{previous}}{T_{current} - T_{previous}}$$

Если наша UI-корутина обновляет экран ровно раз в 1 секунду (1000 мс), то скорость в байтах в секунду равна просто разнице между текущим значением счетчика и значением секунду назад.

Однако, сеть нестабильна. В одну секунду может скачаться 5 МБ, в следующую — 0 МБ, затем снова 5 МБ. Если выводить скорость «как есть», цифры будут дико скакать, раздражая пользователя. В системном программировании для сглаживания таких скачков применяют алгоритм скользящего среднего (Moving Average). Мы сохраняем историю скоростей за последние 3-5 секунд и выводим их среднее арифметическое.

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

Вычисление ETA

Зная текущую скорость, мы можем предсказать, сколько времени потребуется на скачивание оставшихся байт:

$$ETA = \frac{B_{total} - B_{current}}{Speed}$$

Где:

• $ETA$ — оставшееся время в секундах.
• $B_{total} - B_{current}$ — сколько байт осталось скачать.
• $Speed$ — текущая скорость в байтах в секунду.

Пример: Осталось скачать 150 000 000 байт (около 150 МБ). Текущая скорость — 5 000 000 байт/сек (5 МБ/с). $ETA = 150 000 000 / 5 000 000 = 30$ секунд.

Форматирование данных для человека

Выводить в консоль «Скорость: 15432981 байт/сек» бессмысленно — человеческий мозг плохо воспринимает длинные ряды цифр. Нам нужна утилитарная функция, которая переводит байты в Килобайты, Мегабайты или Гигабайты.

В IT исторически сложилось два стандарта: десятичный (1 КБ = 1000 байт) и двоичный (1 КиБ = 1024 байта). Операционные системы семейства Windows используют двоичную систему (деление на 1024), её же мы применим в нашем менеджере.

fun formatBytes(bytes: Long): String {
    if (bytes < 1024) return "$bytes B"
    val exp = (Math.log(bytes.toDouble()) / Math.log(1024.0)).toInt()
    val prefix = "KMGTPE"[exp - 1]
    val value = bytes / Math.pow(1024.0, exp.toDouble())
    // Форматируем до двух знаков после запятой
    return String.format("%.2f %cB", value, prefix)
}

fun formatTime(seconds: Long): String {
    if (seconds < 0) return "--:--"
    val h = seconds / 3600
    val m = (seconds % 3600) / 60
    val s = seconds % 60
    return if (h > 0) {
        String.format("%02d:%02d:%02d", h, m, s)
    } else {
        String.format("%02d:%02d", m, s)
    }
}

Сборка UI-корутины

Теперь объединим все компоненты в единый механизм. Мы запустим отдельную корутину с помощью билдера launch, которая будет работать в бесконечном цикле while(isActive), пока идет загрузка.

import kotlinx.coroutines.*
import java.util.concurrent.atomic.AtomicLong

suspend fun startDownload(url: String, totalSize: Long) = coroutineScope {
    val totalBytesDownloaded = AtomicLong(0)
    val startTime = System.currentTimeMillis()

    // 1. Запускаем UI-корутину в фоновом режиме
    val uiJob = launch(Dispatchers.Default) {
        var previousDownloaded = 0L

        while (isActive) { // Проверяем, не отменена ли корутина
            val currentDownloaded = totalBytesDownloaded.get()

            // Вычисляем мгновенную скорость (за последние 500 мс)
            val bytesInHalfSecond = currentDownloaded - previousDownloaded
            val speedBps = bytesInHalfSecond * 2 // Умножаем на 2, чтобы получить байты в секунду
            previousDownloaded = currentDownloaded

            // Вычисляем ETA
            val etaSeconds = if (speedBps > 0) {
                (totalSize - currentDownloaded) / speedBps
            } else {
                -1L
            }

            // Формируем компоненты строки
            val bar = buildProgressBar(currentDownloaded, totalSize, 40)
            val speedStr = "${formatBytes(speedBps)}/s"
            val downloadedStr = "${formatBytes(currentDownloaded)} / ${formatBytes(totalSize)}"
            val timeStr = "ETA: ${formatTime(etaSeconds)}"

            // Собираем итоговую строку.
            // \u001B[K - ANSI код для очистки остатка строки от мусора
            val output = "\r$bar | $downloadedStr | $speedStr | $timeStr\u001B[K"

            // Выводим в консоль без перехода на новую строку
            print(output)

            // Засыпаем на 500 миллисекунд
            delay(500)
        }
    }

    // 2. Запускаем рабочие корутины для скачивания чанков
    // (Код из предыдущих статей, использующий async и awaitAll)
    // ...

    // 3. Когда скачивание завершено, останавливаем UI-корутину
    uiJob.cancel()

    // 4. Выводим финальную строку со 100% прогрессом и переходом на новую строку
    val totalTime = (System.currentTimeMillis() - startTime) / 1000
    val finalBar = buildProgressBar(totalSize, totalSize, 40)
    println("\r$finalBar | Успешно скачано за ${formatTime(totalTime)}\u001B[K")
}

В этом коде мы используем Dispatchers.Default для UI-корутины, так как форматирование строк и математические вычисления — это задачи, нагружающие процессор (CPU-bound), а не сеть.

Использование ANSI-кода \u001B[K в конце строки элегантно решает проблему «призрачных символов», о которой мы говорили в начале. Если новая строка окажется короче предыдущей (например, скорость упала с 10.5 MB/s до 9.2 MB/s), терминал автоматически сотрет лишние символы справа от курсора.

Реализовав этот механизм, вы превратите безликий скрипт в полноценный системный инструмент. Пользователь будет видеть плавную анимацию загрузки, точную скорость и время, а архитектура приложения останется надежной и потокобезопасной благодаря разделению логики скачивания и отрисовки интерфейса.

Приостановка и возобновление загрузки

Представьте ситуацию: вы скачиваете образ операционной системы размером 10 ГБ. Прогресс-бар, который мы реализовали на прошлом этапе, показывает 95%. Внезапно происходит сбой в сети, или вы случайно нажимаете комбинацию Ctrl+C в терминале. Программа мгновенно завершается. При повторном запуске загрузка начинается с нуля. В мире системного программирования и качественных CLI-инструментов такое поведение недопустимо.

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

Анатомия прерывания: сигналы операционной системы

Когда пользователь работает в консоли и нажимает Ctrl+C, операционная система не просто «убивает» процесс. Она отправляет ему специальный сигнал прерывания — SIGINT (Signal Interrupt).

По умолчанию, когда JVM (Java Virtual Machine, на которой работает Kotlin) получает SIGINT, она немедленно останавливает выполнение всех потоков и завершает программу. Данные, находящиеся в оперативной памяти (RAM), безвозвратно теряются. Буферы, которые не успели записаться на диск, очищаются.

Чтобы предотвратить потерю данных, нам необходимо перехватить этот сигнал. В экосистеме JVM для этого используется механизм Shutdown Hooks (хуки завершения работы). Это специальные потоки, которые JVM запускает непосредственно перед своим выключением.

Runtime.getRuntime().addShutdownHook(Thread {
    println("\nПолучен сигнал завершения. Сохранение состояния...")
    // Здесь будет логика сохранения прогресса
})

Важно понимать: код внутри Shutdown Hook должен выполняться максимально быстро. Если он зависнет (например, пытаясь выполнить долгий сетевой запрос), операционная система принудительно завершит процесс через сигнал SIGKILL (который перехватить невозможно).

Управление состоянием: от RAM к диску

В предыдущих статьях мы хранили информацию о скачанных байтах в оперативной памяти с помощью атомарного счетчика AtomicLong. Для возобновления загрузки этого недостаточно. Нам нужно знать точное состояние каждого отдельного фрагмента (чанка).

Проектирование структуры данных

Состояние загрузки — это слепок данных, описывающий текущий прогресс. Для его хранения на диске отлично подходит формат JSON. Он легко читается человеком (что полезно для отладки) и без проблем сериализуется стандартными библиотеками, такими как kotlinx.serialization.

Создадим доменную модель состояния:

import kotlinx.serialization.Serializable

@Serializable
data class DownloadState(
    val url: String,
    val totalSizeBytes: Long,
    val chunks: List<ChunkState>
)

@Serializable
data class ChunkState(
    val id: Int,
    val startByte: Long,
    val endByte: Long,
    var downloadedBytes: Long // Изменяемое поле
)

Каждый раз, когда мы начинаем скачивание нового файла ubuntu.iso, наша программа должна создавать рядом скрытый файл состояния, например .ubuntu.iso.info.

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

Возникает вопрос: когда именно записывать этот JSON на диск?

Если мы будем обновлять файл на диске после каждого скачанного килобайта, мы создадим колоссальную нагрузку на подсистему ввода-вывода (I/O). Диск станет узким местом (bottleneck), и скорость скачивания резко упадет.

Оптимальный подход — комбинированный:

1. Периодическое сохранение: отдельная корутина раз в 2-3 секунды сбрасывает текущее состояние из RAM на диск.
2. Экстренное сохранение: при получении сигнала SIGINT (через Shutdown Hook) мы делаем финальную запись на диск перед выходом.

Математика возобновления: пересчет диапазонов

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

Главная задача — правильно сформировать HTTP-заголовок Range для каждого чанка. Нам больше не нужно скачивать чанк целиком. Нам нужен только остаток.

Формула расчета оставшихся байт для конкретного чанка:

$$B_{remaining} = B_{end} - (B_{start} + B_{downloaded}) + 1$$

Где:

• $B_{remaining}$ — количество байт, которое осталось скачать в этом фрагменте.
• $B_{end}$ — конечный байт фрагмента (включительно).
• $B_{start}$ — начальный байт фрагмента.
• $B_{downloaded}$ — количество уже скачанных и сохраненных на диск байт для этого фрагмента.
• $+ 1$ — компенсация, так как диапазоны в HTTP указываются включительно (например, байты 0-99 — это 100 байт).

Пример: Чанк должен скачать данные с 1000 по 1999 байт. В прошлый раз мы успели скачать 400 байт. Новый стартовый байт для запроса будет равен $1000 + 400 = 1400$. Мы отправим серверу заголовок: Range: bytes=1400-1999.

Интеграция с корутинами: безопасная отмена

В статье про структурированную конкурентность мы обсуждали иерархию корутин. Когда срабатывает Shutdown Hook, мы должны корректно остановить все сетевые запросы.

В Kotlin для этого используется метод cancel() у объекта Job или CoroutineScope. Под капотом этот метод выбрасывает специальное исключение CancellationException внутри всех дочерних корутин.

import kotlinx.coroutines.*
import java.io.File

// Глобальный scope для нашего приложения
val appScope = CoroutineScope(Dispatchers.IO + SupervisorJob())

fun setupShutdownHook(stateFile: File, state: DownloadState) {
    Runtime.getRuntime().addShutdownHook(Thread {
        // 1. Отменяем все корутины (останавливаем скачивание)
        appScope.cancel()

        // 2. Сохраняем финальное состояние на диск
        val json = encodeToJson(state)
        stateFile.writeText(json)

        println("\nЗагрузка приостановлена. Состояние сохранено.")
    })
}

Проблема прерванного буфера

При отмене корутины есть тонкий нюанс. Представьте, что корутина считала из сети 8 КБ данных в буфер, но не успела записать их на диск (вызвался cancel()). Если мы просто прибавим эти 8 КБ к downloadedBytes в нашем состоянии, при следующем запуске в файле образуется «дыра» — нули вместо реальных данных.

Правило надежного I/O: обновлять счетчик скачанных байт в объекте состояния нужно строго после успешной записи буфера на диск (вызова метода write у FileChannel).

// Внутри рабочей корутины:
while (isActive) { // Проверяем, не отменили ли нас
    val bytesRead = channel.readAvailable(buffer)
    if (bytesRead <= 0) break

    // 1. Сначала пишем на диск
    fileChannel.write(ByteBuffer.wrap(buffer, 0, bytesRead), currentPosition)

    // 2. Только потом обновляем состояние
    chunkState.downloadedBytes += bytesRead
    currentPosition += bytesRead
}

Проверка целостности: изменился ли файл на сервере?

Существует неочевидный граничный случай (edge case). Вы начали качать файл в понедельник, остановили загрузку, а во вторник автор файла обновил его на сервере. Если вы продолжите загрузку в среду, вы скачаете половину старого файла и половину нового. Итоговый файл будет поврежден (хэш-сумма не совпадет).

Чтобы избежать этого, протокол HTTP предоставляет механизмы валидации кэша. При первом запросе (когда мы получаем метаданные) сервер часто возвращает два важных заголовка:

1. ETag (Entity Tag) — уникальный идентификатор версии файла (часто это хэш).
2. Last-Modified — дата последнего изменения файла на сервере.

При проектировании менеджера загрузок мы должны сохранить эти значения в наш JSON-файл состояния DownloadState.

При попытке возобновить загрузку, мы обязаны отправить HEAD запрос и сверить текущие ETag и Last-Modified сервера с теми, что сохранены у нас. Если они отличаются — файл был изменен. В этом случае возобновление невозможно, старые данные и файл состояния необходимо удалить, а загрузку начать с самого начала, уведомив об этом пользователя.

Сборка алгоритма возобновления

Теперь мы можем описать полный жизненный цикл нашего консольного приложения с поддержкой докачки:

1. Инициализация: Пользователь вводит команду download http://.../file.zip.
2. Поиск состояния: Программа ищет файл .file.zip.info в текущей директории.
3. Ветвление логики:
   • Если файла состояния нет: Выполняем HEAD запрос, получаем размер, делим на чанки, создаем новый объект DownloadState, выделяем место на диске.
   • Если файл состояния есть: Загружаем JSON. Выполняем HEAD запрос для проверки ETag. Если файл не изменился, фильтруем чанки (отбрасываем те, где $B_{downloaded} == B_{end} - B_{start} + 1$). Для оставшихся пересчитываем стартовый байт.
4. Запуск: Регистрируем Shutdown Hook. Запускаем корутины для каждого незавершенного чанка с новыми заголовками Range.
5. Мониторинг: Запускаем UI-корутину для отрисовки прогресс-бара и корутину для периодического сохранения JSON на диск.
6. Завершение: Когда все чанки скачаны, удаляем файл состояния .info, так как он больше не нужен, и выводим сообщение об успехе.

Реализация этого механизма превращает простую «качалку» в отказоустойчивый системный инструмент, способный пережить обрывы связи, перезагрузки компьютера и ручные прерывания пользователем.

Очередь и управление несколькими загрузками

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

Операционная система имеет жесткие лимиты на количество одновременно открытых сетевых соединений (сокетов). Ваш домашний роутер может не справиться с резким скачком тысяч мелких пакетов и начнет их терять. А жесткий диск (особенно если это классический HDD) физически не сможет записывать данные в пятьдесят разных мест одновременно — магнитная головка будет метаться по блинам диска, что приведет к падению скорости записи практически до нуля.

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

Проблема неконтролируемой конкурентности

В предыдущих статьях мы научились разбивать один файл на части и скачивать их параллельно. Для этого мы использовали корутины и функцию launch. Самый наивный подход к скачиванию списка файлов выглядит так:

val urls = listOf("http://.../file1", "http://.../file2", /* еще 48 ссылок */)

// НАИВНЫЙ И ОПАСНЫЙ ПОДХОД
urls.forEach { url ->
    launch(Dispatchers.IO) {
        downloadFile(url)
    }
}

Этот код мгновенно создаст 50 корутин. Каждая из них попытается установить HTTP-соединение, получить метаданные, разбить свой файл на чанки (допустим, по 8 чанков на файл) и запустить еще по 8 корутин. Итого: 400 одновременно работающих сетевых потоков. Это классический пример исчерпания ресурсов (Resource Exhaustion).

Исчерпание ресурсов — это ситуация в системном программировании, когда программа запрашивает у операционной системы больше вычислительных мощностей, памяти или лимитов ввода-вывода, чем доступно физически, что приводит к деградации производительности или аварийному завершению (крашу).

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

В программировании эта концепция реализуется через структуру данных «Очередь» и архитектурный паттерн Пул воркеров (Worker Pool).

Каналы (Channels): потокобезопасная очередь в Kotlin

Для организации очереди задач в многопоточной среде нельзя использовать обычные списки вроде ArrayList. Если две корутины одновременно попытаются взять из обычного списка один и тот же URL, возникнет состояние гонки (Race Condition), и обе корутины начнут качать один и тот же файл.

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

Специально для корутин в Kotlin был создан Канал (Channel).

Канал — это примитив синхронизации, который концептуально похож на трубу: с одной стороны в нее кладут данные (отправители), а с другой стороны их забирают (получатели). Главная особенность канала в том, что его операции приостанавливают корутину (suspend), а не блокируют поток ОС.

• Если канал пуст, корутина-получатель «засыпает» до тех пор, пока в канале не появятся данные.
• Если канал переполнен (имеет лимит емкости), корутина-отправитель «засыпает», пока в канале не освободится место.

Создадим канал, который будет хранить ссылки на файлы:

import kotlinx.coroutines.channels.Channel

// Создаем канал с бесконечной емкостью (буфером)
// В него мы будем складывать объекты наших задач
val downloadQueue = Channel<String>(Channel.UNLIMITED)

Архитектурный паттерн Worker Pool

Теперь, когда у нас есть труба с задачами, нам нужны «кассиры», которые будут эти задачи обрабатывать. В терминологии системного программирования они называются воркерами (Workers).

Пул воркеров — это фиксированное количество постоянно работающих корутин, которые в бесконечном цикле слушают очередь. Как только в очереди появляется задача, свободный воркер забирает ее, выполняет, а затем снова обращается к очереди за следующей.

Давайте реализуем этот паттерн на Kotlin. Допустим, мы хотим, чтобы одновременно скачивалось не более трех файлов.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel

val MAX_CONCURRENT_DOWNLOADS = 3

suspend fun processDownloads(urls: List<String>) = coroutineScope {
    val queue = Channel<String>(Channel.UNLIMITED)

    // 1. Заполняем очередь задачами
    for (url in urls) {
        queue.send(url)
    }
    // Закрываем канал для отправки (больше ссылок не будет)
    queue.close()

    // 2. Создаем Пул Воркеров
    repeat(MAX_CONCURRENT_DOWNLOADS) { workerId ->
        launch(Dispatchers.IO) {
            // Воркер читает канал, пока тот не опустеет и не закроется
            for (url in queue) {
                println("Воркер $workerId начал загрузку: $url")
                downloadFile(url) // Наша suspend-функция скачивания
                println("Воркер $workerId завершил загрузку: $url")
            }
        }
    }
}

Разберем этот код по шагам:

1. Мы создаем канал queue.
2. Циклом for отправляем в него все ссылки через метод send().
3. Вызываем queue.close(). Это критически важный шаг. Он говорит каналу: «Новых данных больше не предвидится». Без этого воркеры будут бесконечно ждать новых задач и программа никогда не завершится.
4. Функция repeat(3) запускает ровно три корутины.
5. Внутри каждой корутины используется цикл for (url in queue). Этот цикл под капотом безопасно извлекает элементы из канала. Если два воркера одновременно обратятся к каналу, внутренние механизмы Kotlin гарантируют, что они получат разные ссылки.

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

Изоляция сбоев: SupervisorJob

В статье про основы корутин мы изучали структурированную конкурентность (Structured Concurrency). Ее базовое правило гласит: если одна дочерняя корутина завершается с ошибкой (выбрасывает исключение), родительская корутина отменяет все остальные дочерние корутины.

Для скачивания одного файла по частям это идеальное поведение. Если один чанк не удалось скачать (например, сервер оборвал соединение), нет смысла докачивать остальные чанки — файл все равно будет битым. Загрузка должна прерваться целиком.

Но при скачивании очереди независимых файлов это правило становится разрушительным. Представьте: вы скачиваете 50 файлов. 49 из них доступны, а один удален с сервера (ошибка 404 Not Found). Если воркер выбросит исключение на этом файле, стандартный coroutineScope мгновенно отменит загрузку остальных 49 файлов!

Чтобы изменить это поведение, в Kotlin существует SupervisorJob (Супервизор).

SupervisorJob — это специальный элемент контекста корутины, который меняет направление распространения ошибок. При его использовании ошибка в одной дочерней корутине не приводит к отмене остальных.

Сравним поведение:

• Обычный Job: Ошибка в задаче А убивает задачи Б и В.
• SupervisorJob: Ошибка в задаче А убивает только задачу А. Задачи Б и В продолжают работу.

Для применения супервизора мы заменим coroutineScope на supervisorScope:

suspend fun processDownloadsSafely(urls: List<String>) = supervisorScope {
    val queue = Channel<String>(Channel.UNLIMITED)
    urls.forEach { queue.send(it) }
    queue.close()

    repeat(MAX_CONCURRENT_DOWNLOADS) { workerId ->
        launch(Dispatchers.IO) {
            for (url in queue) {
                try {
                    downloadFile(url)
                } catch (e: Exception) {
                    // Перехватываем ошибку конкретного файла
                    System.err.println("Ошибка при скачивании $url: ${e.message}")
                    // Воркер не умирает, он пойдет за следующей задачей в очередь!
                }
            }
        }
    }
}

Обратите внимание на блок try-catch внутри цикла воркера. Даже при использовании supervisorScope, необработанное исключение «убьет» саму корутину-воркера. Если у нас 3 воркера, и 3 файла подряд выдадут ошибку, все воркеры умрут, и очередь зависнет навсегда. Оборачивая downloadFile в try-catch, мы гарантируем, что воркер выживет, зафиксирует ошибку в лог и возьмет из канала следующую ссылку.

Глобальный прогресс: математика и агрегация состояний

Когда мы качали один файл, мы выводили в консоль один прогресс-бар. Теперь у нас одновременно качаются три файла, а остальные ждут в очереди. Как отобразить это пользователю в консоли?

Выводить три прыгающих прогресс-бара в стандартный вывод (stdout) терминала очень сложно — строки будут перекрывать друг друга, создавая визуальный мусор. В профессиональных CLI-утилитах (например, docker pull или npm install) часто используют агрегированный (общий) прогресс-бар для всей очереди.

Для этого нам нужно вычислять общий процент завершения всех задач.

Математическая формула расчета глобального прогресса в процентах выглядит так:

$$P_{total} = \frac{\sum_{i=1}^{n} D_i}{\sum_{i=1}^{n} S_i} \times 100$$

Где:

• $P_{total}$ — общий прогресс в процентах (от 0 до 100).
• $n$ — общее количество файлов в очереди.
• $D_i$ — количество уже скачанных байт для файла под номером $i$.
• $S_i$ — общий размер файла под номером $i$ в байтах.
• $\sum$ — математический знак суммы (означает, что нужно сложить значения для всех файлов).

Пример из жизни: Вы качаете 3 файла размером 100 МБ, 200 МБ и 50 МБ. Общий размер (сумма $S_i$) = 350 МБ. На данный момент скачано: 100 МБ первого файла (он завершен), 40 МБ второго и 0 МБ третьего (он еще в очереди). Сумма скачанного (сумма $D_i$) = 140 МБ. Считаем: $P_{total} = (140 / 350) \times 100 = 40\%$.

Реализация единого источника истины

Чтобы UI-корутина могла раз в полсекунды вычислять эту формулу, ей нужен доступ к состояниям всех загрузок. Создадим потокобезопасное хранилище на базе ConcurrentHashMap.

import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicLong

// Класс для хранения состояния одного файла
class FileProgress(val totalSize: Long) {
    val downloaded = AtomicLong(0)
}

// Глобальное хранилище прогресса по всем URL
val globalProgressMap = ConcurrentHashMap<String, FileProgress>()

Когда воркер берет задачу, он сначала делает HTTP-запрос HEAD, узнает размер файла и кладет его в globalProgressMap. По мере скачивания чанков, он обновляет счетчик downloaded.

А выделенная UI-корутина (о которой мы говорили в статье про отображение прогресса) просто бежит по всем значениям этой мапы:

launch(Dispatchers.Default) {
    while (isActive) {
        var totalDownloaded = 0L
        var totalSize = 0L

        // Суммируем данные по всем известным файлам
        for (progress in globalProgressMap.values) {
            totalDownloaded += progress.downloaded.get()
            totalSize += progress.totalSize
        }

        if (totalSize > 0) {
            val percent = (totalDownloaded.toDouble() / totalSize) * 100
            print("\rОбщий прогресс: ${"%.1f".format(percent)}% ($totalDownloaded / $totalSize байт)")
        }

        delay(500) // Обновляем консоль дважды в секунду
    }
}

Такая архитектура полностью разделяет бизнес-логику (воркеры качают данные и обновляют атомарные счетчики) и слой представления (UI-корутина только читает данные и рисует текст). Это исключает блокировки интерфейса и делает код легко тестируемым.

Внедрение очереди и пула воркеров превращает ваш менеджер загрузок из простой «качалки» в масштабируемую систему, способную надежно переваривать списки из тысяч файлов, не перегружая компьютер пользователя и сеть.

Тестирование конкурентного кода

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

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

Главный враг тестирования многопоточных программ — недетерминированность (Non-determinism). Это свойство системы, при котором один и тот же код с одними и теми же входными данными может выдавать разные результаты при каждом новом запуске. Это происходит потому, что операционная система переключает контекст между потоками в случайные моменты времени. Ошибка состояния гонки (Race Condition) может проявить себя один раз на миллион запусков.

Стандартные инструменты тестирования (например, обычные функции JUnit) созданы для синхронного, детерминированного кода. Если мы попытаемся протестировать наши корутины обычным способом, тест либо завершится раньше, чем корутины успеют выполниться, либо зависнет навсегда.

Виртуальное время и runTest

Для тестирования корутин в Kotlin существует специализированная библиотека kotlinx-coroutines-test. Ее фундаментом является функция-строитель runTest.

В обычном приложении мы используем runBlocking, чтобы удержать консольную программу от завершения, пока работают фоновые задачи. В тестах мы используем runTest. Главное отличие заключается в том, как они работают со временем.

Если в коде есть функция приостановки delay(5000), обычный runBlocking честно заблокирует поток на 5 секунд. Если у вас 100 таких тестов, прогон всего набора займет почти 10 минут. Это недопустимо для процесса разработки.

Функция runTest использует механизм виртуального времени (Virtual Time). Она подменяет системные часы внутри корутин. Когда код вызывает delay(5000), runTest не ждет реальные 5 секунд. Он мгновенно перематывает внутренний счетчик времени вперед и сразу же возобновляет выполнение корутины.

Автоматический переход времени для обычных функций приостановки... Немедленно выполнить тела запуска или асинхронные блоки кода.

fandroid.info

Математически это можно выразить так: если $T_{real}$ — это реальное время выполнения теста, а $T_{virtual}$ — суммарное время всех задержек в коде, то при использовании runTest всегда выполняется условие $T_{real}

import kotlinx.coroutines.delay
import kotlinx.coroutines.test.runTest
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Test
import kotlin.time.Duration.Companion.seconds

class DownloadTimeoutTest {

    suspend fun simulateNetworkDelay() {
        delay(5000) // Имитация долгого ответа сервера
    }

    @Test
    fun testNetworkDelay() = runTest {
        val startTime = currentTime // Виртуальное время (0 мс)

        simulateNetworkDelay()

        val endTime = currentTime

        // Тест пройдет успешно, и выполнится за миллисекунды реального времени!
        assertEquals(5000, endTime - startTime)
    }
}

Важный подводный камень: runTest перематывает только корутинный delay. Если внутри вашего кода используется блокирующий вызов операционной системы, например Thread.sleep(5000) или синхронное чтение из сокета, тест честно зависнет на 5 секунд. Именно поэтому в системном программировании на Kotlin критически важно использовать только неблокирующие (suspend) функции.

Внедрение зависимостей для диспетчеров

В предыдущих статьях мы часто писали код вида launch(Dispatchers.IO) { ... }. Для рабочего приложения это правильно — мы явно указываем, что сетевые операции должны выполняться в пуле потоков, оптимизированном для ввода-вывода.

Но для тестирования это архитектурная катастрофа. Жестко зашитый (hardcoded) диспетчер лишает нас контроля над выполнением кода в тесте. Мы не сможем управлять виртуальным временем для этих корутин, потому что Dispatchers.IO ничего не знает про runTest.

Чтобы код был тестируемым, необходимо применять принцип Внедрения зависимостей (Dependency Injection). Мы должны передавать диспетчер в наши классы и функции извне.

Сравним два подхода:

Плохой подход (невозможно протестировать):

class DownloadManager {
    suspend fun start() = coroutineScope {
        // Жесткая привязка к Dispatchers.IO
        launch(Dispatchers.IO) {
            downloadFile()
        }
    }
}

Хороший подход (тестируемый):

import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.Dispatchers

// По умолчанию используем IO, но в тестах сможем передать другой диспетчер
class DownloadManager(private val ioDispatcher: CoroutineDispatcher = Dispatchers.IO) {
    suspend fun start() = coroutineScope {
        launch(ioDispatcher) {
            downloadFile()
        }
    }
}

В тестах вместо Dispatchers.IO мы будем передавать специальные тестовые диспетчеры. Библиотека предоставляет два основных варианта:

1. StandardTestDispatcher — помещает новые корутины в очередь и не выполняет их, пока мы явно не попросим (например, вызвав advanceUntilIdle()). Это дает полный ручной контроль над порядком выполнения.
2. UnconfinedTestDispatcher — запускает новые корутины жадно (eagerly), немедленно в текущем потоке. Это удобно для простых тестов, где нам не важен порядок переключения контекста.

Тестирование пула воркеров и очереди

Теперь применим эти знания для тестирования логики пула воркеров, которую мы разработали для управления несколькими загрузками. Наша цель — убедиться, что если мы передаем 5 ссылок, а лимит воркеров равен 2, то одновременно скачиваются не более двух файлов.

Для этого мы создадим заглушку (Mock) для функции скачивания. Заглушка не будет ходить в реальную сеть (сеть нестабильна и сделает тесты хрупкими), она будет просто имитировать работу с помощью delay.

import kotlinx.coroutines.*
import kotlinx.coroutines.test.*
import org.junit.jupiter.api.Assertions.*
import org.junit.jupiter.api.Test
import java.util.concurrent.atomic.AtomicInteger

class WorkerPoolTest {

    @Test
    fun `should not exceed max concurrent downloads`() = runTest {
        // 1. Подготовка (Arrange)
        val maxWorkers = 2
        val urls = listOf("url1", "url2", "url3", "url4", "url5")

        // Счетчики для проверки состояния
        val activeDownloads = AtomicInteger(0)
        var maxObservedConcurrent = 0
        val completedDownloads = AtomicInteger(0)

        // Заглушка функции скачивания
        val mockDownload: suspend (String) -> Unit = { url ->
            val currentActive = activeDownloads.incrementAndGet()

            // Фиксируем максимальное количество одновременных загрузок
            synchronized(this) {
                if (currentActive > maxObservedConcurrent) {
                    maxObservedConcurrent = currentActive
                }
            }

            // Имитируем долгую загрузку (виртуальное время)
            delay(1000)

            activeDownloads.decrementAndGet()
            completedDownloads.incrementAndGet()
        }

        // 2. Действие (Act)
        // Передаем тестовый диспетчер (testScheduler) в нашу функцию
        val dispatcher = StandardTestDispatcher(testScheduler)

        launch(dispatcher) {
            // Вызываем функцию из предыдущей статьи, передавая заглушку
            processDownloadsSafely(urls, maxWorkers, mockDownload, dispatcher)
        }

        // Проматываем виртуальное время до завершения всех корутин
        advanceUntilIdle()

        // 3. Проверка (Assert)
        assertEquals(5, completedDownloads.get(), "Все файлы должны быть скачаны")
        assertEquals(2, maxObservedConcurrent, "Лимит воркеров не должен превышаться")
    }
}

В этом тесте мы используем AtomicInteger для безопасного подсчета активных корутин. Благодаря StandardTestDispatcher и advanceUntilIdle(), тест прогоняет сценарий, который в реальности занял бы 3 секунды (первые два файла — 1 сек, следующие два — 1 сек, последний — 1 сек), за несколько миллисекунд, при этом строго проверяя соблюдение лимита конкурентности.

Проверка отказоустойчивости (SupervisorJob)

Вторая важнейшая задача — проверить, что сбой при скачивании одного файла не обрушит всю очередь. Мы использовали SupervisorJob (через supervisorScope), и теперь обязаны доказать тестом, что он работает корректно.

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

nuancesprog.ru

В неструктурированной конкурентности (как в старой Java) ошибки теряются или приводят к утечкам. В Kotlin мы должны явно тестировать изоляцию ошибок.

Настроим нашу заглушку так, чтобы она выбрасывала исключение на конкретном URL:

@Test
fun `error in one download should not cancel others`() = runTest {
    val urls = listOf("good_url_1", "bad_url", "good_url_2")
    val completed = mutableListOf<String>()

    val mockDownload: suspend (String) -> Unit = { url ->
        if (url == "bad_url") {
            throw RuntimeException("Network Error 404")
        }
        delay(500)
        completed.add(url)
    }

    val dispatcher = StandardTestDispatcher(testScheduler)

    launch(dispatcher) {
        processDownloadsSafely(urls, maxWorkers = 2, mockDownload, dispatcher)
    }

    advanceUntilIdle()

    // Проверяем, что хорошие ссылки скачались, несмотря на ошибку посередине
    assertEquals(2, completed.size)
    assertTrue(completed.contains("good_url_1"))
    assertTrue(completed.contains("good_url_2"))
}

Если бы мы случайно использовали обычный coroutineScope вместо supervisorScope в реализации функции processDownloadsSafely, этот тест мгновенно бы упал, так как исключение от bad_url отменило бы загрузку good_url_2.

Lincheck: Тяжелая артиллерия для структур данных

Тестирование логики оркестрации (кто кого запускает) — это лишь половина дела. В нашем менеджере загрузок есть общее состояние — класс FileProgress, который хранит атомарные счетчики скачанных байт. Несколько потоков одновременно обновляют эти счетчики.

Как доказать, что наша структура данных потокобезопасна? Обычные тесты здесь бессильны, так как они не могут перебрать все возможные варианты переключения контекста процессором.

Для решения этой задачи в экосистеме Kotlin существует фреймворк Lincheck.

Lincheck - это фреймворк для языка Kotlin... который предоставляет инструментарий для тестирования структур данных на линеаризуемость (ну или если менее точно -- на потокобезопасность) и некоторые другие контракты.

tune-it.ru

Линеаризуемость (Linearizability) — это строгая математическая модель согласованности. Простыми словами: если несколько потоков одновременно вызывают методы объекта, результат должен быть таким, как если бы эти вызовы происходили строго последовательно, один за другим, без наложений.

Представьте банковский счет. Поток А кладет 100 долл., Поток Б кладет 50 долл. Если начальный баланс был 0, итоговый должен быть 150 долл. Если из-за состояния гонки баланс стал 100 долл. (Поток А затер данные Потока Б) — структура не линеаризуема.

Lincheck работает методом стресс-тестирования (Stress Testing) и проверки моделей (Model Checking). Он генерирует случайные сценарии вызовов методов из разных потоков, выполняет их тысячи раз, а затем анализирует граф состояний.

Напишем тест для нашего счетчика прогресса с использованием аннотаций Lincheck:

import org.jetbrains.kotlinx.lincheck.annotations.Operation
import org.jetbrains.kotlinx.lincheck.check
import org.jetbrains.kotlinx.lincheck.strategy.stress.StressOptions
import org.junit.jupiter.api.Test
import java.util.concurrent.atomic.AtomicLong

// Наша структура данных из предыдущей статьи
class FileProgress {
    private val downloaded = AtomicLong(0)

    // Аннотация @Operation говорит Lincheck, что этот метод нужно дергать из разных потоков
    @Operation
    fun addBytes(amount: Long) {
        downloaded.addAndGet(amount)
    }

    @Operation
    fun getProgress(): Long {
        return downloaded.get()
    }
}

class FileProgressLincheckTest {

    @Test
    fun stressTest() {
        // Запускаем стресс-тест: множество потоков будут хаотично вызывать addBytes и getProgress
        StressOptions()
            .actorsBefore(1) // Потоки до начала конкурентной фазы
            .threads(3)      // Количество конкурирующих потоков
            .actorsPerThread(3) // Операций на поток
            .actorsAfter(1)  // Потоки после завершения
            .iterations(100) // Количество прогонов всего сценария
            .check(FileProgress::class.java)
    }
}

Если бы мы реализовали FileProgress с использованием обычной переменной var downloaded: Long = 0 вместо AtomicLong, Lincheck мгновенно бы провалил тест. Он бы вывел в консоль конкретный сценарий переплетения потоков (Interleaving), который привел к потере байтов. Это бесценный инструмент для системного программиста, позволяющий находить баги, которые иначе проявились бы только у пользователей на продакшене при скачивании гигабайтных файлов.

Тестирование конкурентного кода требует изменения мышления. Вы больше не тестируете просто входные и выходные данные. Вы тестируете поведение системы во времени, реакцию на сбои соседей и устойчивость к хаосу планировщика операционной системы. Использование runTest для управления временем, внедрения зависимостей для диспетчеров и Lincheck для проверки структур данных делает ваш консольный менеджер загрузок по-настоящему надежным инженерным продуктом.

Оптимизация использования памяти и сети

Представьте, что вы построили идеальный водопровод. Трубы проложены ровно, насосы работают синхронно, а система управления безупречна. Но когда вы открываете главный вентиль на полную мощность, резервуар переполняется, а трубы начинают протекать под избыточным давлением.

В разработке системного программного обеспечения происходит то же самое. Мы спроектировали архитектуру менеджера загрузок, настроили пул воркеров и научились безопасно объединять фрагменты файлов. Наш код работает корректно. Но если мы попытаемся скачать файл размером 50 ГБ на слабом компьютере или при нестабильном мобильном интернете, приложение может аварийно завершиться или зависнуть.

Чтобы консольный менеджер загрузок стал по-настоящему профессиональным инструментом, необходимо научиться управлять двумя самыми ценными ресурсами компьютера: оперативной памятью (RAM) и сетевыми соединениями.

Анатомия OutOfMemoryError

Самая частая причина падения программ при работе с файлами — это утечка памяти (Memory Leak) или её банальная нехватка, приводящая к исключению java.lang.OutOfMemoryError.

java.lang.OutOfMemoryError генерируется, когда виртуальная машина Java (JVM) не может выделить объект из-за нехватки памяти, и сборщик мусора (Garbage Collector) больше не может освободить память для новых объектов.

habr.com

Оперативная память — это рабочий стол вашего приложения. Если вы скачиваете файл размером 2 ГБ и пытаетесь сохранить его в переменную типа ByteArray (массив байтов), JVM попытается найти на "рабочем столе" 2 ГБ непрерывного свободного места. Если приложению выделено всего 512 МБ памяти, произойдет неизбежный крах.

В Kotlin и библиотеке Ktor существует два способа получить тело HTTP-ответа:

1. Синхронная загрузка в память (Плохо для больших файлов)
2. Потоковое чтение (Отлично для любых файлов)

Рассмотрим пример того, как делать не нужно:

// ОПАСНЫЙ КОД: Загрузка всего файла в оперативную память
val response: HttpResponse = client.get("https://example.com/large-file.zip")
val fileBytes: ByteArray = response.body() // Здесь приложение упадет с OOM
File("downloaded.zip").writeBytes(fileBytes)

В этом коде метод body() ждет, пока весь файл скачается из сети, помещает его в оперативную память, и только потом записывает на диск.

Правильный подход в системном программировании — это потоковая передача (Streaming). Мы должны читать данные небольшими порциями и сразу сбрасывать их на жесткий диск.

Буферизация и системные вызовы

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

Мы зачерпываем порцию данных из сети (наполняем ведро), затем выливаем её на диск (опустошаем ведро), и повторяем процесс.

Размер этого буфера критически важен для производительности. Если буфер слишком маленький (например, 1 байт), мы будем обращаться к жесткому диску миллионы раз. Каждое обращение к диску или сети — это системный вызов (System Call), который требует переключения контекста процессора из пространства пользователя (User Space) в пространство ядра ОС (Kernel Space). Это очень дорогая операция.

Математически количество системных вызовов для записи файла можно выразить так:

$$N = \frac{S}{B}$$

Где:

• $N$ — количество системных вызовов (операций записи).
• $S$ — общий размер скачиваемого фрагмента в байтах.
• $B$ — размер буфера в байтах.

Если мы скачиваем файл размером 1 ГБ (1 073 741 824 байт):

• При буфере 1 байт: $N = 1 073 741 824$ вызовов. Диск будет непрерывно трещать, скорость упадет до килобайтов в секунду.
• При буфере 8 КБ (8192 байт): $N = 131 072$ вызовов. Это оптимальный баланс.
• При буфере 100 МБ: $N = 10$ вызовов. Вызовов мало, но мы неоправданно расходуем 100 МБ оперативной памяти на каждый поток. Если у нас 10 воркеров, мы потратим 1 ГБ RAM только на буферы.

В индустрии стандартом де-факто для сетевых и дисковых операций считается размер буфера в 8 КБ (8192 байт).

Давление на сборщик мусора (GC Pressure)

В языках с автоматическим управлением памятью, таких как Kotlin (работающий на JVM), за очистку неиспользуемых объектов отвечает Сборщик мусора (Garbage Collector, GC).

Когда вы создаете новый объект, он помещается в специальную область памяти (Eden Space). Когда эта область заполняется, JVM приостанавливает выполнение вашей программы (событие Stop-the-World), находит мусор и удаляет его. Чем больше объектов вы создаете, тем чаще происходят эти микро-зависания, снижая общую скорость загрузки.

Посмотрим на типичную ошибку новичка при реализации потокового чтения:

// НЕЭФФЕКТИВНЫЙ КОД: Создание объектов внутри цикла
val channel: ByteReadChannel = response.bodyAsChannel()

while (!channel.isClosedForRead) {
    // ОШИБКА: На каждой итерации создается новый массив на 8 КБ!
    val buffer = ByteArray(8192)
    val bytesRead = channel.readAvailable(buffer)
    if (bytesRead > 0) {
        randomAccessFile.write(buffer, 0, bytesRead)
    }
}

Если мы скачиваем файл размером 1 ГБ с буфером 8 КБ, цикл выполнится 131 072 раза. Это значит, что мы создадим 131 072 объекта ByteArray, суммарно выделив и тут же выбросив 1 ГБ оперативной памяти. Сборщик мусора будет работать на пределе возможностей, потребляя ресурсы процессора.

Правильный подход — переиспользование объектов (Object Reuse). Мы должны вынести выделение памяти за пределы цикла:

// ОПТИМИЗИРОВАННЫЙ КОД: Переиспользование одного буфера
val channel: ByteReadChannel = response.bodyAsChannel()

// Выделяем память ОДИН раз
val buffer = ByteArray(8192)

while (!channel.isClosedForRead) {
    // Перезаписываем старые данные в том же самом массиве
    val bytesRead = channel.readAvailable(buffer)
    if (bytesRead > 0) {
        randomAccessFile.write(buffer, 0, bytesRead)
    }
}

В этом случае на протяжении всей загрузки фрагмента используется ровно 8 КБ памяти. Сборщик мусора отдыхает, а процессор занят исключительно полезной работой — перекладыванием байтов из сети на диск.

Оптимизация сети: TCP Handshake и Connection Pooling

Теперь перейдем к сетевой части. Протокол HTTP работает поверх протокола TCP. Чтобы отправить даже один байт данных на сервер, клиент и сервер должны сначала установить соединение. Этот процесс называется Тройным рукопожатием (TCP 3-way Handshake).

Процесс выглядит так:

1. Клиент посылает пакет SYN (Synchronize).
2. Сервер отвечает пакетом SYN-ACK (Synchronize-Acknowledge).
3. Клиент подтверждает получение пакетом ACK (Acknowledge).

Если сервер находится на другом континенте, физическая передача сигнала по оптоволоконному кабелю занимает время (пинг).

Допустим, пинг до сервера составляет 100 миллисекунд (мс). Только на установку TCP-соединения уйдет 150 мс. Если используется защищенное соединение HTTPS, добавляется еще процесс согласования шифрования (TLS Handshake), что требует еще 2-3 пересылок пакетов. В итоге, прежде чем начнется скачивание данных, мы теряем около 300-500 мс.

В нашем менеджере загрузок мы разбиваем файл на множество фрагментов (чанков). Если файл разбит на 50 чанков, и для каждого чанка мы будем открывать новое соединение, мы потратим $50 \times 500 \text{ мс} = 25 \text{ секунд}$ только на рукопожатия!

Для решения этой проблемы используется механизм Пул соединений (Connection Pooling) и HTTP-заголовок Connection: keep-alive.

Несмотря на появление альтернатив (например, Ktor), связка OkHttp + Retrofit + Coroutines остаётся самой стабильной и предсказуемой. Причины просты: OkHttp обеспечивает низкоуровневую работу с HTTP: управление соединениями, кеширование, перехватчики, таймауты.

habr.com

Суть пула соединений в том, что после завершения загрузки чанка, сетевой сокет (соединение с сервером) не закрывается. Он помещается в специальный пул (очередь) в режиме ожидания. Когда воркеру нужно скачать следующий чанк с того же сервера, он берет уже открытое соединение из пула. Затраты времени на TCP и TLS Handshake становятся равными нулю.

Настройка пула соединений в Ktor

В Ktor за низкоуровневую работу с сетью отвечает HTTP-движок (Engine). По умолчанию мы используем движок CIO (Coroutine-based I/O), но его необходимо правильно сконфигурировать для высоконагруженных задач.

import io.ktor.client.*
import io.ktor.client.engine.cio.*

val httpClient = HttpClient(CIO) {
    engine {
        // Максимальное количество одновременных соединений ко всем серверам
        maxConnectionsCount = 1000

        // Максимальное количество соединений к одному конкретному хосту (серверу)
        // Если у нас 10 воркеров качают с одного сайта, ставим с запасом
        endpoint {
            maxConnectionsPerRoute = 20

            // Время удержания соединения открытым (Keep-Alive) в миллисекундах
            keepAliveTime = 20000 // 20 секунд

            // Таймаут на установку соединения
            connectTimeout = 5000
        }
    }
}

Разберем эти параметры:

• maxConnectionsPerRoute: Если этот параметр равен 2, а мы запустим 10 корутин для скачивания чанков с одного сервера, 8 корутин будут простаивать в ожидании, пока не освободятся 2 доступных соединения. Для параллельного менеджера загрузок этот лимит должен быть не меньше размера пула воркеров.
• keepAliveTime: Если сервер не присылает данные в течение этого времени, клиент сам закроет соединение, чтобы освободить оперативную память ОС.

Важное архитектурное правило: Объект HttpClient должен быть один на всё приложение (Singleton).

Стоит учитывать, что создание HttpClient является затратной операцией, и лучше использовать уже ранее созданные экземпляры HttpClient для выполнения нескольких запросов.

metanit.com

Если вы будете создавать HttpClient(CIO) внутри функции скачивания каждого чанка, пул соединений работать не будет, так как каждый клиент имеет свой собственный изолированный пул. Кроме того, создание самого клиента потребляет ресурсы.

Ограничение конкурентности (Concurrency Limiting)

Мы оптимизировали память внутри одной корутины и настроили сеть. Но что будет, если пользователь добавит в очередь 10 000 файлов, и мы запустим 10 000 корутин одновременно?

Корутины в Kotlin действительно очень легковесные. Вы можете создать миллион корутин, и приложение не упадет с OutOfMemoryError (в отличие от потоков ОС). Но корутины — это лишь абстракция. Под капотом они выполняют реальные сетевые запросы.

Если вы запустите 10 000 одновременных HTTP-запросов:

1. Вы исчерпаете лимит файловых дескрипторов операционной системы (каждый сетевой сокет — это файл для ОС).
2. Ваш домашний роутер может зависнуть от переполнения NAT-таблицы.
3. Удаленный сервер решит, что вы проводите DDoS-атаку, и заблокирует ваш IP-адрес (выдаст ошибку 429 Too Many Requests или просто разорвет соединение).

Именно поэтому в предыдущих статьях мы реализовали паттерн Worker Pool (Пул воркеров) с использованием Channel.

Оптимизация — это всегда поиск компромисса. Мы искусственно ограничиваем скорость (запуская, например, не более 4 параллельных загрузок), чтобы обеспечить стабильность системы в целом.

Сводная таблица параметров оптимизации

Параметр	Рекомендуемое значение	Что произойдет при слишком малом значении	Что произойдет при слишком большом значении
Размер буфера (I/O)	8 КБ (8192 байт)	Высокая нагрузка на CPU из-за системных вызовов	Перерасход оперативной памяти (RAM)
Пул соединений (на хост)	10 - 20	Простаивание корутин в ожидании сокета	Блокировка со стороны сервера (DDoS защита)
Пул воркеров (глобальный)	4 - 8	Неполная утилизация ширины интернет-канала	Исчерпание файловых дескрипторов ОС, зависание роутера

Системное программирование требует постоянного контроля над ресурсами. Понимая, как данные перемещаются из сети в буфер, как работает сборщик мусора и как устанавливаются TCP-соединения, вы можете писать на Kotlin консольные приложения, которые работают так же быстро и надежно, как программы на C++ или Rust, сохраняя при этом безопасность и удобство высокоуровневого языка.

Подготовка проекта курсовой работы к защите

Разработка сложного программного обеспечения — это лишь половина пути инженера. Вторая половина заключается в умении грамотно упаковать свой продукт, объяснить заложенные в него архитектурные решения и доказать их эффективность. Ваш консольный менеджер загрузок на Kotlin уже умеет разбивать файлы на части, управлять пулом корутин и безопасно склеивать фрагменты на диске. Теперь перед вами стоит задача академического уровня: превратить исходный код в полноценную курсовую работу по системному программированию.

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

Сборка автономного исполняемого файла (Fat JAR)

Первое правило успешной защиты консольного приложения: программа должна запускаться одной простой командой на любом компьютере, где установлена Java Virtual Machine (JVM). Преподаватель или член комиссии не должен устанавливать IntelliJ IDEA, настраивать Gradle или искать точку входа в вашем проекте.

Стандартная сборка Kotlin-проекта создает легковесный JAR-файл, который содержит только ваш скомпилированный код. Если вы попытаетесь запустить его, программа немедленно завершится с ошибкой NoClassDefFoundError. Это происходит потому, что стандартный JAR не включает в себя внешние зависимости: стандартную библиотеку Kotlin, Ktor, библиотеку корутин и другие модули.

Для решения этой проблемы используется концепция Fat JAR (или Uber JAR) — «толстого» архива, в который упаковывается как ваш код, так и все необходимые сторонние библиотеки.

Математически размер итогового файла можно выразить так:

$$S_{total} = S_{app} + \sum_{i=1}^{n} S_{dep_i}$$

Где:

• $S_{total}$ — итоговый размер Fat JAR архива.
• $S_{app}$ — размер скомпилированных классов вашего приложения.
• $n$ — количество внешних зависимостей.
• $S_{dep_i}$ — размер $i$-й внешней зависимости (например, kotlinx-coroutines-core).

Для автоматизации этого процесса в экосистеме Gradle существует популярный плагин Shadow. Добавьте его в ваш файл build.gradle.kts:

plugins {
    kotlin("jvm") version "1.9.22"
    // Добавляем плагин для сборки Fat JAR
    id("com.github.johnrengelman.shadow") version "8.1.1"
    application
}

application {
    // Указываем полный путь к классу с функцией main
    mainClass.set("ru.gurufy.downloadmanager.MainKt")
}

tasks.withType<com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar> {
    // Настраиваем имя итогового файла
    archiveBaseName.set("ab-download-manager-cli")
    archiveClassifier.set("")
    archiveVersion.set("1.0.0")
}

После настройки выполните команду ./gradlew shadowJar в терминале. В директории build/libs/ появится файл ab-download-manager-cli-1.0.0.jar. Теперь ваше приложение можно запустить на любой машине командой:

java -jar ab-download-manager-cli-1.0.0.jar https://example.com/large-file.iso

Структура пояснительной записки

Пояснительная записка — это официальный документ, который подтверждает вашу квалификацию как инженера-программиста. Комиссия часто уделяет тексту записки больше внимания, чем самому исходному коду.

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

1. Введение

Введение задает вектор всей работе. Здесь необходимо четко сформулировать академический аппарат исследования.

• Актуальность темы: Почему создание менеджера загрузок все еще важно? Обоснуйте это ростом объемов передаваемых данных, нестабильностью мобильных сетей и необходимостью обхода ограничений скорости со стороны серверов.
• Объект исследования: Процесс передачи данных по сети Интернет.
• Предмет исследования: Алгоритмы параллельной загрузки файлов и управление многопоточностью.
• Цель работы: Разработка консольного приложения для многопоточной загрузки файлов с использованием корутин Kotlin.
• Задачи: Проанализировать предметную область, спроектировать архитектуру, реализовать алгоритм разделения файла, разработать механизм слияния фрагментов, провести тестирование.

2. Теоретическая часть: Обоснование выбора технологий

В этом разделе вы должны доказать, что выбрали Kotlin и корутины не случайно, а на основе инженерного расчета. Сравните классические потоки операционной системы (OS Threads) и корутины.

Приведите математическое обоснование потребления памяти. Каждый поток ОС резервирует около 1 МБ оперативной памяти под свой стек вызовов. Если мы хотим запустить $N$ параллельных загрузок, потребление памяти только на стеки составит:

$$M_{threads} = N \times 1024 \text{ KB}$$

Для 1000 одновременных соединений потребуется около 1 ГБ оперативной памяти. В противовес этому, корутины в Kotlin являются объектами в куче (Heap) и весят всего несколько сотен байт.

Корутины позволяют писать асинхронный код последовательно, избегая проблемы «Callback Hell», и обеспечивают высочайшую плотность конкурентных задач на один системный поток.

kotlinlang.org

3. Практическая часть: Проектирование и реализация

Это сердце вашей курсовой работы. Здесь необходимо описать архитектуру приложения. Избегайте вставки простыней кода на десятки страниц — код должен быть в приложении к работе (Приложение А, Приложение Б). В основном тексте приводите только ключевые алгоритмы.

Опишите паттерн Worker Pool (Пул воркеров), который вы реализовали для ограничения конкурентности. Объясните, как используется Channel для передачи задач (чанков) воркерам, и почему был выбран Mutex для защиты общего счетчика прогресса.

Академический стиль описания архитектуры

Одна из главных трудностей для студентов — перевод мыслей с «программистского сленга» на строгий академический язык. Комиссия ожидает увидеть профессиональную терминологию.

Сленг разработчика	Академический язык для курсовой работы
Программа качает файл кусками	Реализован алгоритм сегментированной загрузки данных с независимой обработкой фрагментов
Я заюзал корутины, чтобы не тормозило	Для обеспечения неблокирующего ввода-вывода применена парадигма структурированной конкурентности
Склеиваем куски в один файл	Выполняется конкатенация временных бинарных файлов с использованием механизма Zero-Copy
Защитил переменную через Mutex	Обеспечена потокобезопасность критической секции путем применения механизма взаимного исключения
Если инет отвалится, программа попробует еще раз	Внедрен механизм отказоустойчивости с использованием алгоритма экспоненциальной задержки (Exponential Backoff)

Используя термины из правого столбца, вы демонстрируете глубокое понимание принципов системного программирования.

Визуализация многопоточности (UML-диаграммы)

Текстового описания недостаточно для понимания сложных многопоточных процессов. Стандартом де-факто в индустрии является язык визуального моделирования UML (Unified Modeling Language). Для курсовой работы по менеджеру загрузок обязательны две диаграммы.

Диаграмма прецедентов (Use Case Diagram)

Она показывает, что может делать пользователь (Actor) с вашей системой. В консольном приложении прецеденты инициируются через аргументы командной строки.

• Запуск загрузки по URL.
• Установка лимита потоков (флаг -t или --threads).
• Приостановка загрузки (сигнал SIGINT / Ctrl+C).
• Возобновление загрузки (повторный запуск с тем же URL).

Диаграмма последовательности (Sequence Diagram)

Это самая важная диаграмма для многопоточного приложения. Она показывает взаимодействие объектов во времени.

На диаграмме последовательности для вашего проекта должны быть отражены следующие шаги:

1. Main отправляет HTTP-запрос HEAD к серверу.
2. Сервер возвращает заголовки Content-Length и Accept-Ranges.
3. ChunkAllocator вычисляет размеры фрагментов.
4. WorkerPool запускает $N$ корутин.
5. Каждая корутина отправляет HTTP-запрос GET с заголовком Range.
6. Данные асинхронно пишутся в RandomAccessFile.
7. После завершения всех корутин FileMerger объединяет данные.

Подготовка к демонстрации (Live Demo)

Защита курсовой работы часто включает практическую демонстрацию. И здесь кроется главная опасность: закон Мерфи для программистов. Если что-то может сломаться во время презентации, оно обязательно сломается. Университетский Wi-Fi может отключиться, удаленный сервер может заблокировать вас за слишком частые запросы, а скорость интернета может упасть до нуля.

Чтобы избежать провала на защите, подготовьте локальный тестовый стенд.

Вместо того чтобы скачивать реальный файл из интернета, поднимите локальный HTTP-сервер на своем ноутбуке. Если у вас установлен Python, это делается одной командой в терминале (в папке с тестовым файлом):

python -m http.server 8080

Теперь вы можете демонстрировать работу вашего менеджера загрузок, скачивая файл по адресу http://localhost:8080/test-file.zip.

Преимущества локального стенда:

• Абсолютная независимость от качества интернета в аудитории.
• Мгновенный отклик сервера, что позволяет показать максимальную скорость работы ваших корутин.
• Возможность искусственно прервать работу сервера (Ctrl+C в терминале Python), чтобы эффектно продемонстрировать комиссии, как ваша программа обрабатывает сетевые ошибки и таймауты.

Защита: ответы на каверзные вопросы комиссии

Преподаватели любят проверять глубину понимания материала. Если вы использовали корутины и мьютексы, будьте готовы к тому, что вас спросят об их внутреннем устройстве. Вот три типичных вопроса и стратегии ответа на них.

Вопрос 1: «Вы используете Dispatchers.IO. Чем он отличается от Dispatchers.Default, и почему вы выбрали именно его?»

Правильный ответ: Dispatchers.Default оптимизирован для задач, нагружающих процессор (CPU-bound), и размер его пула потоков равен количеству ядер процессора. Если запустить на нем сетевые запросы, потоки быстро заблокируются в ожидании ответа от сервера, и приложение зависнет. Dispatchers.IO специально спроектирован для операций ввода-вывода (I/O-bound). Его пул эластичен и по умолчанию может расширяться до 64 потоков (или больше, в зависимости от настроек JVM), что позволяет эффективно удерживать множество одновременных сетевых соединений без блокировки вычислительных ресурсов.

Вопрос 2: «Что такое состояние гонки (Race Condition) и как вы его предотвратили при подсчете скачанных байт?»

Правильный ответ: Состояние гонки возникает, когда несколько потоков одновременно читают и модифицируют общую переменную. Операция progress += bytesRead не является атомарной (она состоит из чтения, сложения и записи). Если два воркера выполнят ее одновременно, часть данных будет потеряна. Для решения этой проблемы я использовал класс AtomicLong из пакета java.util.concurrent.atomic. Он использует низкоуровневые процессорные инструкции CAS (Compare-And-Swap), обеспечивая потокобезопасное обновление счетчика без дорогостоящей блокировки потоков операционной системы.

Вопрос 3: «Почему вы не загружаете весь файл в оперативную память перед сохранением на диск?»

Правильный ответ: Загрузка файла целиком в память приведет к исключению OutOfMemoryError при работе с большими файлами (например, ISO-образами на несколько гигабайт). В моей архитектуре реализована потоковая передача данных (Streaming). Данные вычитываются из сетевого сокета в небольшой буфер фиксированного размера (например, 8 КБ) и немедленно сбрасываются на жесткий диск через FileChannel. Это гарантирует, что потребление оперативной памяти остается минимальным и константным ($O(1)$), независимо от размера скачиваемого файла.

Успешная защита курсовой работы — это комбинация чистого кода, грамотно оформленной документации и уверенного владения инженерной терминологией. Подготовив Fat JAR, локальный тестовый стенд и отрепетировав ответы на вопросы об архитектуре, вы не только получите отличную оценку, но и закрепите навыки, необходимые для работы профессиональным разработчиком.