Messagerie avec AMQP à l'aide de kombu

À propos de l'AMQP

AMQP est un protocole pour les services de messagerie. Quelqu'un envoie un message dans la file d'attente et quelqu'un récupère le message de la file d'attente. Je comprends que.

La page suivante est très utile pour plus de détails. Page de l'ingénieur GREE Facile à lire et très agréable.

Cependant, la qualité est incompréhensible sans la déplacer, alors j'ai expérimenté. Les outils utilisés pour l'expérience sont les suivants

• RabbitMQ
• Serveur AMQP
  • https://www.rabbitmq.com/
• kombu
• Une bibliothèque qui utilise AMQP avec Python
  • https://pypi.python.org/pypi/kombu
  • http://kombu.readthedocs.org/en/latest/

La procédure de construction de l'environnement sera décrite plus loin.

L'écran provient du plugin de gestion de Rabbit MQ.

Faire un échange

Selon la page de l'ingénieur GREE, c'est ʻExchange qui reçoit Message, et ʻExchange le transmet à Queue.

Pour le moment, j'ai décidé de le créer à partir de ʻExchange, et voici ce que je pouvais voir dans [Exemple de code](http://kombu.readthedocs.org/en/latest/reference/kombu.html#exchange) de kombu`.

from kombu import Connection,Exchange

exchange = Exchange('foo_exc', type='direct')

with Connection('amqp://guest:guest@localhost:5672//') as c:
    bound = exchange(c.default_channel)
    bound.declare()

ʻExchangeest terminé. En bas,foo_exc` créé par le processus ci-dessus.

Créer une file d'attente

Continuez à créer une file d'attente basée sur Exemple de code.

from kombu import Connection,Exchange,Queue

exchange = Exchange('foo_exc', type='direct')
queue = Queue('bar_queue', exchange=exchange, routing_key='hoge.fuga')

with Connection('amqp://guest:guest@localhost:5672//') as c:
    bound = queue(c.default_channel)
    bound.declare()

La «file d'attente» est créée.

Faire et lancer un message

Enfin, créez un Message et plongez-vous dans ʻExchange`. Voir Référence plutôt qu'un exemple de code.

from kombu import Connection,Exchange

exchange = Exchange('foo_exc', type='direct')

with Connection('amqp://guest:guest@localhost:5672//') as c:
    bound_exc = exchange(c.default_channel)
    msg = bound_exc.Message("Hello, World")
    bound_exc.publish(msg, routing_key='hoge.fuga')

Il semble que «Hello, World» soit correctement inclus. Autour Recevez des messages.

Prendre un message

Le Message qui ne peut pas être pris est juste une poubelle, alors prenez-le. Exemple de code

from kombu import Connection,Exchange,Queue,Consumer

exchange = Exchange('foo_exc', type='direct')
queue = Queue('bar_queue', exchange=exchange, routing_key='hoge.fuga')

def callback(body, message):
    print body
    message.ack()

with Connection('amqp://guest:guest@localhost:5672//') as c:

    with Consumer(c.default_channel, queues=[queue], callbacks=[callback]):
        c.drain_events()

résultat

ubuntu@ubuntu:~$ python consume.py
Hello, World

«Hello, World» s'affiche.

Si vous n'appelez pas message.ack (), le message ne disparaîtra pas de Queue.

Environnement

RabbitMQ

Construit avec Docker + docker-compose

docker-compose.yml

rabbit:
    image: rabbitmq:3-management
    hostname: rabbit001
    ports:
        - "15672:15672"
        - "5672:5672"

kombu

Installer avec pip normalement

pip install kombu