Параллельная обработка данных с помощью PHP

Расширенные примеры с расширением parallel

Пул потоков для пакетной обработки

Создание пула из 4 потоков для выполнения 10 задач:

<?
use parallel\Runtime;

$tasks = range(1, 10);
$runtimes = [];
$futures = [];

// Создание пула воркеров
for ($i = 0; $i < 4; $i++) {
    $runtimes[$i] = new Runtime();
}

// Распределение задач
foreach ($tasks as $task) {
    $runtime = $runtimes[array_rand($runtimes)]; // случайное назначение
    $futures[] = $runtime->run(function($id) {
        sleep(1); // имитация работы
        return $id * 2;
    }, [$task]);
}

// Сбор результатов
$results = [];
foreach ($futures as $future) {
    $results[] = $future->value();
}

print_r($results);
?>

Array
(
    [0] => 2
    [1] => 4
    [2] => 6
    [3] => 8
    [4] => 10
    [5] => 12
    [6] => 14
    [7] => 16
    [8] => 18
    [9] => 20
)

Пояснение: каждый поток переиспользуется для нескольких задач. Функция run() принимает замыкание и аргументы. Результаты сохраняются и собираются после завершения. Время выполнения сокращается примерно в 4 раза по сравнению с последовательным выполнением.

Обработка большого массива данных

Разбиение массива на части и параллельная обработка:

<?
use parallel\Runtime;

$data = range(1, 100);
$chunks = array_chunk($data, 25); // 4 части

$runtimes = [];
$futures = [];

foreach ($chunks as $key => $chunk) {
    $runtimes[$key] = new Runtime();
    $futures[$key] = $runtimes[$key]->run(function($chunk) {
        $sum = 0;
        foreach ($chunk as $value) {
            $sum += $value; // может быть любая ресурсоемкая операция
        }
        return $sum;
    }, [$chunk]);
}

$total = 0;
foreach ($futures as $future) {
    $total += $future->value();
}

echo "Сумма: $total\n"; // должно быть 5050
?>

Сумма: 5050

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

Обработка ошибок в потоках

Исключения внутри run() пробрасываются в основной поток при вызове value():

<?
use parallel\Runtime;

$runtime = new Runtime();
$future = $runtime->run(function() {
    throw new \Exception("Ошибка в потоке");
});

try {
    $future->value();
} catch (\Exception $e) {
    echo "Поймано исключение: " . $e->getMessage() . "\n";
}
?>

Поймано исключение: Ошибка в потоке

Для обработки ошибок без прерывания всего пула можно внутри run() перехватывать исключения и возвращать код ошибки.

Мультиплексирование с parallel\Events

Класс parallel\Events позволяет ожидать завершение первого из нескольких Future (как select):

<?
use parallel\Runtime;
use parallel\Events;

$r1 = new Runtime();
$r2 = new Runtime();

$f1 = $r1->run(function() { sleep(3); return "A"; });
$f2 = $r2->run(function() { sleep(1); return "B"; });

$events = new Events();
$events->addFuture("task1", $f1);
$events->addFuture("task2", $f2);

$event = $events->poll(); // вернет объект Event
if ($event->type === Events\Event\Type::Success) {
    echo $event->name . " -> " . $event->value . "\n";
}
?>

task2 -> B

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

Передача объектов с автозагрузкой

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

<?
use parallel\Runtime;

// Файл MyClass.php
class MyClass {
    public function greet() { return "Hello"; }
}

$runtime = new Runtime();
$future = $runtime->run(function() {
    require_once 'MyClass.php'; // подгрузка внутри потока
    $obj = new MyClass();
    return $obj->greet();
});
echo $future->value();
?>

Hello

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