Конкурентность и многопоточность в PHP: продвинутый уровень

Введение: почему конкурентность в PHP — это не миф

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

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

1. Основы: процессы vs потоки в PHP

Прежде чем погружаться в код, важно понять разницу между процессами и потоками в контексте PHP.

• Процессы — изолированные экземпляры, каждый со своей памятью. Создаются через pcntl_fork(). Подходят для тяжелых задач, но требуют межпроцессного взаимодействия (IPC).
• Потоки — легковесные "подзадачи" внутри одного процесса, разделяющие память. Реализуются через расширение pthreads. Идеальны для работы с общими данными, но требуют синхронизации.

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

1.1 Параллельные процессы через pcntl_fork()

Системный вызов pcntl_fork() создает дочерний процесс, который копирует память родителя. Рассмотрим пример параллельной обработки нескольких URL:

<?php
$urls = ['https://api.example.com/data1', 'https://api.example.com/data2', 'https://api.example.com/data3'];
$pids = [];

foreach ($urls as $url) {
    $pid = pcntl_fork();
    if ($pid == -1) {
        die('Не удалось создать процесс');
    } elseif ($pid) {
        // Родительский процесс: сохраняем PID
        $pids[] = $pid;
    } else {
        // Дочерний процесс: выполняем загрузку
        $data = file_get_contents($url);
        file_put_contents('/tmp/' . md5($url) . '.json', $data);
        exit(0); // Завершаем дочерний процесс
    }
}

// Ожидаем завершения всех процессов
foreach ($pids as $pid) {
    pcntl_waitpid($pid, $status);
}
echo "Все URL обработаны параллельно!";
?>

Важно: после fork() дочерние процессы наследуют все открытые дескрипторы файлов и соединения с БД. Закрывайте их явно в дочернем процессе, чтобы избежать конфликтов.

2. Асинхронные задачи с ReactPHP и Amp

Для веб-приложений, где важна реактивность, используйте библиотеки событийно-ориентированного цикла. ReactPHP и Amp позволяют выполнять неблокирующие операции, не создавая отдельные процессы.

2.1 Пример с ReactPHP

Установите библиотеку через Composer: composer require react/react. Ниже — асинхронный HTTP-клиент, который отправляет три запроса одновременно:

<?php
require 'vendor/autoload.php';

use React\EventLoop\Factory;
use React\HttpClient\Client;

$loop = Factory::create();
$client = new Client($loop);

$urls = ['http://example.com', 'http://example.org', 'http://example.net'];
$promises = [];

foreach ($urls as $url) {
    $request = $client->request('GET', $url);
    $request->on('response', function ($response) use ($url) {
        echo "Ответ от {$url}: " . $response->getCode() . PHP_EOL;
    });
    $request->end();
}

$loop->run();
?>

ReactPHP создает единый цикл событий, который обрабатывает все запросы без блокировки. Это идеально для API-шлюзов или микросервисов, где важна скорость ответа.

3. Управление потоками с pthreads (ZTS-сборка)

Расширение pthreads — единственный способ получить настоящую многопоточность в PHP. Внимание: требует сборки PHP с флагом --enable-zts (Zend Thread Safety).

3.1 Создание и синхронизация потоков

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

<?php
class Worker extends Thread {
    private $data;
    private $result;
    private $mutex;

    public function __construct(array $data, &$result, Mutex $mutex) {
        $this->data = $data;
        $this->result = &$result;
        $this->mutex = $mutex;
    }

    public function run() {
        foreach ($this->data as $value) {
            $processed = $value * 2; // Имитация сложной обработки
            Mutex::lock($this->mutex);
            $this->result[] = $processed;
            Mutex::unlock($this->mutex);
        }
    }
}

$dataChunks = array_chunk(range(1, 30), 10);
$result = [];
$mutex = Mutex::create();
$threads = [];

foreach ($dataChunks as $chunk) {
    $thread = new Worker($chunk, $result, $mutex);
    $thread->start();
    $threads[] = $thread;
}

foreach ($threads as $thread) {
    $thread->join();
}

Mutex::destroy($mutex);
echo "Результат: " . implode(', ', $result);
?>

Здесь Mutex гарантирует, что только один поток одновременно добавляет данные в массив $result. Без мьютекса вы получите состояние гонки (race condition).

3.2 Пул потоков для масштабирования

Создавать поток на каждую задачу — неэффективно. Используйте Pool из pthreads для повторного использования потоков:

<?php
class Task extends Threaded {
    private $id;
    public function __construct($id) { $this->id = $id; }
    public function run() {
        echo "Задача {$this->id} выполняется в потоке " . Thread::getCurrentThreadId() . PHP_EOL;
        sleep(1);
    }
}

$pool = new Pool(4); // 4 потока в пуле
for ($i = 1; $i <= 10; $i++) {