Programmation distribuée

1. Présentation générale

ØMQ est une solution élégante et performante qui va nous permettre de présenter les patrons de conception essentiels liés à cette problématique. Pour l’installer :

$ pip install zmq 

Lorsque l’on écrit un programme et que l’on souhaite utiliser cette bibliothèque, il y a toujours le même type de code à écrire :

>>> from zmq import Context 
>>> context = Context() 
>>> socket = context.socket(METHODE) 

Ce code se trouve dans les deux applications (ou plus) qui communiquent entre elles. Il permet de créer le socket réseau dans le cadre duquel les échanges de messages auront lieu. La méthode à utiliser dépend du patron de conception que l’on souhaite utiliser.

Une fois le socket créé, on peut visualiser ses méthodes :

>>> dir(socket) 

Comme on travaille avec un socket, on utilisera la méthode bind pour écouter sur une interface réseau et un port particulier. Les protocoles supportés sont TCP, UDP, PGM, EPGM, INPROC et IPC. Il existe également bind_to_random_port qui permet de laisser au programme le choix du port. Pour fermer une connexion, il faudra utiliser la méthode unbind. L’attribut closed permet de connaître l’état du socket.

Le programme qui réalisera l’opération de bind sera le serveur, celui qui écoute le port (et il ne peut y en avoir qu’un). L’autre programme devra utiliser connect et sera le client. Suivant le paradigme utilisé, la communication est monodirectionnelle ou bidirectionnelle. Autrement dit, dans certains protocoles, le client et le serveur peuvent à la fois envoyer et recevoir de la donnée, alors que dans d’autres, l’un ne fait que recevoir et l’autre qu’envoyer. Pour fermer une telle connexion, il faut utiliser disconnect.

Les sockets peuvent ensuite envoyer ou recevoir des données et ils supportent plusieurs formats :

1. Installation

Pour installer RabbitMQ, il faut utiliser un dépôt spécifique, car les paquets issus de la distribution, s’ils existent dans le dépôt officiel, sont souvent datés et souffrent de vulnérabilités :

# wget -O - "https://packagecloud.io/rabbitmq/rabbitmq-server/gpgkey" | 
apt-key add - 
# apt update 
# apt install rabbitmq-server 

Pour les autres distributions, il faudra se référer à la documentation officielle (https://www.rabbitmq.com/download.html).

2. Configuration

Lorsque l’on utilise PostgreSQL, par exemple, il faut créer un utilisateur avec son mot de passe ainsi qu’une base de données. On peut alors l’utiliser dans notre programme Python à partir d’une URL telle que :

database_url = 'postgres://user:secret@localhost:5432/db_name' 

Avec RabbitMQ, il s’agit de la même chose :

rabbit_broker_url = 'amqp://user:secret@localhost:5672/vhost' 

Il existe ensuite quelques commandes à connaître pour créer un utilisateur, un vhost et pour paramétrer les permissions :

# rabbitmqctl add_user test 
# sudo rabbitmqctl add_vhost vhost 
# sudo rabbitmqctl set_permissions -p vhost test ".*" ".*" ".*" 

Nous pouvons maintenant utiliser RabbitMQ à partir de l’URL définie plus haut.

Pour terminer, il faut également installer pika :

pip install pika 

3. Introduction

Le premier exemple de code est très basique : un producteur et un consommateur utilisant une file d’attente (queue) pour échanger des messages. Il s’agit essentiellement de voir comment créer une connexion, envoyer ou recevoir des messages.

Voici le producteur :

from pika import BlockingConnection 
from pika.connection import URLParameters 
 
connection = BlockingConnection( 
    URLParameters('amqp://test:test@localhost:5672/vhost')) ...

1. Présentation

Kafka est un système de distribution de message en temps réel utilisé notamment dans les canaux de données mais également en tant que système d’échange dans les systèmes de micro-services. 

Au niveau de l’écosystème Kafka en Python, il existe notamment kafka-python et aiokafka, sa version asynchrone, permettant de créer des producteurs et des consommateurs.

Il existe également le projet faust qui permet de gérer l’équivalent du pattern publish/subscribe.

2. Principes généraux

Kafka est un système de distribution de messages. Ces messages sont isolés dans un topic. Ce topic est un mécanisme de stockage qui conserve l’ordre d’émission des messages et les messages peuvent être vus comme des événements.

Chaque topic peut avoir une à plusieurs partitions et l’ordre des messages est garanti au sein d’une même partition mais pas entre les différentes partitions.

Lorsqu’un producteur produit un message, ce dernier sera émis dans une partition seulement. 

Il existe également la notion de groupe. En effet, dans un système où de multiples parties échangent des messages, le message doit être lu par chaque groupe à l’écoute. Si un groupe n’a qu’un seul consommateur, ce dernier écoutera toutes les partitions....

1. Présentation

Celery est un ordonnanceur de tâche :

Il s’agit d’un outil de haut niveau offrant des fonctions et une syntaxe pour répondre facilement à des besoins d’architecture applicative complexe.

2. Dans quel cas utiliser Celery ?

Le cas classique est lorsque l’on a une fonction qui doit générer un résultat, puis qui doit faire quelque chose annexe avec ce résultat et le renvoyer à l’utilisateur. Si l’on fait ces actions annexes dans le même tunnel de traitement, alors on retarde d’autant la réponse à l’utilisateur. L’idée est donc de mettre ces actions annexes dans des tâches distribuées (exécutées à côté du processus principal). Ainsi, on lance ces tâches qui seront exécutées plus tard, mais on n’attend pas qu’elles soient terminées et l’on peut continuer le processus principal et rendre la main à l’utilisateur.

Les tâches placées dans une file d’attente sont exécutées au plus tôt. Il n’y a cependant aucune garantie de la temporalité de leur exécution ni sur le fait qu’elles réussissent ou échouent.

L’expérience utilisateur est ainsi améliorée, car l’application répond plus vite, mais il doit aussi être tenu au courant qu’il y a des tâches en attente et être notifié lorsque celles-ci se terminent.

Un exemple concret serait une tâche pour envoyer un e-mail ou générer un document à partir des résultats du processus principal.

Il est possible de prévoir un moyen pour Celery de restituer le résultat, ceci permet de résoudre toutes les situations classiques où la programmation asynchrone est efficace, sauf qu’on le fait avec une solution tierce qui permet une mise...

1. Présentation

Pour terminer ce chapitre sur la programmation distribuée, nous allons introduire WAMP (Web Application Messaging Protocol). Il s’agit d’un standard ouvert de WebSocket qui permet à deux applications de s’échanger des messages RPC (Remote Procedure Call, soit appel distant de procédure) ou de type Publish/Subscribe (que nous avons déjà croisé deux fois dans ce chapitre).

La conception d’origine de ce protocole a été réalisée par crossbar.io en 2012 et est relativement utile et facile à mettre en place pour des projets web.

La solution présentée ici utilise massivement la bibliothèque asynchrone de Python. Il est donc conseillé d’avoir parcouru le chapitre Programmation asynchrone : avancée avant de s’attaquer à cette partie.

Pour installer les outils nécessaires :

$ pip install autobahn 

2. WebSocket

En guise d’introduction, nous allons réaliser un WebSocket de type Echo. Ce dernier va donc recevoir de la donnée et la renvoyer telle quelle. Voici ce serveur :

import json 
import logging 
import coloredlogs 
from autobahn.asyncio.websocket import ( 
    WebSocketServerProtocol, 
    WebSocketServerFactory) 
 
 
logger = logging.getLogger("notifications") 
coloredlogs.install(level='DEBUG', logger=logger) 
 
 
class TestServerProtocol(WebSocketServerProtocol): 
 
    def onConnect(self, request): 
        logger.info("Client connecting: {0}".format(request.peer)) 
 
    def onOpen(self): 
        logger.info("WebSocket connection open.") 
 
    def onMessage(self, payload, isBinary): 
        if isBinary: 
            result = self.onMessageBinary(payload) 
        else: 
            payload = payload.decode("utf-8") 
            try: ...