Communication interprocessus en Python

Question

Quel est le meilleur moyen de communiquer entre deux exécutions Python distinctes ? Les choses que j'ai essayées :

• lecture/écriture sur des tuyaux nommés, par exemple os.mkfifo (c'est un peu comme si c'était un jeu d'enfant)
• dbus services (fonctionnait sur le bureau, mais trop lourd pour le headless)
• sockets (semble trop bas niveau ; il y a sûrement un module de plus haut niveau à utiliser)

Mon exigence de base est de pouvoir exécuter python listen.py comme un démon, capable de recevoir des messages de la part de python client.py . Le client doit simplement envoyer un message au processus existant et se terminer, avec le code de retour 0 en cas de succès et non nulle en cas d'échec (c'est-à-dire qu'une communication bidirectionnelle sera nécessaire).

Answer 1

Consultez une bibliothèque/serveur multiplateforme appelée RabbitMQ. Il est peut-être trop lourd pour une communication entre deux processus, mais si vous avez besoin d'une communication multi-processus ou multi-codebase (avec différents moyens, par ex. one-to-many, files d'attente, etc), c'est une bonne option.

Exigences :

$ pip install pika
$ pip install bson # for sending binary content
$ sudo apt-get rabbitmq-server # ubuntu, see rabbitmq installation instructions for other platforms

Éditeur (envoie des données) :

import pika, time, bson, os

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='logs', type='fanout')

i = 0
while True:
    data = {'msg': 'Hello %s' % i, b'data': os.urandom(2), 'some': bytes(bytearray(b'\x00\x0F\x98\x24'))}
    channel.basic_publish(exchange='logs', routing_key='', body=bson.dumps(data))
    print("Sent", data)
    i = i + 1
    time.sleep(1)

connection.close()

Abonné (reçoit les données, peut être multiple) :

import pika, bson

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

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

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

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

def callback(ch, method, properties, body):
    data = bson.loads(body)
    print("Received", data)

channel.basic_consume(callback, queue=queue_name, no_ack=True)
channel.start_consuming()

Exemples basés sur https://www.rabbitmq.com/tutorials/tutorial-two-python.html

Answer 2

J'utiliserais des sockets, mais j'utiliserais Twisted pour vous donner une certaine abstraction, et pour rendre les choses plus faciles. Leur exemple de client/serveur Echo simple est un bon point de départ.

Il suffit de combiner les fichiers, d'instancier et d'exécuter le client ou le serveur en fonction des arguments transmis.

Answer 3

J'ai trouvé ce fil de discussion comme l'un des premiers résultats sur Python IPC, mais je cherchais quelque chose qui pourrait fonctionner avec AsyncIO . J'ai fini par trouver IPyC qui offrait de belles capacités asynchrones, alors je suis revenu ici pour partager ce joyau. IPyC supporte également les implémentations synchrones.

Il est possible d'utiliser la bibliothèque IPyC à partir de deux processus différents, mais voici un petit exemple avec deux tâches asyncio dans le même fichier. Il utilise TCP:9999 comme port par défaut.

REMARQUE : Cet exemple ne fonctionne pas avec unexpected keyword argument 'loop' sur Python >=3.10 ; cela est dû à un changement d'interface dans asyncio. J'ai testé avec la version 3.9.

import asyncio
import ipyc
import json
import logging
logging.basicConfig(level=logging.INFO)  # Set to DEBUG to see inner workings

# IPyC has support for custom (de)serialization; using json as example here
ipyc.IPyCSerialization.add_custom_serialization(list, json.dumps)
ipyc.IPyCSerialization.add_custom_deserialization(list, json.loads)

## Host stuff
host = ipyc.AsyncIPyCHost()

@host.on_connect
async def on_client_connect(connection: ipyc.AsyncIPyCLink):
    logging.info("Got a connection")
    while connection.is_active():
        message = await connection.receive()
        if message:
            logging.info(f"Received: {message}")
    logging.info("Connection closed")

## Client stuff
async def client_task():
    client = ipyc.AsyncIPyCClient()

    link = await client.connect()

    for i in range(3):
        await link.send(["Hello World!", i, 3.14])
        await asyncio.sleep(1)

    await client.close()  # Close the connection
    await asyncio.sleep(1)

## AsyncIO stuff
loop = asyncio.get_event_loop()
loop.create_task(host.start())
loop.run_until_complete(client_task())