Использование AMQP на примере RabbitMQ

Работа ларька шаурмы очень похожа на работу web-сайта, потому что и там и там посетители выполняют похожие действия: 1 предоставляют данные, которые нужно обработать; 2 получают какие-то данные в ответ. В обычной жизни аналог GET-запросов - это меню заведения и всё то, что мы видим, внешняя оболочка. А POST-запросы состоят из желания сделать заказ, просьбы дать книгу жалоб и так далее; всё, что мы можем запросить у продавца. В чём минус обычного меню?

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

Поэтому ему приходится выяснять, есть ли шаурма в толстом лаваше или, наоборот, в тонком.

Продавец иногда будет отвечать «да», иногда - «нет», и было бы неплохо всё это автоматизировать, чтобы избавиться от лишних операций.

Какой самый простой способ оптимизировать большое количество заказов, например во время бизнес-ланча?

Торговцы шаурмой уже вовсю его используют.

Нужно обеспечить покупателям возможность заказа по звонку (заказать и забрать через 15 минут).

Минус в том, что нам всё ещё нужно обрабатывать эту очередь, и телефонная линия может не справиться с наплывом звонков.

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

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

Всё потому, что очередью никто не управляет.

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

Напротив каждой записи проставляют статус «выполнено» / «не выполнено» и далее берут данные для работы из этой таблички.

Наиболее явные минусы этого способа: 1 очевидно, что мы не блокируем содержимое, т. е. несколько работников (исполнителей) могут взяться за один и тот же заказ; 2 если будет много заказов, а MySQL лежит у нас на том же сервере, что и база (а в основном так и бывает), мы можем просто «положить» нашу базу - и всё пойдёт плохо.

Заказы (и не только) гарантированно записываются в очередь.

Заказы распределяются между поварами.

Если повар, скажем, отошёл или сломал руку, заказ остаётся в очереди. Этот повар может взять заказ позже, или его может взять другой повар.

Заказ блокируется (чтобы один и тот же заказ не взяли два повара одновременно).

Есть отдельные очереди на пиццу/роллы.

Плавно переходим к реализации брокера очередей в виде RabbitMQ.

Это гибридный брокер, он поддерживает несколько протоколов.

Сегодня наиболее распространённый протокол - это AMQP. О нём по большей части и будем вести речь.

Либо исполнитель подписывается на сообщения и, как только они появляются, пытается их исполнить.

Брокер также может возвращать сообщения в очередь, чтобы они оставались там при необходимости.

В большинстве случаев разработчики пишут так: «Спасибо за ваш запрос, мы создадим PDF и отправим его вам на почту».

Если проект небольшой, то логика при этом обычно происходит в MySQL-таблицах.

Но как только встаёт вопрос масштабирования, MySQL даёт сбой - возможны блокировки, отсутствие управления очередью, медленная (по сравнению с memory-based) скорость обработки.

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

Сообщение - это не интерпретируемая брокером единица сущности.

Это может быть как просто строка, так и объект в виде JSON или любая структура, представленная в виде строки. Неважно, что мы туда запишем; мы можем указать тип сообщения (например application/JSON), и тогда при обработке будет легко узнать, что закладывалось в начале.

Сообщение не интерпретируется, оно просто хранится.

Затем оно попадает в exchange, в точку роутинга.

Задача точки роутинга - определить, в какую из очередей должно попасть сообщение.

Вместе с сообщением мы можем прислать какой-то ключ.

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

Есть несколько видов точки роутинга, видов работы: 1 простейший вариант - fanout, когда сообщение попадает во все доступные очереди; 2 direct - полное совпадение ключа.

Мы можем создать очередь для одного ключа, очередь для другого; у очереди может быть несколько ключей - всё как в обычном STP-роутинге; 3 topic - ключ удовлетворяет маске (#*), где * - любое непустое слово (одно), а # - то же самое, что и *, только здесь допустимо и пустое слово; 4 headers - мы имеем возможность указать какой-то тип прямо в message и роутиться по нему (используется, когда нет отдельного ключа). Допустим, у нас есть разные routing keys, которые состоят из нескольких ключей.

Если у нас указаны несколько ключей, например europe.weather, то это сообщение нужно поместить и в europe, и в weather.

Если usa.weather - и в usa, и в weather.

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

TTL & expiration В time to live мы можем указать каждому сообщению или всем сообщениям в очереди (очереди целиком), сколько они будут жить. Например, нам поступает заказ. Мы понимаем, что через 60 минут он уже станет неактуальным (на примере биржи - через час уже значительно изменятся ставки; на примере шаурмы - ни один клиент не будет ждать шаурму час), и мы можем задать TTL, чтобы каждое сообщение, которое пролежит больше 60 минут, исчезало из очереди.

Dead lettering Dead lettering - когда наше сообщение вернулось с ошибкой (блек-джеком) в очередь и нам нужно обработать его не сразу, а с задержкой 5-10 секунд. Такое часто бывает, когда провайдер данных не доступен (т. е. провайдер возвращает нам ошибку, скажем, 404, 504 или что-то подобное). В этом нет ничего страшного, мы возвращаем сообщение в очередь и при возврате можем указать, через какое время этот элемент будет доступен для повторного просмотра.

Чтобы начать работать с Rabbit, нам не нужно никаких суперприблуд, нужен банальный composer.

Пуллим из Docker образ и запускаем его.

Советую пуллить по тегу 3-management.

По этому тегу будет доступен крайний стабильный релиз RabbitMQ; приписка management означает, что он будет поставлен вместе с панелью администрирования в виде Web UI (пользовательский интерфейс, представленный в виде сайта, который запускается в web-браузере).

Это очень удобно, примерно как MySQL сразу с phpMyAdmin.

Здесь мы можем увидеть сразу все наши очереди (queues), все наши соединения с RabbitMQ (connections), каналы (channels), обменники (exchanges) - в общем, всю информацию подробно.

Также здесь мы видим статистику: сколько сообщений было обработано; сколько сообщений не было обработано; какие сообщения не попали ни в одну очередь. В RabbitMQ есть очереди по умолчанию (default), куда попадают все сообщения, которые никому не подошли по роутингу, - мы можем разобраться с ними позже. В PHP всё подключается довольно просто.

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

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

После настройки очереди нам необходимо сделать собственно exchange (точку роутинга) и привязать его к какой-то из очередей (сделать binding). У exchange тоже немного параметров: type (всего четыре типа, о которых мы уже говорили ранее); passive; durable (будем ли хранить юнит при рестарте); auto_delete (очищается ли очередь, когда соединение закрывается).

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

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

Вызываем basic_publish - и всё отправляется сразу в очередь.

Затем мы можем из очереди что-то получить.

Для этого используем basic_get, указываем, из какой очереди получаем, указываем message. В этом message дополнительно хранится (помимо нашего сообщения и данных, которые мы ему задали) информация о том, какой канал был использован и какой используется сейчас, какой использовался routing_key, и в целом всё, что было вокруг нашего сообщения во время его обработки.

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

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

Поэтому и придумали consume - это продолжительное соединение, через которое можно что-то получать.

Получается немного асинхронная модель работы.

Поскольку мы делаем process_message и указываем callback-функцию, мы должны эту функцию обговорить и обозначить.

Она не сильно сложна: нам приходит сообщение, и мы можем с ним что-то сделать.

Также мы можем выйти из этого цикла, т. е. приостановить consume изнутри callback'а.

Для этого нам необходимо, как указано в примере, проверять на body.

Если в body, как в примере, нам пришёл quit, нужно выйти.

Чтобы выйти, мы через message возвращаемся к объекту channel, который содержится в свойстве сообщения, и берём basic_cancel с каким-то тегом. Consumer_tag показывает, кто взял это сообщение для обработки. Соответственно, мы выходим - и наш цикл «падает», заканчивается.

Цикл while нам всё равно понадобится, но решение через consume (против решения через постоянный запрос на получение сообщения) более оптимально, т. к. соединение не создаётся для каждого запроса.

Здесь register_shutdown_function - стандартная функция PHP, в которой мы можем обозначить callback-функцию (например shutdown), которая будет выполнена по окончании работы скрипта.

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

Lazy queues (3.9+) Если мы переживаем за сохранность данных, используем медленные очереди. Хотя я не считаю, что это best practice, т. к. страдает производительность. Мы получаем сохранение на диск, что не очень быстро, но это гарант того, что данные не потеряются никогда.

Limit queue size with TTL or max-length Ограничиваем размер очереди через ограничение длины сообщения или времени его жизни. Тут всё логично: поскольку мы работаем с памятью, нам не надо, чтобы она «текла» и использовалась не по назначению.