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

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

В обычной жизни аналог GET-запросов — это меню заведения и всё то, что мы видим, внешняя оболочка.

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

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

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

Какой самый простой способ оптимизировать большое количество заказов, например во время бизнес-ланча? Торговцы шаурмой уже вовсю его используют. Нужно обеспечить покупателям возможность заказа по звонку (заказать и забрать через 15 минут). Минус в том, что нам всё ещё нужно обрабатывать эту очередь, и телефонная линия может не справиться с наплывом звонков. Тогда клиенты останутся без заказов, и мы потеряем в выручке.

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

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

Если этого не сделать, у нас появляются следующие риски:

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

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

Посмотрим, что предлагается в коробке большинства брокеров (и в RabbitMQ тоже), на примере решаемой задачи.

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

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

Есть и другие протоколы, например MQTT, когда на TCP/IP повешена возможность подписки и получения, исполнения сообщений.

В AMQP есть несколько дефолтных сущностей, с которыми и происходит работа:

Если смотреть принципиально (сверху), то AMQP-протокол вообще и RabbitMQ в частности представляют собой следующую модель.

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

Самый банальный пример такой модели — создание PDF.

Почему создание PDF и подобных задач имеет смысл закинуть в очередь? Обычно они занимают много времени, и тратить его, пока наш пользователь находится между моментом совершения заказа и моментом отправки файла, не хочется. В большинстве случаев разработчики пишут так: «Спасибо за ваш запрос, мы создадим PDF и отправим его вам на почту». Если проект небольшой, то логика при этом обычно происходит в MySQL-таблицах. Но как только встаёт вопрос масштабирования, MySQL даёт сбой — возможны блокировки, отсутствие управления очередью, медленная (по сравнению с memory-based) скорость обработки. Реляционная база данных хороша, но лучше хранить сообщения в хранилище побыстрее и более предназначенном для этого. Мы можем записать этот запрос в очередь и уже из очереди выполнять его и отправлять пользователю.

Соответственно, есть несколько видов точки роутинга, видов работы:

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

Кроме того, нужно логировать, создавать отдельную очередь, которая бы писала всеми любимую реляционную базу данных. Best practice для high-load-проектов, когда мы держим только кеш в состоянии памяти (все новые заказы например). Одно сообщение можно отправить в несколько очередей, и одна из очередей будет использоваться для связи с реляционной базой данных и для постоянного хранения в ней — так же, как у тебя, наверно, сделано и везде с Elastic (или MySQL, или PostgreSQL, или чем-нибудь другим, что у тебя используется на backend'е). В случае с Elastic он используется как кеш и связывается с постоянным хранилищем. В случае «смерти» Elastic'а всё будет нормально, мы не потеряем данные, а вот в случае «смерти» базы данных нормально уже не будет.

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

Допустим, у нас есть разные routing keys, которые состоят из нескольких ключей, и тогда нам будут полезны разные вариации exchange. Если у нас указаны несколько ключей, например europe.weather, то это сообщение нужно поместить и в europe, и в weather. Если usa.weather — и в usa, и в weather. Таким образом, мы можем разроутить все сообщения сразу в несколько очередей.

Выглядит это примерно так.

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

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

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

Чтобы создать новую очередь, есть простейшая функция queue_declare, где мы указываем:

Вот и все настройки очереди.

После настройки очереди нам необходимо сделать собственно exchange (точку роутинга) и привязать его к какой-то из очередей (сделать binding).

У exchange тоже немного параметров:

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

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

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

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

Далее используем basic_ack (acknowledge), т. е. указываем базовую систему работы с блокировкой, получаем тестовое сообщение (в нашем случае — body) и можем его посмотреть.

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

Первый параметр — очередь.

Далее:

• consumer_tag — это идентификатор consumer'а;
• nowait — не ждать ответа от сервера (т. е. продолжать получение);
• callback — функция PHP, которая будет вызвана при получении сообщения. Получается немного асинхронная модель работы. Поскольку мы делаем process_message и указываем callback-функцию, мы должны эту функцию обговорить и обозначить.

Она не сильно сложна: нам приходит сообщение, и мы можем с ним что-то сделать. Также мы можем выйти из этого цикла, т. е. приостановить consume изнутри callback'а. Для этого нам необходимо, как указано в примере, проверять на body. Если в body, как в примере, нам пришёл quit, нужно выйти. Чтобы выйти, мы через message возвращаемся к объекту channel, который содержится в свойстве сообщения, и берём basic_cancel с каким-то тегом. Consumer_tag показывает, кто взял это сообщение для обработки. Соответственно, мы выходим — и наш цикл «падает», заканчивается.

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

В итоге этих действий мы добавили к очереди сообщение с текстом, у нас вызывается basic_cancel и закрываются каналы и соединения.

В delivery хранится дополнительная информация о том, как обрабатывалось сообщение.

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

Best practices

Сегодня отказоустойчивость и готовность сервиса к масштабированию — важные критерии, которые отличают по-настоящему хороший сервис. Достичь этого помогают в том числе и очереди сообщений во главе с брокером сообщений. Понимание кейсов, в которых лучше переложить работу с сообщениями на брокер очередей, очень важно в современной разработке. Даже если наш сервис не будет масштабироваться на много worker'ов и много очередей, сама очередь (и работа с ней через брокер) может быть хорошим буфером для снижения нагрузки и гарантом доставки и обработки сообщений, что позволит нашему сервису продолжить обработку без потерь, даже если во время обработки что-то пойдёт не так.

Понравилась статья?

Подпишись на нашу рассылку — стань первым, кто узнает о наборе
на обучающие курсы, митапах и новых открытых вакансиях!