Многопроцессные и многопоточные вычисления в PHP

Расширенные примеры параллельных вычислений

Пример 1: Обработка массива изображений с помощью parallel

Задача: изменить размер 10 изображений в несколько потоков и замерить время выполнения.

<?php
require 'vendor/autoload.php';
use parallel\Runtime;
use parallel\Future;

$images = glob('images/*.jpg');
$total = count($images);
$workers = 4;
$chunkSize = ceil($total / $workers);
$futures = [];
$start = microtime(true);

for ($i = 0; $i < $workers; $i++) {
    $chunk = array_slice($images, $i * $chunkSize, $chunkSize);
    $runtime = new Runtime();
    $futures[] = $runtime->run(function($files) {
        $processed = 0;
        foreach ($files as $file) {
            $img = imagecreatefromjpeg($file);
            $resized = imagescale($img, 200, 150);
            imagejpeg($resized, 'output/' . basename($file));
            imagedestroy($img);
            imagedestroy($resized);
            $processed++;
        }
        return $processed;
    }, [$chunk]);
}

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

echo "Обработано изображений: $totalProcessed за " . (microtime(true) - $start) . " сек.\n";
?>

Обработано изображений: 10 за 2.345 сек.

Каждый поток обрабатывает свою часть списка. Время уменьшается почти пропорционально числу потоков (с учётом накладных расходов).

Пример 2: Параллельная обработка CSV файла с помощью pcntl_fork

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

<?php
$file = 'data.csv';
$lines = file($file, FILE_IGNORE_NEW_LINES);
$total = count($lines);
$workers = 4;
$chunkSize = ceil($total / $workers);
$childPids = [];

for ($i = 0; $i < $workers; $i++) {
    $pid = pcntl_fork();
    if ($pid == -1) { die('Fork failed'); }
    if ($pid == 0) {
        // Дочерний процесс
        $startLine = $i * $chunkSize;
        $chunk = array_slice($lines, $startLine, $chunkSize);
        $outFile = tempnam(sys_get_temp_dir(), "chunk_$i");
        $fh = fopen($outFile, 'w');
        foreach ($chunk as $line) {
            $processed = strtoupper($line); // Пример обработки
            fwrite($fh, $processed . PHP_EOL);
        }
        fclose($fh);
        // Передаём имя файла обратно через exit-код (ограничение) или через канал
        exit();
    } else {
        $childPids[] = $pid;
    }
}

// Родитель ожидает всех
foreach ($childPids as $pid) {
    pcntl_waitpid($pid, $status);
}
echo "Все дочерние процессы завершены.\n";
?>

Здесь не показан объединение результатов. Для передачи данных между процессами лучше использовать именованные каналы или разделяемую память.

Все дочерние процессы завершены.

Пример 3: Асинхронная загрузка нескольких URL с помощью curl_multi и прогресс бар

<?php
$urls = [];
for ($i = 0; $i < 20; $i++) {
    $urls[] = "https://httpbin.org/delay/1?num=$i";
}
$mh = curl_multi_init();
$handles = [];
foreach ($urls as $i => $url) {
    $ch = curl_init($url);
    curl_setopt_array($ch, [
        CURLOPT_RETURNTRANSFER => true,
        CURLOPT_TIMEOUT => 10,
        CURLOPT_CONNECTTIMEOUT => 5
    ]);
    curl_multi_add_handle($mh, $ch);
    $handles[$i] = $ch;
}
$active = null;
do {
    $status = curl_multi_exec($mh, $active);
    $info = curl_multi_info_read($mh);
    if ($info) {
        $ch = $info['handle'];
        $index = array_search($ch, $handles, true);
        echo "Завершён запрос #$index\n";
    }
    if ($active) {
        curl_multi_select($mh);
    }
} while ($active);

$contents = [];
foreach ($handles as $ch) {
    $contents[] = curl_multi_getcontent($ch);
    curl_multi_remove_handle($mh, $ch);
    curl_close($ch);
}
curl_multi_close($mh);
echo "Загружено " . count($contents) . " страниц.\n";
?>

Завершён запрос #0
Завершён запрос #3
...
Загружено 20 страниц.

Пример 4: Очередь задач с RabbitMQ (amqp расширение)

<?php
// producer.php
$connection = new AMQPConnection([
    'host' => 'localhost',
    'port' => 5672,
    'login' => 'guest',
    'password' => 'guest'
]);
$connection->connect();
$channel = new AMQPChannel($connection);
$exchange = new AMQPExchange($channel);
$exchange->setName('tasks');
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->declare();

for ($i = 0; $i < 100; $i++) {
    $msg = json_encode(['id' => $i, 'data' => "task_$i"]);
    $exchange->publish($msg, 'processing');
}
echo "Опубликовано 100 задач\n";
$connection->disconnect();
?>

<?php
// worker.php
$connection = new AMQPConnection([
    'host' => 'localhost',
    'port' => 5672,
    'login' => 'guest',
    'password' => 'guest'
]);
$connection->connect();
$channel = new AMQPChannel($connection);
$queue = new AMQPQueue($channel);
$queue->setName('processing');
$queue->setFlags(AMQP_DURABLE);
$queue->declare();
$queue->bind('tasks', 'processing');

while (true) {
    $envelope = $queue->get();
    if ($envelope) {
        $msg = json_decode($envelope->getBody(), true);
        echo "Обрабатывается задача #{$msg['id']}\n";
        // Имитация работы
        usleep(100000);
        $queue->ack($envelope->getDeliveryTag());
    } else {
        sleep(1);
    }
}
$connection->disconnect();
?>

Обрабатывается задача #0
Обрабатывается задача #1
...

Запуск нескольких worker'ов позволит обрабатывать задачи параллельно. Каждая задача гаратированно будет обработана (благодаря подтверждению ack).