ZeroMQ

ZeroMQ

ZeroMQ est une librairie de transfert de données asynchrone de haute-performance. Très utile, elle permet d'envoyer des données via des méthodes qui sort du cadre des sockets habituelles ou des IPC (même si elles sont utilisés).

Pour plus d'informations, nous vous conseillons le site officiel, ainsi qu'un excellent résumé sur le site de Amaury Bouchard

Les différentes parties de cette documentation seront en constante construction, considérez-les comme des notes rapides de recherches. Les différents morceaux de code seront écrits en python (sauf indication contraire).

Il existe une multitude de code source d'exemple sur ce site: http://zguide.zeromq.org/page:all

ZeroMQ et le contexte

Quand vous commencez, il faut définir un "contexte" à l'aide de *zmq.Context*.

Définition d'un nouveau contexte :

context = zmq.Context()

Réutilisation du contexte actuel :

context = zmq.Context.instance()

Attention, en "inproc", utilisez un seul contexte par processus, sinon vous ne pourrez communiquer. Sur les autres méthodes (tcp, ipc, egpm), vous pouvez initier de nouveau contexte.

Très important à retenir, surtout dans un environnement "fork" ou de "thread".

ZeroMQ avec une socket de type PAIR

La socket de type PAIR permet de joindre deux sockets de façon exclusive. Aucun autre "PAIR" ne peut se connecter en plus. Si vous avez plusieurs threads avec différents PAIR, il est normal qu'un seul marche. Les autres ne pourront jamais se connecter.

__Avantages__

• PAIR ne supprime aucun paquet même en cas d'inondation (flood) de données, il garantit la réception des données
• PAIR semble attendre que le sender et le receiver soit connectés pour s'envoyer des données. Vous avez donc la garantie de l'envoi et de la réception de vos données.(*1)

__Désavantages__

• PAIR ne peuvent être que deux.

Exemple de code "serveur"

import zmq from zmq.eventloop import ioloop, zmqstream

context = zmq.Context.instance()

sender = context.socket(zmq.PAIR) sender.connect("inproc://mon_pairing")

with open("votre_fichier", 'rb') as file:

       for buffer in iter(lambda: file.read(65536), b):
               sender.send(buffer)

file.close() return(True)

Exemple de code "client"

import zmq from zmq.eventloop import ioloop, zmqstream

context = zmq.Context.instance()

receiver = context.socket(zmq.PAIR) receiver.bind("inproc://mon_pairing")

while receiver.poll():

     buffer = receiver.recv()
     print(buffer)

• 1 = En tout cas, dans les nombreux tests effectués, il n'y a eu aucune perte de données ni même de retard dans la réception.

ZeroMQ avec une socket de type PUB/SUB

Si vous souhaitez avoir plusieurs sockets en écoute, vous pouvez utiliser PUBLISH et SUBSCRIBE.

Exemple de la partie serveur PUBLISHER (corps seulement)

context = zmq.Context.instance()

sender = context.socket(zmq.PUB) sender.bind("inproc://thread")

while (...)

Exemple de la partie client SUBSCRIBE (corps seulement)

context = zmq.Context.instance()

receiver = context.socket(zmq.SUB) receiver.connect("inproc://thread") receiver.setsockopt(zmq.SUBSCRIBE, b)

while (...)

Le setsockopt "SUBSCRIBE" à (vide) défini qu'on souhaite l'ensemble des données (aucun filtre) Vous pouvez définir plusieurs clients avec les mêmes éléments. Chaque client va recevoir les données du "PUBLISHER". Il est important de définir le setsockopt SUBSCRIBE après le connect.

Important à savoir, les sockets de type "PUB/SUB" comme ROUTER/DEALER/etc. vont supprimer des paquets si vous envoyez trop de données en même temps. Pour éviter ce genre de problème, définissez les "Watermarks" (HWM) à 0 (illimité). Il est important de définir les watermarks APRÉS la socket et AVANT le connect ou bind.

Exemple

context = zmq.Context.instance()

sender = context.socket(zmq.PUB) sender.setsockopt(zmq.SNDHWM, 0) sender.setsockopt(zmq.RCVHWM, 0) sender.bind("inproc://thread")

while (...) context = zmq.Context.instance()

receiver = context.socket(zmq.SUB) receiver.setsockopt(zmq.SNDHWM, 0) receiver.setsockopt(zmq.RCVHWM, 0) receiver.connect("inproc://thread") receiver.setsockopt(zmq.SUBSCRIBE, b)

while (...)

Il est important de respecter cette hiérarchie (socket, setsockopt HWM, connect/bind, setsockopt SUBSCRIBE). Sans cela, vous aurez soit des pertes de paquets, soit aucune information. Important aussi: il faut démarrer les instances clientes (SUB) AVANT le publisher. Sinon, il peut manquer des informations en tout début d'envoi ou de réception.