Как реализовать многопоточность в PHP приложениях?

Проблема многопоточности в PHP

Вы правы — в PHP нет нативной многопоточности в том виде, как в Java или C++. Каждый PHP-скрипт выполняется в одном потоке. Однако это не значит, что нельзя организовать параллельную обработку. Есть несколько проверенных подходов, и мы рассмотрим лучший из них для вашей задачи — использование расширения parallel (доступно с PHP 7.4+).

Решение с помощью расширения parallel

Расширение parallel предоставляет API для создания и управления потоками (Thread) и каналов (Channel) для обмена данными. Это идеально подходит для вашей задачи: вы можете разбить массив на части, обработать каждую часть в отдельном потоке, а затем собрать результаты.

Установка (на Linux):

sudo pecl install parallel

Добавьте extension=parallel.so в php.ini.

Пример реализации

<?php
// Исходные данные (в вашем случае 100k записей)
$allData = getLargeArrayFromDB(); // предположим, это массив

// Количество потоков (зависит от CPU и кол-ва ядер)
$numThreads = 8;

// Разбиваем данные на части
$chunks = array_chunk($allData, ceil(count($allData) / $numThreads));

// Создаем каналы для сбора результатов
$channel = new \parallel\Channel($numThreads);

// Функция, которая будет выполняться в каждом потоке
$worker = function(array $items, \parallel\Channel $channel) {
    $results = [];
    foreach ($items as $item) {
        // Ваша логика: HTTP-запрос, обработка, сохранение
        $result = callExternalAPI($item['id']);
        $processed = processResult($result);
        saveToFile($processed);
        $results[] = $processed;
    }
    // Отправляем результат обратно в основной поток
    $channel->send($results);
};

// Запускаем потоки
$futures = [];
foreach ($chunks as $chunk) {
    $futures[] = \parallel\run($worker, [$chunk, $channel]);
}

// Ждем завершения всех потоков
foreach ($futures as $future) {
    $future->value(); // блокируется до завершения потока
}

// Собираем результаты из канала
$allResults = [];
for ($i = 0; $i < $numThreads; $i++) {
    $allResults = array_merge($allResults, $channel->recv());
}

// Закрываем канал
$channel->close();

echo "Обработано записей: " . count($allResults);
?>

Альтернатива: Process Pool (для более сложных сценариев)

Если нужно управлять очередью задач, используйте \parallel\bootstrap и пул потоков:

<?php
$runtime = new \parallel\Runtime();
$future = $runtime->run(function() {
    // Код, выполняемый в отдельном потоке
    return heavyComputation();
});
$result = $future->value();
?>

Важные замечания

• Безопасность потоков: PHP не потокобезопасен (ZTS не гарантирует безопасность). Используйте parallel только для изолированных задач, без общего состояния.
• Ограничения: Нельзя передавать замыкания с внешними переменными (кроме примитивов). Используйте use или передавайте данные через каналы.
• Ресурсы: Каждый поток — отдельный процесс PHP, поэтому потребление памяти умножается на количество потоков.
• Ошибки: Обрабатывайте исключения внутри потока, иначе они "потеряются".

Когда использовать другие подходы

• curl_multi_exec — если задача сводится только к HTTP-запросам (без дополнительной обработки).
• pcntl_fork — для Unix-систем, требует осторожности с разделяемыми ресурсами.
• Очереди (RabbitMQ, Redis) — для распределенной обработки на нескольких серверах.

В вашем случае parallel — оптимальный выбор. Попробуйте разбить 100k записей на 8-16 потоков, и время обработки сократится с часов до минут (в зависимости от скорости API).

<?php
$data = getLargeArrayFromDB();
$chunks = array_chunk($data, 1000);
$processes = [];

foreach ($chunks as $chunk) {
    $pid = pcntl_fork();
    if ($pid == -1) {
        die('Ошибка форка');
    } elseif ($pid) {
        // Родительский процесс: сохраняем PID
        $processes[] = $pid;
    } else {
        // Дочерний процесс: обрабатываем свой кусок
        foreach ($chunk as $item) {
            $result = callExternalAPI($item['id']);
            processResult($result);
            saveToFile($result);
        }
        exit(0); // завершаем дочерний процесс
    }
}

// Ждем завершения всех дочерних процессов
foreach ($processes as $pid) {
    pcntl_waitpid($pid, $status);
}
?>