Mensageria e AMQP

Padrões de Mensagens

• A comunicação ocorre pelo pareamento de sockets: um tipo específico de socket para envio é pareado com o tipo correspondente para recebimento.
• Cada par de tipos de socket corresponde a um padrão de comunicação.
• Três padrões mais comuns:
  • request-reply
  • publish-subscribe
  • pipeline

Padrão Request-Reply

• Usado em comunicações tradicionais cliente-servidor / RPC.
• Cliente utiliza um request socket (REQ).
• Servidor utiliza um reply socket (REP).
• Não é necessário chamar as operações listen nem accept.

Padrão Publish-Subscribe

• Clientes subscrevem mensagens específicas que são publicadas no servidor.
• Somente as mensagens subscritas são notificadas.
• Base para sistemas orientados a eventos.
• Implementa um modelo multicast de comunicação.

Padrão Pipeline

• Nós/processos empurram (push) mensagens para um duto.
• Outros nós/processos puxam (pull) mensagens do duto.
• Quem empurra não se importa quem vai puxar a mensagem — e vice-versa.
• Objetivo: performance no fluxo de mensagens.

Notações — RabbitMQ

Hello World — enviando

import pika

# Conectando com o rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Criando 1 fila
channel.queue_declare(queue='hello')

# Enviando 1 msg para fila
channel.basic_publish(exchange='',
                       routing_key='hello',  # nome da fila
                       body='Hello World!')
print(" [x] Sent 'Hello World!'")
connection.close()
        

Hello World — recebendo

import pika

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

channel.queue_declare(queue='hello')

def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)

channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
        

Um novo pedido, vários interessados

Um único evento novo_pedido publicado no fanout — cada fila decide o que fazer com ele.

Fanout — publicando logs (emit_log.py)

import pika
import sys

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

channel.exchange_declare(exchange='logs', exchange_type='fanout')

message = ' '.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange='logs', routing_key='', body=message)
print(" [x] Sent %r" % message)
connection.close()
        

Declara um exchange fanout — ele faz o broadcast da mensagem para todas as filas ligadas a ele.

Fanout — assinando logs (receive_logs.py)

import pika

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

channel.exchange_declare(exchange='logs', exchange_type='fanout')

result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue

channel.queue_bind(exchange='logs', queue=queue_name)

print(' [*] Waiting for logs. To exit press CTRL+C')
        

Uma fila com nome aleatório e temporário (exclusive=True) é criada e ligada (binding) ao exchange.

Direct × Fanout — uma rede de sensores

Referências bibliográficas

• TANENBAUM, A. S.; VAN STEEN, M. Distributed Systems. 3. ed. 2018. ISBN 978-90-815406-2-9. distributed-systems.net
• RABBITMQ. RabbitMQ Tutorials. rabbitmq.com/tutorials