Как именно JMS-брокер RabbitMQ обеспечивает взаимодействие разных приложений: смотрим на примере Python-сервисов, запущенных в Google Colab для публикации и потребления сообщений из очередей облачной платформы cloudamqp.com.
Постановка задачи
Хотя системный, и, тем более, бизнес-аналитик и не пишет исходный код, некоторые архитектурные концепции, нужные в работе на проектах интеграции информационных систем, можно понять, заглянув под капот технологии. Поэтому сегодня посмотрим, как именно JMS-брокер сообщений RabbitMQ обеспечивает взаимодействие разных приложений. Для этого я написала 2 небольших Python-сервиса в Goggle Colab для публикации и потребления сообщений из очередей RabbitMQ, развернутых на облачной платформе cloudamqp.com.
Напомню, Google Colab – это интерактивная среда для написания и интерпретации Python-кода. Как она работает, я ранее писала здесь. Свой экземпляр брокера сообщений RabbitMQ развернем в облачной платформе cloudamqp.com с бесплатным планом. Как она устроена, и какие возможности предоставляет пользователю, смотрите в прошлой статье.
Для демонстрации возьмем пример из моего недавнего материала про RabbitMQ и немного упростим его. Пусть приложением-продюсером, которое создает данные и отправляет их в RabbitMQ, будет умный датчик с 2-мя видами измерений: температура и давление. Хотя эти данные и считывает одно приложение-потребитель, предположим, что оно обрабатывает их по-разному, в разных функциях. Чтобы маршрутизировать сообщения разных измерений в разные очереди temperature и pressure, промаркируем сообщение с разными измерениями ключом маршрутизации (temperature и pressure) соответственно и будем принимать их на прямой тип обменника (Direct), который обеспечивает простую маршрутизацию по точному значению ключа.
Пусть данные от умного датчика, т.е. приложения-продюсера, отправляются в JSON-формате по следующей схеме
{ "$schema": "http://json-schema.org/draft-04/schema#", "type": "object", "properties": { " device_id": { "type": "integer" }, "measure": { "type": "string" }, "value": { "type": "integer" } }, "required": [ " device_id", "measure", "value" ] }
Чтобы не заполнять данные по измерениям вручную, в Python-скрипте для продюсера будем использовать оператор автоматической генерации случайных значений (random).
Подготовка RabbitMQ и создание приложения-продюсера
Теперь подготовим свой облачный экземпляр RabbitMQ. Создадим обменник прямого типа под названием sensor_exchange и 2 очереди: temperature и pressure.
Далее напишем код приложения продюсера в интерактивном блокноте Google Colab. Это Python-приложение отправляет JSON-сообщение в обменник RabbitMQ со случайными значениями температуры или давления каждые 3 секунды:
!pip install pika import pika import json import random import time # Connect to RabbitMQ server connection = pika.BlockingConnection(pika.URLParameters('здесь_должна_быть_URL-адрес_вашего_экзампляра_RabbitMQ:порт_5671_или_5672/виртуальный_хост_вашего_экземпляра')) channel = connection.channel() # Declare an exchange channel.exchange_declare(exchange='sensor_exchange', exchange_type='direct') # Prepare a list of divices devices =[random.randint(0, 100) for i in range(100)] # Prepare a list of possible routing keys routing_keys = ['pressure', 'temperature'] while True: # Prepare random message in JSON format measure=random.choice(routing_keys) data = {'device': random.choice(devices), 'measure': measure, 'value': random.randint(0,100)} message = json.dumps(data) # Send the message to the exchange channel.basic_publish(exchange='sensor_exchange', routing_key=measure, body=message) print(f' [x] Sent {message}') # Sleep for 3 seconds time.sleep(3) channel.close() connection.close()
В Google Colab этот код лучше распределить по 3-м ячейкам:
- в 1-ой ячейке будет установка и импорт Python-библиотек для работы с RabbitMQ (pika), а также форматом JSON, операторами random и времени;
- во 2-ой ячейке код подключения к RabbitMQ и публикации сообщений;
- в 3-ей ячейке закрытие канала и активного подключения.
Запустим на исполнение этот код, выполнив последовательно первые 2 ячейки, и увидим вывод сгенерированных и отправленных RabbitMQ сообщений в области отладки в Colab:
Основы архитектуры и интеграции информационных систем
Код курса
OAIS
Ближайшая дата курса
5 ноября, 2024
Продолжительность
16 ак.часов
Стоимость обучения
36 000 руб.
Посмотрим в GUI-интерфейсы платформы cloudamqp, как сообщения, отправленные на один обменник, попадают в разные очереди по ключу маршрутизации:
Также в GUI-интерфейсе платформы можно посмотреть статистику по активным подключениям, открытым каналам и общий обзор опубликованных и считанных сообщений.
Потребление сообщений
Теперь напишем код приложения-потребителя, которое не просто потребляет данные из двух разных очередей RabbitMQ, но и обрабатывает принятые значения измерений, маркируя состояние устройства отметкой OK или ошибкой ERROR:
!pip install pika import pika import json import random from time import sleep # Connect to RabbitMQ server connection = pika.BlockingConnection(pika.URLParameters('здесь_должна_быть_URL-адрес_вашего_экзампляра_RabbitMQ:порт_5671_или_5672/виртуальный_хост_вашего_экземпляра')) channel = connection.channel() # Declare an exchange channel.exchange_declare(exchange='sensor_exchange', exchange_type='direct') temperature_queue = channel.queue_declare(queue='temperature') pressure_queue = channel.queue_declare(queue='pressure') channel.queue_bind(exchange='sensor_exchange', queue=temperature_queue.method.queue, routing_key='temperature') channel.queue_bind(exchange='sensor_exchange', queue=pressure_queue.method.queue, routing_key='pressure') def on_temperature_message(ch, method, properties, body): #print("Температура ", json.loads(body)) dictDataTemperature = json.loads(body) temperature_state = 'состояние температуры' temperature_value=int(dictDataTemperature['value']) if ((temperature_value) <= 80) and ((temperature_value) >= 20) : temperature_state = 'OK' else: temperature_state = 'ERROR' print("Устройство № ", dictDataTemperature["device"], "состояние температуры", temperature_state, "значение ", temperature_value) def on_pressure_message(ch, method, properties, body): #print("Pressure: ", json.loads(body)) dictDataPressure = json.loads(body) pressure_state = 'состояние давления' pressure_value=int(dictDataPressure['value']) if ((pressure_value) <= 50) and ((pressure_value) >= 10) : pressure_state = 'OK' else: pressure_state = 'ERROR' print("Устройство № ", dictDataPressure["device"], "состояние давления", pressure_state, "значение ", pressure_value) print('Waiting for messages. To exit press CTRL+C') channel.basic_consume(queue=temperature_queue.method.queue, on_message_callback=on_temperature_message, auto_ack=True) channel.basic_consume(queue=pressure_queue.method.queue, on_message_callback=on_pressure_message, auto_ack=True) channel.start_consuming() channel.close() connection.close()
Аналогично продюсеру, разделим его на 3 ячейки в Google Colabб сохранив в отдельном блокноте. Чтобы потребление сообщений работало, следует параллельно запустить их публикацию. Поэтому будет 2 разных ipynb-файла. Посмотрим, как работает потребление сообщений из 2-х очередей.
При публикации и потреблении сообщений пропускную способность обменника можно посмотреть в GUI платформы cloudamqp:
Используемый обменник sensor_exchange имеет отличные от нуля значения скоростей публикации и потребления сообщений.
Весь код, приведенный в этой статье, работает и вы можете повторить все действия, настроив свой облачный экземпляр брокера сообщений и подставив нужные учетные данные. Надеюсь, после этого краткого ликбеза стали понятные принципы работы этого популярного JMS-брокера сообщений, который часто используется для интеграции ИС и асинхронного взаимодействия микросервисов в виде приложений-потребителей и продюсеров. Повторить подобный пример для Apache Kafka можно, прочитав мою новую статью.
Основы архитектуры и интеграции информационных систем
Код курса
OAIS
Ближайшая дата курса
5 ноября, 2024
Продолжительность
16 ак.часов
Стоимость обучения
36 000 руб.
Подробнее познакомиться со всеми рассмотренными темами, а также другими основами архитектуры и интеграции информационных систем вы сможете на курсах Школы прикладного бизнес-анализа в нашем лицензированном учебном центре обучения и повышения квалификации системных и бизнес-аналитиков в Москве: