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

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

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

1. Основы многопоточности в PHP: расширение pthreads

Расширение pthreads — это основной инструмент для работы с потоками в PHP. Оно позволяет создавать, запускать и синхронизировать несколько потоков выполнения в рамках одного процесса. Важно: pthreads работает только в CLI-режиме и требует сборки PHP с поддержкой потоков (ZTS — Zend Thread Safety).

Пример создания простого потока:

<?php
class WorkerThread extends Thread {
    private $id;
    
    public function __construct($id) {
        $this->id = $id;
    }
    
    public function run() {
        echo "Поток {$this->id} запущен\n";
        sleep(2);
        echo "Поток {$this->id} завершен\n";
    }
}

$threads = [];
for ($i = 1; $i <= 5; $i++) {
    $threads[$i] = new WorkerThread($i);
    $threads[$i]->start();
}

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

Метод start() запускает поток, а join() ожидает его завершения. Без join() основной поток может завершиться раньше дочерних.

2. Синхронизация потоков и разделяемые ресурсы

При работе с многопоточностью критически важно синхронизировать доступ к общим данным. PHP предоставляет несколько механизмов: Mutex, Volatile и Threaded объекты.

Mutex (взаимное исключение) — блокирует доступ к ресурсу для других потоков, пока один поток работает с ним.

<?php
class Counter extends Threaded {
    private $count = 0;
    
    public function increment() {
        $this->synchronized(function() {
            $this->count++;
        });
    }
    
    public function getCount() {
        return $this->count;
    }
}

class IncrementThread extends Thread {
    private $counter;
    
    public function __construct(Counter $counter) {
        $this->counter = $counter;
    }
    
    public function run() {
        for ($i = 0; $i < 1000; $i++) {
            $this->counter->increment();
        }
    }
}

$counter = new Counter();
$threads = [];

for ($i = 0; $i < 10; $i++) {
    $threads[$i] = new IncrementThread($counter);
    $threads[$i]->start();
}

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

echo "Итоговое значение: " . $counter->getCount(); // 10000
?>

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

$sharedFlag = new Volatile();
$sharedFlag->flag = false;

// В одном потоке
$sharedFlag->flag = true;

// В другом потоке
if ($sharedFlag->flag) {
    echo "Флаг установлен";
}

3. Параллельные процессы и асинхронные задачи

Если расширение pthreads недоступно (например, на общем хостинге), можно использовать параллельные процессы через pcntl_fork() или асинхронные библиотеки, такие как ReactPHP или Amp.

pcntl_fork() — создает дочерний процесс, который выполняется параллельно с родительским.

<?php
$pid = pcntl_fork();
if ($pid == -1) {
    die("Не удалось создать процесс");
} elseif ($pid) {
    // Родительский процесс
    echo "Родительский процесс, PID дочернего: $pid\n";
    pcntl_wait($status); // Ожидание завершения дочернего процесса
} else {
    // Дочерний процесс
    echo "Дочерний процесс, PID: " . getmypid() . "\n";
    sleep(2);
    echo "Дочерний процесс завершен\n";
    exit(0);
}
?>

ReactPHP — библиотека для асинхронного программирования. Пример асинхронного HTTP-запроса:

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

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

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

$request = $client->request('GET', 'https://api.example.com/data');
$request->on('response', function ($response) {
    $response->on('data', function ($chunk) {
        echo "Получен чанк: " . strlen($chunk) . " байт\n";
    });
});

$loop->run();
?>

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

4. Управление пулом потоков и очередями

Для эффективного управления ресурсами часто используется пул потоков (Thread Pool). Это позволяет ограничить количество одновременно работающих потоков и переиспользовать их.

Реализация простого пула потоков на основе pthreads:

<?php
class Worker extends Thread {
    private $pool;
    
    public function __construct($pool) {
        $this->pool = $pool;
    }
    
    public function run() {
        while (true) {
            $task = $this->pool->getTask();
            if ($task === null) break;
            $task->execute();
        }
    }
}

class Task extends Threaded {
    private $data;
    
    public function __construct($data) {
        $this->data = $data;
    }
    
    public function execute() {
        echo "Обработка: " . $this->data . "\n";
        sleep(1);
    }
}

class ThreadPool {
    private $workers = [];
    private $queue = [];
    
    public function __construct($numWorkers) {
        for ($i = 0; $i < $numWorkers; $i