Библиотека для реализации паттерна pub/sub для фреймворка Phalcon

Библиотека реализует событийную архитектуру приложений (Event-Driven Architecture). Работает с фреймворком Phalcon 3.x, но при желании можно легко адаптировать под другие фреймворки.

Рабочий пример можно посмотреть вот здесь: https://github.com/chocofamilyme/pubsub/tree/master/examples

Возможности

• Транзакционное сохранение моделей ORM и публикация события
• Публикация событий без транзакции
• Подписка на события
• Повторная отправка события в ту же очередь при необходимости
• Сохранение в общую очередь всех не обработанных и истекших сообщений. Из этой очереди потом можно сохранить куда-то в БД и обработать индивидуально

Требования

• Phalcon 3.x+
• PHP 7.0+

Установка

composer require chocofamilyme/pubsub

Настройка

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

Настройка конфигов

return [
    'eventsource' => [
        'default' => 'rabbitmq',
        
        'drivers' => [
            'rabbitmq' => [
                'adapter'    => 'RabbitMQ',
                'hosts' => [
                    [
                        'host'     => 'host',
                        'port'     => 5672,
                        'user'     => 'user',
                        'password' => 'password',
                        'vhost'    => '/',
                    ],
                ],
    
                // Не объязательные параметры
                'heartbeat'          => 60,
                'read_write_timeout' => 60,
                'connection_timeout' => 60,
                'wait_timeout'       => 0,
            ],
        ],
    ],
];

Полный список смотрите - https://github.com/php-amqplib/php-amqplib

Добавляем брокер в DI контейнер

$di = \Phalcon\Di::getDefault();
$config      = $di->get('config')->eventsource;
$config      = $config->drivers[$config->default];

$serviceName = $di->get('config')->domain;
$cache       = $di->get('cache');

$di->setShared('eventsource',
    function () use ($config, $serviceName, $cache) {
        $adapter = $config->adapter;
        $config  = array_merge($config->toArray(), ['app_id' => $serviceName]);
        $class   = 'Chocofamily\PubSub\Provider\\'.$adapter;

        $repeater = new Repeater($cache);
        return $class::getInstance($config, $repeater);
    }
);

Здесь $cache объект реализующий интерефейс Phalcon\Cache\BackendInterface. Кэш используется для подсчета количества повторной обработки определенного сообщения.

Таблица параметров с настройками

Ключ	Значение	Описание
connection	По умолчанию PhpAmqpLib\Connection\AMQPLazyConnection::class	php-amqplib
connection_timeout	По умолчанию 3.0 (сек)	Максимальное время на соединение с сервером Rabbitmq
read_write_timeout	По умолчанию 3.0 (сек)	Максимальное время на получение
heartbeat	По умолчанию 60 (сек)	RabbitMQ Doc
keepalive	По умолчанию false	RabbitMQ Doc
context	По умолчанию null	RabbitMQ Doc
prefetch_count	По умолчанию 1	RabbitMQ Doc
no_ack	По умолчанию false	RabbitMQ Doc
durable	По умолчанию true	RabbitMQ Doc
exclusive	По умолчанию false	RabbitMQ Doc
queue	По умолчанию []	RabbitMQ Doc
basic_consume_exclusive	По умолчанию false	RabbitMQ Doc
wait_allowed_methods	По умолчанию null	php-amqplib
wait_non_blocking	По умолчанию true	php-amqplib
wait_timeout	По умолчанию 0	Максимальное время ожидания до получения первого сообщения
exchange_type	По умолчанию topic	RabbitMQ Doc)
app_id	По умолчанию ''	Индификатор приложения которое создает сообщение
sleep	По умолчанию 0.1 секунды	Если wait_non_blocking равно true, то если нету задачи ждать время sleep

Использование

Публикация

Публиковать сообщения можно используя класс Chocofamily\PubSub\Publisher. Минимальный рабочий пример:

$publisher = new Publisher($di->getShared('eventsource'));

$payload = [
	'event_id' => 11995,
	'name' => 'docx',
	'age' => 25
];

$routeKey = 'order.created';

$publisher->send($payload, $routeKey);

Для RabbitMQ переменная $routeKey должна состоять минимум из двух частей разделенных точкой .. Пример order.created. Имя Exchange будет содержать первый блок, т.е. order. После этого если зайдете в админку rabbitmq должен создаться exchange с именем order. Обновленно: начиная с версии 2.* можно указать exchange, которому привяжется маршрут `$routeKey, пример:

$publisher = new Publisher($di->getShared('eventsource'));

$payload = [
	'event_id' => 11995,
	'name' => 'docx',
	'age' => 25
];

$exchange = 'order';
$routeKey = 'order.created';

$publisher->send($payload, $routeKey, $exchange);

Подписка на событие

Для подписки на события используется класс Chocofamily\PubSub\Subscriber. Минимальный рабочий пример:

$params = [
    'queue_name' => 'restapi_orderx',
];

$taskName = 'your_task_name';

$subscriber = new Subscriber($di->getShared('eventsource'), 'order.created.*', $params, $taskName);

$subscriber->subscribe(function ($headers, $body) {
    echo print_r($headers, 1). PHP_EOL;
    echo print_r($body, 1). PHP_EOL;
});

Обновленно: начиная с версии 2.* можно указать exchange и связать с ним маршрут. Теперь можно указать массив маршрутов. Пример:

$params = [
    'queue_name' => 'restapi_orderx',
];

$taskName = 'your_task_name';

$routeKeys = [
    'order.created',
    'order.payed',
];

$exchange = 'order';

$subscriber = new Subscriber($di->getShared('eventsource'), $routeKeys, $params, $taskName, $exchange);

$subscriber->subscribe(function ($headers, $body) {
    echo print_r($headers, 1). PHP_EOL;
    echo print_r($body, 1). PHP_EOL;
});

Чтобы обратно отправить сообщение в очередь необходимо в кэлбэк функции кинуть исключение Chocofamily\PubSub\Exceptions\RetryException. Сообщение может максимум 5 раз обработаться повторно, после этого он попадает в очередь мертвых сообщений (exchange = DLX).

В подписчик можно передавать следующие настройки:

durable: bool — сохранять на диск данные
queue: array — настройки самой очереди
prefetch_count: int — количество предзагрузки сообщений
no_ack: — требуется ли подтверждение сообщений
app_id — уникальный ID приложения. Можно использовать для идентификации откуда собите пошло изначально

Публикация используя транзакции БД

Этот способ необходим для атомарности сохранения сущности в БД и публикования события. Следующая картинка хорошо иллюстрирует как это работает:

Для этого необходимо создать таблицу events:

create table events
(
	id serial not null
		constraint events_pkey
			primary key,
	type smallint not null,
	model_id int not null,
	model_type varchar(100) not null,
	exchange   varchar(100) not null,
	routing_key varchar(100) not null,
	payload json not null,
	status smallint not null,
	created_at timestamp default now() not null,
	updated_at timestamp
);

Пример использования:

use Chocofamily\PubSub\Services\EventPrepare;

...

$order = new Order([
    'user_id' => 11166541,
    'status'  => 0,
    'total'   => 5852,
]);

$eventSource = $di->get('eventsource');

$event = new EventPrepare($order, new OrderSerialize(['name' => 'docx']), 1);
$event->up($eventSource, 'order.created.-5');
	

Модель Order должна реализовывать итерфейс ModelInterface.

Обновленно: начиная с версии 2.* можно указать exchange и связпть с ним маршрут. Привер:

use Chocofamily\PubSub\Services\EventPrepare;

...

$order = new Order([
    'user_id' => 11166541,
    'status'  => 0,
    'total'   => 5852,
]);

$eventSource = $di->get('eventsource');

$routeKey = 'order.created.-5';
$exchange = 'order';

$event = new EventPrepare($order, new OrderSerialize(['name' => 'docx']), 1);
$event->up($eventSource, $routeKey, $exchange);
	

Метод up работает так

• db transaction start
• order->save();
• eventModel->save()
• db transaction commit
• event publish

Повторная отправка события

Для повторной отправке событие используется класс Chocofamily\PubSub\Services\EventRepeater. Рабочий пример:

use Chocofamily\PubSub\Services\EventRepeater;

...

$dateStart = \DateTime::createFromFormat('Y-m-d', '2018-01-01');

$eventDataProvider = new Chocofamily\PubSub\Provider\Event($di->get('eventsource'), $dateStart);

try {
    $event = new EventRepeater($eventDataProvider);
    $event->retry();
} catch (\Exception $e) {
    $message = sprintf('%d %s in %s:%s', $e->getCode(), $e->getMessage(), $e->getFile(), $e->getLine());
    $di->get('logger')->error($message);
}

Очистка журнала событий

Для очистки событий используется класс Chocofamily\PubSub\Services\EventCleaner с методом clean.

Рабочий пример:

use Chocofamily\PubSub\Services\EventCleaner;

...

try {
    $event = new EventCleaner($di->get('modelsManager'));
    $event->clean();
} catch (ModelException $e) {
    $message = sprintf('%d %s in %s:%s', $e->getCode(), $e->getMessage(), $e->getFile(), $e->getLine());
    $di->get('logger')->error($message);
}

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

use Chocofamily\PubSub\Services\EventCleaner;

...

$dateTime  = new \DateTime();
$dateTime = $dateTime->modify('-1 day');

try {
    $event = new EventCleaner($di->get('modelsManager'), $dateTime);
    $event->clean();
} catch (ModelException $e) {
    $message = sprintf('%d %s in %s:%s', $e->getCode(), $e->getMessage(), $e->getFile(), $e->getLine());
    $di->get('logger')->error($message);
}

@todo

• Написать интерфейс для транзакций и убрать зависимость от фреймворка