Начало работы с RabbitMQ: облачное связующее ПО, ориентированное на сообщения

В этой статье рассказывается о RabbitMQ, промежуточном программном обеспечении брокера сообщений, рекомендованном OpenStack для облачных развертываний. Он соответствует стандартам AMQP и разработан в Erlang. Примеры кода разработаны с использованием библиотек Python и PIKA .

1. Message Broker

Посредник сообщений — это программный компонент, который обеспечивает связь между приложениями в кластере корпоративных приложений. Он также известен как Message Oriented Middleware (MOM) в сервис-ориентированной архитектуре (SOA). Приложения в корпоративном кластере используют брокер сообщений, такой как почтовый обмен или почтовое отделение, для отправки сообщений в другие приложения.

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

 2. Модель RabbitMQ

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

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

Система RabbitMQ ECO

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

2.1 Обмены

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

Обмены могут быть настроены на длительный срок, так что они будут выдерживать перезапуски сервера RabbitMQ. Обмены могут быть «внутренними» для сервера, так что они могут быть опубликованы другими обменами внутри сервера. Биржи могут быть настроены на автоматическое удаление, при этом очереди больше не связаны с ним.

2.2 Очереди

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

Очереди можно настроить на длительный срок, чтобы они могли выдержать перезапуски сервера RabbitMQ. Он также может быть настроен как «исключительный» для соединения, после того как соединение закрыто, очередь будет удалена. Автоматически удаляемые очереди удаляются, когда ни один потребитель не подписан на очередь.

2.3 Привязки

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

3. Производитель и Потребитель

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

3.1 Продюсер

Согласно дизайну RabbitMQ, производитель отправляет сообщение на обмен. Поведение обмена определяется типом обмена, route_key & bindings в очередях сообщений. Давайте посмотрим, как создать продюсера, используя библиотеку Python pika. Если у вас нет установки RabbitMQ, обратитесь к разделу Приложение-А, чтобы установить и настроить RabbitMQ.

import pika 

""" Establish connection and declare the exchange"""
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange="SMS-Exchange",exchange_type="direct")

""" Publish a Message on the Channel """
Msg = "Balance as of today is 200.00$"
channel.basic_publish(exchange = "SMS-Exchange",routing_key="Sms-Alert",body=Msg)

"""Close the communication channel """
connection.close()

import pika

""" define a callback function to receive the message """
def callback(channel,method,properties,body):
    print "[X] Received Message ",body

""" Establish connection """
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')
channel = connection.channel()

""" Declare the queue & bind to the exchange with routing key """
channel.queue_declare(queue="SMS-Queue")
channel.queue_bind(exchange="SMS-Exchange",queue="SMS-Queue",routing_key="SMS-Alert")

""" register the call back function """
channel.basic_consume(callback,queue="SMS-Queue",no_ack=True)

""" listen to the queue """
channel.start_consuming()

import pika
credits = pika.PlainCredentials('scott','tiger')
params = pika.ConnectionParameters('10.0.0.1',5672,'/',credits)
connection = pika.BlockingConnection(parameters=params)

import pika 

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

""" Declare a exchange with type as direct """
channel.exchange_declare(exchange="Direct-X",exchange_type="direct")

""" Publish a Message on the Channel """
Msg = raw_input("Please enter the message :")
channel.basic_publish(exchange = "Direct-X",routing_key="Key1",body=Msg)

"""Close the communication channel """
connection.close()

import pika

""" define a callback function to receive the message """
def callback(channel,method,properties,body):
    print "[X] Received Message ",body

""" Establish connection """
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')
channel = connection.channel()

""" Declare the queue & bind to the exchange with routing key """
channel.queue_declare(queue="Direct-Q")
channel.queue_bind(exchange="Direct-X",queue="Direct-Q",routing_key="Key1")
""" register the call back function """
channel.basic_consume(callback,queue="Direct-Q",no_ack=True)

""" listen to the queue """
channel.start_consuming()

import pika 

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

""" Declare a exchange with type as direct """
channel.exchange_declare(exchange="Fanout-X",exchange_type="fanout")

""" Publish a Message on the Channel """
Msg = raw_input("Please enter the message :")
channel.basic_publish(exchange = "Fanout-X",routing_key="Key1",body=Msg)

"""Close the communication channel """
connection.close()

import pika

""" define a callback function to receive the message """
def callback(channel,method,properties,body):
    print "[X] Received Message ",body

""" Establish connection """
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')
channel = connection.channel()

""" Declare the queue & bind to the exchange with routing key """
channel.queue_declare(queue="Fanout-Q1")
channel.queue_bind(exchange="Direct-X",queue="Fanout-Q1",routing_key="Key1")
""" register the call back function """
channel.basic_consume(callback,queue="Fanout-Q1",no_ack=True)

""" listen to the queue """
channel.start_consuming()

import pika

""" define a callback function to receive the message """
def callback(channel,method,properties,body):
    print "[X] Received Message ",body

""" Establish connection """
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')
channel = connection.channel()

""" Declare the queue & bind to the exchange with routing key """
channel.queue_declare(queue="Fanout-Q2")
channel.queue_bind(exchange="Fanout-X",queue="Fanout-Q2",routing_key="Key1")
""" register the call back function """
channel.basic_consume(callback,queue="Fanout-Q2",no_ack=True)

""" listen to the queue """
channel.start_consuming()

import pika 

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

""" Declare a exchange with type as direct """
channel.exchange_declare(exchange="Topic-X",exchange_type="topic")

""" Publish a Message on the Channel """
Msg = raw_input("Please enter the message :")
channel.basic_publish(exchange = "Topic-X",routing_key="in.sales.put",body=Msg)
channel.basic_publish(exchange = "Topic-X",routing_key="in.sales.post",body=Msg)
channel.basic_publish(exchange = "Topic-X",routing_key="in.sales.delete",body=Msg)
channel.basic_publish(exchange = "Topic-X",routing_key="in.rnd.put",body=Msg)
channel.basic_publish(exchange = "Topic-X",routing_key="in.rnd.post",body=Msg)
channel.basic_publish(exchange = "Topic-X",routing_key="in.rnd.delete",body=Msg)
"""Close the communication channel """ 
connection.close()

import pika

""" define a callback function to receive the message """
def callback(channel,method,properties,body):
    print "[X] Received Message ",body

""" Establish connection """
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')
channel = connection.channel()

""" Declare the queue & bind to the exchange with routing key """
channel.queue_declare(queue="Topic-Q1")
channel.queue_bind(exchange="Topic-X",queue="Topic-Q1",routing_key="*.*.post")
""" register the call back function """
channel.basic_consume(callback,queue="Topic-Q1",no_ack=True)

""" listen to the queue """
channel.start_consuming()

import pika

""" define a callback function to receive the message """
def callback(channel,method,properties,body):
    print "[X] Received Message ",body

""" Establish connection """
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')
channel = connection.channel()

""" Declare the queue & bind to the exchange with routing key """
channel.queue_declare(queue="Topic-Q2")
channel.queue_bind(exchange="Topic-X",queue="Topic-Q2",routing_key="in.rnd.*")
""" register the call back function """
channel.basic_consume(callback,queue="Topic-Q2",no_ack=True)

""" listen to the queue """
channel.start_consuming()

import pika

""" A dictionary to store the customer headers """
header={}
header['Source']='Core-Banking'
header['Destination']='Message-System'

""" Establish Blocking Connection with RabbitMQ Server """
conn = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
chan = conn.channel()

""" Declare an exchange and set type as headers """
chan.exchange_declare(exchange="Header-X",exchange_type="headers")

"""Set the headers using BasicProperties api """
prop = pika.BasicProperties(headers=header)

"""Publish the message to the exchange """
msg = "Hello Message-System, I am Core-Banking"
chan.basic_publish(exchange='Header-X',routing_key="Key1",body=msg,properties=prop)
chan.close()

que='Header-Q'
exg = 'Header-X'
key='Source'
val='Core-Banking'

""" Declare a call back function to receive the message """
def callback(ch,method,properties,body):
 print "[X] Received Msg",body

""" Establish blocking connection to the server """
conn = pika.BlockingConnection( pika.ConnectionParameters(host='localhost'))
chan = conn.channel()

""" Declare a Queue """
chan.queue_declare(queue=que)

"""Bind the queue to exchange and set arguments to match any of the given header """
chan.queue_bind(exchange=exg,queue=que, arguments={key:val,'x-match':'any'})

print '[*] Wait for messages. To exit press CTRL+C'

""" Link the callback function to the queue """
chan.basic_consume(callback,queue=que,no_ack=True)

"""Wait for the messages to arrive """
chan.start_consuming()

rabbitmq-plugins enable rabbitmq_management

rabbitmqctl add_vhost <virtual_host_name>

rabbitmqctl add_user <username> <password>

Eg. rabbitmqctl add_user scott tiger

set_permissions [-p vhostpath] {user} {conf} {write} {read}

Eg: set_permission -p "/" "scott" ".*"  ".*" ".*"

sudo iptables -I INPUT -p tcp --dport 5672 --syn -j ACCEPT
sudo iptables -I INPUT -p tcp --dport 15672 --syn -j ACCEPT