В очередь, ...! Как управлять состоянием системы через события17.01.2023 11:01

Х/ф «Собачье сердце»

Существует множество разных подходов к построению архитектуры серверных приложений. В данной статье рассмотрим Event-driven архитектуру (она же событийно-ориентированная). Рассмотрим основные принципы, как перейти от связей Компонент А <-> Компонент В, к связям через события Компонент А → Событие B → Компонент B, и зачем это нужно.

Немного скучной теории

Любое программное обеспечение можно представить в очень упрощённом виде:

рис. 1

У нас есть программа, которая реализует некоторую бизнес-логику. В подавляющем большинстве случаев, программа не работает сама по себе в вакууме. Она берет определенные данные (входящие данные), некоторым образом их преобразует и отдает вовне (исходящие данные).

Входящими данными могут быть:

• Файлы на диске;

• Пользовательский ввод;

• Данные полученные по сети;

• Данные полученные от другой программы;

• И т.д.

Исходящими данными могут быть:

• Данные отображаемые на мониторе;

• Распечатка на бумаге;

• Данные переданные по сети;

• Запись в БД.

• И т.д.

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

рис. 2

В данном примере компонентами системы являются блоки A-I.

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

Компоненты A и B принимают данные извне.

Компоненты H и I отдают данные наружу.

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

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

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

Event-driven (или событийно-ориентированная) архитектура основана на событиях, которые происходят с сущностями в нашей системе.

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

рис. 3

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

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

Если мы переходим на событийно-ориентированную архитектуру системы, эта схема могла бы выглядеть следующим образом:

рис. 4

Каждая точка создания заказа (веб-сайт, мобильное приложение и публичное API) генерирует событие создания заказа. Эти события складываются в выделенный для них канал, который слушает специальный обработчик (event handler A).

Наш Event Handler A принимает событие и создает на основе него три новых события, адресованных для разных подсистем: Склада (Handler B), Бонусов (Handler C) и Истории (Handler D), и отправляет их в соответствующие каналы. В эти каналы события могут поступать из разных источников, но для соответствующих обработчиков они являются единственной точкой входа. В такой архитектуре мы можем выделить основные элементы:

• Сущность (Order, User, etc.);

• Событие, связанное с сущностью (Order create event, etc.);

• Генератор событий (Web-сайт, Mobile app, etc.);

• Канал событий (Event channel B, etc.);

• Обработчик событий (Event handler A, etc.).

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

Основные элементы архитектуры

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

Событие — событие может произойти при создании/удалении сущности или изменения её состояния. Примеры: создание заказа, смена его статуса, регистрация пользователя, добавление товара в корзину, прочтение сообщения, отмена заказа, удаление товара из корзины и т.д. Каждое событие представляет из себя простую структуру, которая содержит тип события и данные, которые позволяют это событие нужным образом обработать. В нашем случае это может быть простой JSON объект:

{
	"event_type": "order_create",
	"data": {
		"order_id": 12345
	}
}

Генератор событий — компонент системы, который реализует логику создания или изменения состояния сущности. Например, веб-форма оформления заказа создает сущность заказа и генерирует соответствующее событие.

Канал событий — механизм передачи события от его генератора к обработчику. Это может быть специальный файл, сеть, очередь, БД.

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

Где применим данный подход

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

Данный подход не требует полного переписывания проекта с нуля, он может внедряться в уже существующих системах, построенных на (микро)сервисной архитектуре или в монолитах. Основная задача — разбить связи между компонентами Компонент А → Компонент Б, переведя их на схему: Компонент А, генерация события → канал → обработчик Компонента Б.

Как бы это могло выглядеть в (микро)сервисной архитектуре:

рис. 5

И как бы это могло выглядеть в рамках монолита:

рис. 6

¯⁠\⁠_⁠(⁠ツ⁠)⁠_⁠/⁠¯

Поговорим про инструменты

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

Сущности, события, генераторы

Кажется, здесь всё выглядит достаточно просто: берем свой любимый фреймворк (symfony, laravel, yii, свой самописный велосипед) и разрабатываем на нём.

Каналы событий

Так как нам нужно передавать данные между компонентами/сервисами, будем использовать подходящие для этого инструменты:

• RabbitMQ — популярный брокер очередей, который позволяет строить сложный роутинг сообщений в очередях.

• Kafka — event-streaming платформа, которая позволяет пропускать через себя огромное количество событий, сохранять их и переиспользовать в будущем,  а также при необходимости заново обрабатывать, начиная с какого-то определенного события в прошлом.

• Redis — in-memory key-value СУБД, которая из коробки поддерживает кучу структур данных и операций с ними, на основе которых можно строить собственные реализации очередей. А если не хватает функциональности из коробки — её можно расширить самописными Lua-скриптами.

• SQS — если вы пользуетесь облачной инфраструктурой, можно попробовать SQS или его аналоги. Простой инструмент, которым можно пользоваться из коробки, не нужно заморачиваться с установкой и настройкой — берёте и используете. Облачный провайдер берёт заботы по администрированию на себя.

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

Обработчики событий

А теперь самое интересное: как же нам обрабатывать информацию из очередей? Самым очевидным решением являются демоны — программы/скрипты, которые работают в фоне в системе, они постоянно считывают информацию из очереди, и при получении сообщения оттуда — обрабатывают его.

Все мы знаем, что PHP был создан для того, чтобы постоянно умирать :)

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

Для того, чтобы скрипт запустился, достаточно реализовать нашу бизнес-логику внутри бесконечного while цикла.

getHttpInput();
        if (false === $this->validateInput($input)) {
            throw new HttpException('Wrong data', 400);
        }
		// Создаем нашу сущность Message на основе данных из вебхука
        $message = MessageBuilder::build($input);
        $message->save(); // сохраняем в БД

		// Добавляем сообщение в движок полнотекстового поиска
        FullTextSearch::addToIndex($message); 
		// Считаем статистику с учетом нового сообщения
        Statistics::add($message);

		// Проверяем текст сообщения, если там есть запрос на погоду
		// идем во внешнее API, запрашиваем погоду на завтра.
		// После этого отправляем прогноз погоды
		// пользователю через Telegram API
        if (TextMatch::contains($message->text, 'погода')) {
            $weather = WeatherAPI::getTomorrowWeather();
            TelegramAPI::send($message->userId, $weather);
        }
        
		// Отвечаем платформе Telegram, что успешно обработали сообщение
        return HttpResponse('ok', 200);
    }

}

getHttpInput();
        
		// Закидываем в очередь наш вебхук
		Queue::push('WebhookCreate', $input);
				
		// Отвечаем платформе Telegram, что успешно обработали сообщение
        return HttpResponse('ok', 200);
    }

}

isSafePoint()) {
    		$webhook = Queue::pull('WebhookCreate');
			if (null === $webhook) {
				sleep(1);
				continue;
			}

			if (false === $this->validateInput($webhook)) {
                Log::error('Invalid webhook format');
				continue;
            }
    		// Создаем нашу сущность Message на основе данных из вебхука
            $message = MessageBuilder::build($webhook);
            $message->save(); // сохраняем в БД

    		//Генерируем соответствующие события
    		Queue::push('FullTextSearchIndexAdd', $message);
    		Queue::push('StatisticsAdd', $message);
    		Queue::push('WeatherAPIRequest', $message);
		}
	}
}

isSafePoint()) {
			$message = Queue::pull('WeatherAPIRequest');
			if (null === $message) {
				sleep(1);
				continue;
			}

			// Идём во внешнее API, запрашиваем погоду на завтра.
			// После этого отправляем прогноз погоды
			// пользователю через Telegram API
			$weather = WeatherAPI::getTomorrowWeather();
            TelegramAPI::send($message->userId, $weather);
		}
	}
}