Обработка async в python

Question

Я иду из узла, где обработка асинхронного дизайна так же просто, как добавление обратного вызова и продолжение вашей жизни. Я пытаюсь написать некоторые приложения на python, где я не имею того же успеха, и я изо всех сил пытаюсь найти, что искать, поскольку нет похожего прямого эквивалента.

Вот пример, когда я запускаю клиент обмена сообщениями MQTT и ожидаю сигнала изменения состояния от датчика.

import paho.mqtt.client as mqtt 
from ouimeaux.environment import Environment 
from ouimeaux.signals import receiver, statechange 

def on_connect(client, userdata, rc): 
    print('Connected with result code '+str(rc)) 
    client.subscribe('lights/#') 

def turn_lights_on(client, userdata, rc): 
    for (x, value) in enumerate(devices['switches']): 
     devices['switches'][x].on() 

def turn_lights_off(client, userdata, rc): 
    for (x, value) in enumerate(devices['switches']): 
     devices['switches'][x].off() 

def reply_with_devices(client, userdata, rc): 
    for (x, value) in enumerate(devices['switches']): 
     client.publish('devices/new', switches[x]) 
    for (x, value) in enumerate(devices['motions']): 
     client.publish('devices/new', motions[x]) 

def on_switch(switch): 
    print "Switch found: ", switch.name 
    devices['switches'].append(switch) 

def on_motion(motion): 
    print "Motion found: ", motion.name 
    devices['motions'].append(motion) 

client = mqtt.Client("wemo_controller") 
client.on_connect = on_connect 
client.message_callback_add('lights/on', turn_lights_on) 
client.message_callback_add('lights/off', turn_lights_off) 
client.message_callback_add('devices/discover', reply_with_devices) 

client.connect('localhost', 1883, 60) 

print 'Running WEMO controller - listening for messages on localhost:1883' 

devices = { 'switches': [], 'motions': [] } 

env = Environment(on_switch, on_motion) 
env.start() 
env.discover(seconds=3) 

switch = env.get_switch('Desk lights') 

@receiver(statechange) 
def motion(sender, **kwargs): 
    print 'A THING HAPPENED' 
    print "{} state is {state}".format(sender.name, state="on" if kwargs.get('state') else "off") 

env.wait() 

client.loop_forever() 

Обе библиотеки, кажется, имеют свой собственный путь держит нить, но это, кажется, что я могу иметь только одну блокировку и слушать одновременно. У меня такое ощущение, что ответ на поток может быть ответом, но я изо всех сил пытаюсь понять, как это реализовать, и не уверен, правильно ли это. Я также смущен тем, что на самом деле делают wait() и loop_forever().

Ответ, который я ищу, - это способ «python» для решения этой проблемы.

Answer 1

Вы можете посмотреть на Twisted framework

«Twisted является управляемыми событиями сетевого движка написанного на Python» Он специально разработан для построения асинхронных сетевых приложений.

В частности, читать на reactor, и используя Deffered() регистрировать обратные вызовы

программирования

Answer 2

Асинхронных было интегрированы в питон в последнее время. Итак, , если вы используете python 3.3, тогда python предоставляет встроенную библиотеку Asyncio специально для этой цели (которая ранее называлась «тюльпаны»). Если вы используете python 2.7, вы можете использовать Trollius, который является backporting Asyncio. Если вам ничего не подходит, вы, очевидно, можете использовать полнофункциональную инфраструктуру сетевого программирования Twisted, как это предлагается в других ответах.

Answer 3

Я являюсь автором HBMQTT, брокерской/клиентской библиотеки MQTT, которая использует API Python asyncio.

API-интерфейс клиента не нуждается в обратном вызове. Вы можете использовать клиентский API для подписки на какую-либо тему, а затем запустить цикл для чтения и обработки входящих сообщений. Что-то вроде:

import asyncio 
from hbmqtt.client import MQTTClient 

C = MQTTClient() 

@asyncio.coroutine 
def test_coro(): 
    yield from C.connect(uri='mqtt://localhost/', username=None, password=None) 
    # Adapt QOS as needed 
    yield from C.subscribe([ 
      {'filter': 'lights/on', 'qos': 0x01}, 
      {'filter': 'lights/off', 'qos': 0x01}, 
      {'filter': 'devices/discover', 'qos': 0x01}, 
     ]) 
    while some_condition: 
     # Wait until next PUBLISH message arrives 
     message = yield from C.deliver_message() 
     if message.variable_header.topic_name == 'lights/on': 
      # Lights on 
     elif message.variable_header.topic_name == 'lights/off': 
      # Lights off 
     yield from C.acknowledge_delivery(message.variable_header.packet_id) 
    yield from C.disconnect() 


if __name__ == '__main__': 
    loop=asyncio.get_event_loop() 
    loop.run_until_complete(test_coro()) 

HBMQTT все еще находится в разработке. Это требует Python 3.4.