После выхода PHP7 появилась возможность сравнительно небольшой ценой писать долгоживущие приложения. Для программистов стали доступны такие проекты, как prooph, broadway, tactician, messenger, авторы которых берут на себя решение наиболее частых проблем. Но что если сделать небольшой шаг вперёд, углубившись в вопрос?

Попробуем разобрать судьбу ещё одного велосипеда, который позволяет реализовать Publish/Subscribe приложение.

Для начала постараемся вкратце рассмотреть современные тенденции в мире PHP, а также поверхностно коснёмся асинхронной работы.

PHP создан, чтобы умирать

Долгое время PHP использовался в основном в схеме работы request/response. С точки зрения разработчиков это довольно удобно, ведь нет необходимости беспокоиться об утечках памяти, следить за подключениями.

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

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

• Разбор конфигурации;
• Компиляция контейнера;
• Маршрутизация запроса;
• Выполнение;
• Рендеринг результата.

Как и в случае с PHP (использование акселераторов), фреймворк активно пользуется кешированием (часть задач не будет выполнена при следующем запросе), а также отложенной инициализацией. Начиная с версии 7.4 станет доступен preload, который позволит дополнительно оптимизировать инициализацию приложения.

Тем не менее, полностью все накладные расходы на инициализацию убрать не получится.

Поможем PHP выжить

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

В экосистеме PHP есть такие проекты, как php-pm и RoadRunner. Оба концептуально делают одно и то же:

• Создаётся родительский процесс, который выполняет роль супервайзера;
• Создаётся пул дочерних процессов;
• При получении запроса master извлекает из пула процесс, передаёт ему запрос. Клиент в это время удерживается в ожидании;
• Как только задача выполнена, master возвращает результат клиенту, а дочерний процесс отправляется обратно в пул.

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

Примечание:
в сети гуляет множество примеров из серии «берём ReactPHP и ускоряем Laravel в N раз». Важно понимать разницу между демонизацией (и, как следствие, экономии времени на бутстрапинге приложения) и многозадачностью.
При использовании php-pm или roadrunner ваш код не становится неблокирующим. Вы просто экономите время на инициализации.
Сравнивать php-pm, roadrunner и ReactPHP/Amp/Swoole некорректно по определению.

PHP и I/O

Взаимодействие с I/O в PHP по умолчанию выполняется в блокирующем режиме. Это означает то, что если мы выполним запрос на обновление информации в таблице, то поток выполнения приостановится в ожидании ответа от базы данных. Чем больше таких вызовов в процессе обработки запроса, тем большее время ресурсы сервера простаивают. Ведь в процессе обработки запроса нам необходимо несколько раз сходить в базу данных, записать что-либо в лог, да и вернуть клиенту результат, в конце концов — тоже блокирующая операция.

Представьте, что вы — оператор call-центра и вам за час надо обзвонить 50 клиентов.
Вы набираете первый номер, а там занято (абонент обсуждает по телефону последнюю серию Игры Престолов и то, во что скатился сериал).
И вот вы сидите и до победы пытаетесь до него дозвониться. Время идёт, смена подходит к концу. Потеряв 40 минут на попытку дозвониться до первого абонента, вы упустили возможность связаться с прочими и закономерно получили от начальника.
Но вы можете поступить иначе: не дожидаться пока первый абонент освободится и как только услышали гудки, положить трубку и начать набор следующего номера. К первому можно вернуться несколько позднее.
При таком подходе шансы обзвонить максимальное количество человек сильно повышаются, а скорость вашей работы не упирается в самую медленную задачу.

Код, который не блокирует поток выполнения (не использует блокирующих I/O вызовов, а также функций вроде sleep()), называют асинхронным.

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

Но это всё условности, давайте попробуем выкинуть Symfony и воспользоваться для решения нашей задачи Amp, который предоставляет реализацию Event Loop (включающую ряд биндингов), Promises и Coroutines в качестве вишенки на тортик.

Promise — это один из способов организации асинхронного кода. Например, нам необходимо выполнить обращение к какому-либо http ресурсу.

Мы создаём объект запроса и передаём его в транспорт, который возвращает нам Promise, содержащий текущее состояние. Всего возможны три состояния:

• Успех: наш запрос был успешно выполнен;
• Ошибка: в процессе выполнения запроса что-то пошло не так (например, сервер вернул 500 ответ);
• Ожидание: обработка запроса ещё не началась.

Каждый Promise имеет один метод (в примере разбирается Promise от Amp) — onResolve(), в который передаётся callback функция с двумя аргументами

$promise->onResolve(
    static function(?/Throwable $throwable, $result): void {
        if(null !== $throwable) {
            /** Произошла ошибка */
            return;
        }

        /** Успех */
    }
);

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

Для этого используется Event Loop.

В сущности, Event Loop — это планировщик, который контролирует выполнение. Как только выполнение задачи будет завершено (не важно, как), будет вызван callable, который мы передали в Promise.

Что касается нюансов, то я бы рекомендовал почитать статью от Никиты Попова: Cooperative multitasking using coroutines. Она поможет внести некоторую ясность относительно того, что происходит и причём тут генераторы.

Вооружившись новыми знаниями попробуем вернуться к нашей задаче по рендерингу JSON.
Пример обработки входящего http запроса с помощью amphp/http-server.
Как только мы получили запрос, выполняется асинхронное чтение из базы данных (получаем Promise) и по его завершению пользователю будет отдан заветный JSON, сформированный на основании полученных данных.

Если нам необходимо слушать один порт из нескольких процессов, можно посмотреть в сторону amphp/cluster

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

Дивный мир асинхронного PHP

Дисклеймер
Асинхронный PHP рассматривается в контексте экзотики и не считается чем-то здоровым/нормальным. В основном будут ждать смешки в стиле «возьми GO/Kotlin, дура» и т.д. Я бы не сказал, что эти люди не правы, но...

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

Swoole

Асинхронный фреймворк, написанный в отличии от остальных на Си и поставляемый в виде расширения к PHP. Обладает, пожалуй, лучшими показателями производительности на текущий момент.

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

Что касается комьюнити, то тоже не всё просто и однозначно. Лично я не знаю ни одного живого человека, кто использует Swoole в бою. Возможно, я поборю свои страхи и мигрирую на него, но это случится не в ближайшее время.

К минусам можно добавить ещё и то, что внести свой вклад в проект (с помощью pull request) с какими-либо изменениями тоже затруднительно в случае, если не знаешь Си на должном уровне.

Workerman

Если и проигрывает в скорости своему конкуренту (речь о Swoole), то не сильно ощутимо и разницей в ряде сценариев можно пренебречь.

Имеет интеграцию с ReactPHP, что в свою очередь расширяет количество имплементаций инфраструктурных моментов. Для экономии места, минусы опишу вместе с ReactPHP.

ReactPHP

К плюсам можно отнести довольно большое комьюнити и огромное количество примеров. Минусы же начинают проявляться в процессе использования — это концепция Promise.
Если необходимо выполнить несколько асинхронных операций, код превращается в бесконечную портянку вызовов then (вот пример простого подключения к RabbiqMQ без создания exchange/queue и их биндингов).

При некоторой доработке напильником (считается нормой), можно получить реализацию корутин, которая поможет избавиться от Promise hell.

Без проекта recoilphp/recoil использовать ReactPHP, на мой взгляд, не представляется возможным в сколько-нибудь вменяемом приложении.

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

Amp

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

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

К сожалению, при всех плюсах фреймворка, комьюнити относительно небольшое, но при этом есть реализации, например, работы с PostgreSQL, а также всех базовых вещей (файловая система, http клиент, DNS, etc).

Мне ещё не совсем понятна судьба проекта ext-async, но ребята идут с ним в ногу. Что из этого получится в 3-й версии, покажет время.

Приступаем к реализации

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

Для начала немного формализуем требования:

• Асинхронный обмен сообщениями (само понятие message можно разделить на 2 типа)
  
  • command: указание на необходимость выполнить задачу. Не возвращает результат (как минимум, в случае асинхронного общения);
  • event: сообщает о каком-либо изменении состояния (например, в следствии команды).
• Неблокирующий формат работы с I/O;
• Возможность легко увеличить количество обработчиков;
• Возможность писать обработчики сообщений на любом языке.

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

По списку требований лучше всего подходит простая реализация Publish/Subscribe паттерна.
Чтобы обеспечить распределённое выполнение, в качестве брокера сообщений воспользуемся RabbitMQ.

Прототип был написан с помощью ReactPHP, Bunny и DoctrineDBAL.
Внимательный читатель мог заметить, что Dbal использует внутри блокирующие вызовы pdo/mysqli, но на текущем этапе это было не особо важно, так как надо было понимать, что должно получиться в итоге.

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

После непродолжительных изысканий ReactPHP был убран в пользу Amp, так как он относительно простой и очень активно развивается.

RabbitMQ транспорт

Но при всех плюсах Amp была 1 проблема: драйвера для RabbitMQ у Amp готового нет (Bunny поддерживает только ReactPHP).

В теории, Amp позволяет использовать Promise от конкурента. Казалось бы, всё должно быть просто, но для работы с сокетами в библиотеке используется Event Loop от ReactPHP.
В один момент времени, очевидно, не может быть запущено двух разных Event Loop'ов, поэтому я не мог воспользоваться функцией adapt().

К сожалению, качество кода в bunny оставляло желать лучшего и адекватно подменить одну реализацию на другую не получилось. Что бы не останавливать работу было принято решение немного переписать библиотеку так, чтобы она заработала с Amp и не приводила к блокировке потока выполнения.

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

И такой человек нашёлся. Проект PHPinnacle, помимо всего прочего, предоставляет реализацию адаптера, заточенного под использование Amp.

Автора зовут Антон Шабовта, который расскажет про асинхронный php в рамках PHP Russia и про разработку драйверов на PHP fwdays.

PostgreSQL

Вторая особенность работы — взаимодействие с базой данных. В условиях «традиционного» PHP у нас всё просто: есть подключение и все запросы выполняются последовательно.

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

Механизм работы довольно простой:

• мы открываем N соединений при старте (либо отложенная инициализация, не суть);
• при необходимости мы забираем из пула соединение, гарантируя что им не сможет воспользоваться никто другой;
• выполняем запрос и либо уничтожаем подключение, либо возвращаем его в пул (предпочтительно).

Во-первых, это нам позволяет запускать несколько транзакций в один момент времени, а, во-вторых, ускоряет работу за счёт наличия уже открытых соединений. У Amp есть компонент amphp/postgres. Он берёт на себя работу с подключениями: следит за их количеством, сроком жизни и всё это без блокирования потока выполнения.

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

Mutex

Для эффективной и, самое главное, правильной работы приложения необходимо реализовать что-то на подобии мьютексов. Мы можем выделить 3 сценария их использования:

• В рамках одного процесса подойдёт простой in memory механизм без всяких излишков;
• Если мы хотим обеспечить блокировку в нескольких процессах, то можно воспользоваться файловой системой (разумеется, в неблокирующем режиме);
• Если в разрезе нескольких серверов, то тут уже надо думать о чём-то вроде Zookeeper.

Мьютексы нужны для решения проблем, связанных с race condition. Ведь мы не знаем (и не можем знать) в каком порядке у нас будут выполняться задачи, но тем не менее мы должны обеспечивать целостность данных.

Логирование/контексты

Для логирования используется уже ставший стандартом Monolog, но с некоторыми оговорками: мы не можем использовать встроенные хэндлеры, так как они будут приводить к блокировкам.
Для записи в stdOut можно взять amphp/log, либо написать простую отправку сообщений в какой-нибудь Graylog.

Так как в один момент времени у нас может обрабатывать множество задач и при записи логов необходимо понимать, в каком контексте пишутся данные. В ходе экспериментов было принято решение сделать trace_id (Distributed tracing). Суть в том, что вся цепочка вызовов должна сопровождаться сквозным идентификатором, который можно отследить. Дополнительно в момент приёма сообщения генерируется package_id, который указывает именно на полученное сообщение.

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

Terminating

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

Что бы корректно завершить выполнение нам необходимо:

• Отменить подписку на очередь. Иными словами, сделать невозможным приём новых сообщений;
• Доделать все оставшиеся задачи (дождаться резолвинга промисов);
• И только после этого завершить работу скрипта.

Утечки, отладка

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

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

Также, помимо всего прочего, написал простенький watcher, который опционально будет запускаться каждые 10 минут и вызывать gc_collect_cycles() и gc_mem_caches().
Но принудительный запуск сборщика мусора не является чем-то нужным и принципиальным.

Для того, чтобы постоянно видеть использование памяти, в логирование был добавлен стандартный MemoryUsageProcessor.

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

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

Результаты

На свет появился очередной велосипед: php-service-bus, который позволяет реализовать Message Based приложение.

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

composer create-project php-service-bus/skeleton pub-sub-example
cd pub-sub-example
docker-compose up --build -d

Пока сервис запускается, мы можем немного разобраться в том, как устроен пример.

В файле /bin/consumer описана инициализация демона, а также конфигурация запуска.
В директории /src находится 3 файла: Ping выполняет роль команды; Pong: выступает в качестве события; PingService: контейнер, содержащий наши обработчики.
Если остановиться подробнее на PingService, то в нём можно увидеть 2 метода:

    /** @CommandHandler() */
    public function handle(Ping $command, KernelContext $context): Promise
    {
        return $context->delivery(new Pong());
    }

    /** @EventListener() */
    public function whenPong(Pong $event, KernelContext $context): void
    {
        $context->logContextMessage('Pong message received');
    }

• handle является обработчиком команды (для каждой команды в приложении может быть только 1 обработчик). Все обработчики команд помечаются аннотацией @CommandHandler;
  
  • Метод возвращает Promise потому, что в нём используется отправка сообщения в RabbitMQ (с помощью метода delivery()). Соответственно задача будет выполнена только после того, как от RabbitMQ будет получено подтверждение записи.
• whenPong — слушатель события Pong. Слушателей напротив может быть сколько угодно и каждый из них выполняется независимо. Слушатели событий помечаются аннотацией @EventListener;
  

  Привычные события, представленные в современных фреймворках — не совсем события. Это, скорее, хуки, необходимые в качестве точек расширения. События же в контексте php-service-bus невозможно модифицировать, отменить, а ошибка во время обработки в одном подписчике не повлияет на второго.
  

Любой обработчик содержит минимум 2 аргумента: собственно, само сообщение (всегда идёт первым) и контекст. Контекст содержит в себе методы по работе с входящим сообщением, метод отправки сообщения в шину, а также несколько хэлперов (например, логирование).

Наш пример получает команду Ping, при её получении отправляет событие Pong. При получении события записывается лог.

Для примера есть файл, запустив который мы отправим сообщение в RabbitMQ:

tools/ping

Если подвести некоторые итоги, то php-service-bus позволяет относительно легко реализовать приложение, использующее Message based архитектуру.

Ping\Pong, указанный в примере — это, по сути, обычный Hello, world и не несёт в себе никакой ценности.

Чтобы узнать про все возможности, можно обратиться к документации.

Если сама тема будет кому-либо интересна, распишу дополнительно как применяется, например, Saga pattern (Process manager) и какие проблемы можно решить с его помощью.

Ну и как же не померяться

Что касается производительности, то есть сравнение с symfony/messenger.

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