Асинхронное программирование становится фундаментальным элементом в построении масштабируемых веб-приложений. Причина этого заключается в растущей потребности увеличения количества действий на каждый веб-запрос.Типичным примером этого является отправка электронного письма в рамках запроса.

Во многих веб-приложениях, когда что-то обрабатывается на сервере, мы хотим уведомить людей по электронной почте. Обычно для этого создается отдельный HTTP-запрос к стороннему сервису, такому как SendGrid, Mailchimp и т. д.

Совсем непросто, когда вам необходимо сразу отправить много электронных писем. Отправив десятки или сотни писем (HTTP-процесс при отправке одного письма занимает 100 мс) в PHP, вы сразу же увеличиваете общее время запроса. 

С другой стороны при решении этой проблемы, любой хороший сторонний почтовый сервис предоставит ендпоинт для отправки большого количества сообщений. Но предположим, что вы хотите отправить 100 сообщений, и каждое из них должно быть обработано отдельно.

Итак, нам нужно принять решение: как перенести обработку электронных писем в отдельный процесс, чтобы он не блокировал исходный веб-запрос?

Именно это мы и рассмотрим в данной статье, в частности, всевозможные способы решения данной проблемы на PHP используя новую инфраструктуру (или без нее).

Применение exec()

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

Давайте рассмотрим пример.

<?php
// handle a web request

// record the start time of the web request
$start = microtime(true);
$path = __DIR__ . '/send_email.php';

// output to /dev/null & so we don't block to wait for the result
$command = 'php ' . $path . ' --email=%s > /dev/null &';
$emails = ['joe@blogs.com', 'jack@test.com'];

// for each of the emails, call exec to start a new script
foreach ($emails as $email) {
    // Execute the command
    exec(sprintf($command, $email));
}

// record the finish time of the web request
$finish = microtime(true);
$duration = round($finish - $start, 4);

// output duration of web request
echo "finished web request in $duration\n";

send_email.php

<?php

$email = explode('--email=', $argv[1])[1];
// this blocking sleep won't affect the web request duration
// (illustrative purposes only)
sleep(5);

// here we can send the email
echo "sending email to $email\n";

Вывод результата:

$ php src/exec.php

finished web request in 0.0184

Приведенные выше скрипты показывают, что веб-запрос по-прежнему завершается за миллисекунды, даже несмотря на блокирующий вызов функции sleep в send_email.php.

Причина, по которой он не блокируется, заключается в том, что мы сообщили exec с помощью включения > /dev/null & в команду о том, что не хотим ждать завершения команды exec для получения результата. Поэтому это произойдет в фоновом режиме, и веб-запрос может продолжиться.

Таким образом, скрипт веб-запроса просто отвечает за запуск сценария, а не за мониторинг его выполнения и/или сбоя.

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

exec запускает команду на сервере, поэтому вы должны быть внимательны к тому, как выполняется скрипт, особенно если он включает в себя пользовательский ввод. Управление с помощью exec может быть довольно сложным, особенно при масштабировании приложения, поскольку скрипт, скорее всего, запускается на том же самом серверном  блоке, который обрабатывает внешние веб-запросы. Поэтому в конечном итоге вы можете исчерпать ресурс процессора и память, если через exec будут созданы сотни или тысячи новых процессов.

pcntl_fork

pcntl_fork — это низкоуровневая функция, требующая включения PCNTL-расширения и являющаяся мощным, но потенциально опасным методом при написании асинхронного кода в PHP.

pcntl_fork форкает или клонирует текущий процесс и разделяет его на родительский и несколько дочерних (в зависимости от того, сколько раз она будет вызвана). Определяя идентификатор процесса (Process ID или PID), мы можем выполнять различный код в контексте родительского или дочернего процесса.

Родительский процесс будет отвечать за создание дочерних процессов и ожидание их завершения, прежде чем он сможет прекратиться.

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

Теперь перейдем к коду в нашем примере для неблокируемой отправки электронной почты.

<?php

function sendEmail($to, $subject, $message)
{
    // Code to send email (replace with your email sending logic)
    // This is just a mock implementation for demonstration purposes
    sleep(3); // Simulating sending email by sleeping for 3 seconds
    echo "Email sent to: $to\n";
}

$emails = [
    [
        'to' => 'john@example.com',
        'subject' => 'Hello John',
        'message' => 'This is a test email for John.',
    ],
    [
        'to' => 'jane@example.com',
        'subject' => 'Hello Jane',
        'message' => 'This is a test email for Jane.',
    ],
    // Add more email entries as needed
];

$children = [];

foreach ($emails as $email) {
    $pid = pcntl_fork();

    if ($pid == -1) {
        // Fork failed
        die('Error: Unable to fork process.');
    } elseif ($pid == 0) {
        // Child process
        sendEmail($email['to'], $email['subject'], $email['message']);
        exit(); // Exit the child process
    } else {
        // Parent process
        $children[] = $pid;
    }
}

echo "running some other things in parent process\n";
sleep(3);

// Parent process waits for each child process to finish
foreach ($children as $pid) {
    pcntl_waitpid($pid, $status);
    $status = pcntl_wexitstatus($status);
    echo "Child process $pid exited with status: $status\n";
}

echo 'All emails sent.';

В приведенном выше примере с помощью pcntl_fork мы можем форкнуть текущий процесс, который копирует родительский в новые дочерние процессы и дождаться завершения выполнения. Кроме того, после форкинга дочерних процессов для отправки электронной почты родительский процесс может продолжать выполнение других действий, до того как окончательно убедится, что дочерние процессы завершены.

Это шаг вперед по сравнению с использованием exec. Там мы были весьма ограничены в возможностях, поскольку скрипты являются полностью отдельными контекстами, что делает невозможным их общий мониторинг.

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

Отслеживая идентификаторы процессов, мы можем эффективно контролировать поток выполнения и управлять им.

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

К счастью, есть решение, и оно заключается в том, чтобы объединить exec и pcntl_fork, для извлечения лучшего из обоих подходов. Это выглядит следующим образом:

1. Веб-запрос использует exec() для создания нового PHP-процесса

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

3. Созданный процесс становится родительским, поскольку он форкается для отправки каждого письма по отдельности

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

Давайте посмотрим, как это работает:

<?php

$start = microtime(true);
$path = __DIR__ . '/pcntl_fork_send_email.php';
$emails = implode(',', ['joe@blogs.com', 'jack@test.com']);
$command = 'php ' . $path . ' --emails=%s > /dev/null &';

// Execute the command
echo "running exec\n";
exec(sprintf($command, $emails));
$finish = microtime(true);

$duration = round($finish - $start, 4);
echo "finished web request in $duration\n";

pctnl_fork_send_email.php

<?php

$param = explode('--emails=', $argv[1])[1];
$emails = explode(',', $param);

function sendEmail($to)
{
    sleep(3); // Simulating sending email by sleeping for 3 seconds
    echo "Email sent to: $to\n";
}

$children = [];

foreach ($emails as $email) {
    $pid = pcntl_fork();

    if ($pid == -1) {
        // Fork failed
        die('Error: Unable to fork process.');
    } elseif ($pid == 0) {
        // Child process
        sendEmail($email);
        exit(); // Exit the child process
    } else {
        // Parent process
        $children[] = $pid;
    }
}

echo "running some other things in parent process\n";
sleep(3);

// Parent process waits for each child process to finish
foreach ($children as $pid) {
    pcntl_waitpid($pid, $status);
    $status = pcntl_wexitstatus($status);
    echo "Child process $pid exited with status: $status\n";
}

echo "All emails sent.\n";

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

AMPHP

amphp (Asynchronous Multi-tasking PHP. Асинхронный многозадачный PHP) — это набор библиотек, позволяющих создавать быстрые, параллельные приложения на PHP.

В релизе PHP 8.1 за ноябрь 2021 года появилась поддержка Fibers, реализующих легковесную кооперативную модель параллелизма.

Теперь мы знаем немного о том, как работает amphp и почему это интересно для будущих PHP-программ. Давайте рассмотрим пример:

<?php

require __DIR__ . '/../vendor/autoload.php'; // Include the autoload file for the amphp/amp library

use function Amp\delay;
use function Amp\async;

function sendEmail($to, $subject, $message)
{
    delay(3000)->onResolve(function () use ($to) {
        echo "Email sent to: $to\n";
    });
}

$emails = [
    [
        'to' => 'john@example.com',
        'subject' => 'Hello John',
        'message' => 'This is a test email for John.',
    ],
    [
        'to' => 'jane@example.com',
        'subject' => 'Hello Jane',
        'message' => 'This is a test email for Jane.',
    ],
    // Add more email entries as needed
];

foreach ($emails as $email) {
    $future = async(static function () use ($email) {
        $to = $email['to'];
        $subject = $email['subject'];
        $message = $email['message'];
        sendEmail($to, $subject, $message);
    });

    // block current process by running $future->await();
}

echo "All emails sent.\n";

Приведенный выше скрипт представляет собой очень простую версию асинхронного выполнения действий. Он создает новый файбер асинхронно, используя заданное замыкание, возвращая Future (объект).

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

Очереди и воркеры

Решение этой проблемы существует и за пределами PHP. До выхода PHP 8.1 его можно было считать золотым стандартом, поскольку оно не зависит от языка и обладает высокой масштабируемостью.

Использование сервисов очередей, таких как Amazon SQS, RabbitMQ или Apache Kafka, уже давно является общепринятым решением.

Очереди — это фрагменты инфраструктуры для запуска воркеров, независимых от вашего приложения, для асинхронной обработки любого задания. Это тоже не лишено риска и недостатков, но зато проверено временем.

Давайте разберем пример:

Отправитель, в данном случае, это, как правило, ваше существующее веб-приложение.

sender.php

<?php

require 'vendor/autoload.php';

use Aws\Sqs\SqsClient;

// Initialize the SQS client
$client = new SqsClient([
    'region' => 'us-east-1',
    'version' => 'latest',
    'credentials' => [
        'key' => 'YOUR_AWS_ACCESS_KEY',
        'secret' => 'YOUR_AWS_SECRET_ACCESS_KEY',
    ],
]);

// Define the message details
$message = [
    'to' => 'john@example.com',
    'subject' => 'Hello John',
    'message' => 'This is a test email for John.',
];

// Send the message to SQS
$result = $client->sendMessage([
    'QueueUrl' => 'YOUR_SQS_QUEUE_URL',
    'MessageBody' => json_encode($message),
]);

echo "Message sent to SQS with MessageId: " . $result['MessageId'] . "\n";

Воркеры — это дополнительный деплой исполняемого кода для обработки джобов.

worker.php

<?php

require 'vendor/autoload.php';

use Aws\Sqs\SqsClient;

// Initialize the SQS client
$client = new SqsClient([
    'region' => 'us-east-1',
    'version' => 'latest',
    'credentials' => [
        'key' => 'YOUR_AWS_ACCESS_KEY',
        'secret' => 'YOUR_AWS_SECRET_ACCESS_KEY',
    ],
]);

// Receive and process messages from SQS
while (true) {
    $result = $client->receiveMessage([
        'QueueUrl' => 'YOUR_SQS_QUEUE_URL',
        'MaxNumberOfMessages' => 1,
        'WaitTimeSeconds' => 20,
    ]);

    if (!empty($result['Messages'])) {
        foreach ($result['Messages'] as $message) {
            $body = json_decode($message['Body'], true);

            // Process the message (send email in this case)
            sendEmail($body['to'], $body['subject'], $body['message']);

            // Delete the message from SQS
            $client->deleteMessage([
                'QueueUrl' => 'YOUR_SQS_QUEUE_URL',
                'ReceiptHandle' => $message['ReceiptHandle'],
            ]);
        }
    }
}

function sendEmail($to, $subject, $message)
{
    sleep(3); // Simulating sending email by sleeping for 3 seconds
    echo "Email sent to: $to\n";
}

Это решение состоит из двух частей:

• Отправитель (отправляет сообщение в SQS-очередь)

• Воркер (получает сообщение из очереди и отправляет его по электронной почте).

Данное решение можно масштабировать путем увеличения количества воркеров по отношению к количеству сообщений, посылаемых любым количеством отправителей.

При использовании очереди воркер полностью независим от отправителя и может быть написан на любом языке, поскольку связь между отправителем и воркером осуществляется через JSON-сообщения.

Какое решение лучше?

Почти невозможно сказать, какое из решений, рассмотренных выше, будет лучшим для вашего приложения. Хотя все они направлены на обеспечение выполнения асинхронного кода в PHP, их реализация довольно сильно отличается и имеет различные преимущества и недостатки.

Обобщим каждый из вариантов в нескольких пунктах:

exec()

• Возможно, самый простой и эффективный способ асинхронного выполнения PHP-скриптов.

• Опасен потенциальными последствиями с точки зрения безопасности, особенно в отношении пользовательского ввода

• Ничего не предоставляется для совместного использования, что может быть как благословением, так и проклятием

• Может привести к повышенному потреблению ресурсов сервера (CPU/память)

pcntl_fork()

• Позволяет управлять родительскими и дочерними процессами для кастомизации поведения

• Может быть абстрагирован в более простой API для вашего приложения

• Клонирование текущего процесса может вызвать появление других проблем, возникающих вследствие этого

AMPHP

• Требуется PHP 8.1 для пользователя Fibers

• Библиотека абстрагировалась от "жесткой сцепленности частей" выполнения асинхронного кода

• Крутая кривая обучения по сравнению с другими более традиционными методами (понимание цикла событий и многозадачности в PHP)

Очереди и воркеры

• Независимость от языка, гибкость для любого юзкейса

• Представляет распределенную систему (в долгосрочной перспективе это может быть как хорошо, так и плохо)

• Множество решений и различные провайдеры очередей облегчают работу.

Заключение

Основная причина, по которой я хотел немного углубиться в различные возможности асинхронного кода в PHP, заключается в том, чтобы понять, как введение Fibers в PHP 8.1 изменит перспективы для написания асинхронных программ в будущем.

Существует множество решений прошедших боевые испытания и не требующих PHP 8.1. Однако интересно посмотреть, в каком направлении развивается язык PHP, чтобы конкурировать с Golang и Elixir, которые поддерживают асинхронное программирование и делают это уже много лет.

В конечном счете, я, вероятно, все же бы выбрал подход на основе Queue/Worker (Очередь/Воркер), учитывая масштабируемость и кросс-платформенную/кросс-языковую поддержку. Однако думаю, со временем мы увидим, что такие библиотеки, как AMPHP, станут более функциональными и позволят решить эту проблему без внедрения новой инфраструктуры.

Примеры кода, использованные в этой статье, вы можете найти на GitHub.

Как написать библиотеку на C или Go для вашего PHP-проекта, а главное — зачем? Обсудим этот вопрос на открытом уроке 10 июля. На занятии поговорим о применении технологии, напишем библиотеку, и добавим ее в проект на PHP. Обсудим случаи применимости технологии FFI (Foreign Function Interface), поговорим о том, в каких случаях ее применять не стоит.

Занятие будет полезно для уверенно владеющих PHP разработчиков, которые пришли к вопросу о возможности встраивания низкоуровневых библиотек в свои проекты.

• Записаться на открытый урок