使用扭曲回調的異步/等待語法

我想用Twisted Deferred.addCallback方法使用async/await語法。但正如文檔中所述，addCallback回調被稱爲同步。我已經看到inlineCallbacks用於此目的的裝飾器，但我更喜歡使用async/await語法（如果它甚至可能或有意義）。使用扭曲回調的異步/等待語法

我從pika documentation拿起原來的代碼，但我沒有運氣試圖遷移至異步/ AWAIT語法：

import pika 
from pika import exceptions 
from pika.adapters import twisted_connection 
from twisted.internet import defer, reactor, protocol, task 


async def run_async(connection): 
    channel = await connection.channel() 
    exchange = await channel.exchange_declare(exchange='topic_link',type='topic') 
    queue = await channel.queue_declare(queue='hello', auto_delete=False, exclusive=False) 
    await channel.queue_bind(exchange='topic_link', queue='hello', routing_key='hello.world') 
    await channel.basic_qos(prefetch_count=1) 
    queue_object, consumer_tag = await channel.basic_consume(queue='hello', no_ack=False) 
    l = task.LoopingCall(read_async, queue_object) 
    l.start(0.01) 


async def read_async(queue_object): 
    ch,method,properties,body = await queue_object.get() 
    if body: 
     print(body) 
    await ch.basic_ack(delivery_tag=method.delivery_tag) 


parameters = pika.ConnectionParameters() 
cc = protocol.ClientCreator(reactor, twisted_connection.TwistedProtocolConnection, parameters) 
d = cc.connectTCP('rabbitmq', 5672) 
d.addCallback(lambda protocol: protocol.ready) 
d.addCallback(run_async) 
reactor.run()

這顯然是行不通的，因爲沒有人等待run_async功能。

來源

2017-06-17 user1527491

['ensureDeferred']（https://twistedmatrix.com/documents/current/core/howto/defer-intro.html#coroutines-with-async-await）就是你想要的 –

@ notorious.no我已經嘗試用'defer.ensureDeferred'封裝'cc.connectTCP（'rabbitmq'，5672）'，但它沒有幫助。那是你的意思嗎？謝謝。 – user1527491

@ notorious.no明白了。我必須用ensureDeferred包裝回調本身。我正在封裝回調的結果，這顯然沒有成功。謝謝。 – user1527491

正如notorious.no和Twisted文檔指出的，ensureDeferred是要走的路。不過，你必須包裝回調結果，而不是回調本身，這是我不明白。

這是它的外觀，最終：

def ensure_deferred(f): 
    @functools.wraps(f) 
    def wrapper(*args, **kwargs): 
     result = f(*args, **kwargs) 
     return defer.ensureDeferred(result) 
    return wrapper 


@ensure_deferred 
async def run(connection): 
    channel = await connection.channel() 
    exchange = await channel.exchange_declare(exchange='topic_link', type='topic') 
    queue = await channel.queue_declare(queue='hello', auto_delete=False, exclusive=False) 
    await channel.queue_bind(exchange='topic_link', queue='hello', routing_key='hello.world') 
    await channel.basic_qos(prefetch_count=1) 
    queue_object, consumer_tag = await channel.basic_consume(queue='hello', no_ack=False) 
    l = task.LoopingCall(read, queue_object) 
    l.start(0.01) 


@ensure_deferred 
async def read(queue_object): 
    ch, method, properties, body = await queue_object.get() 
    if body: 
     print(body) 
    await ch.basic_ack(delivery_tag=method.delivery_tag) 


parameters = pika.ConnectionParameters() 
cc = protocol.ClientCreator(reactor, twisted_connection.TwistedProtocolConnection, parameters) 
d = cc.connectTCP('rabbitmq', 5672) 
d.addCallback(lambda protocol: protocol.ready) 
d.addCallback(run) 
reactor.run()

感謝。

來源

2017-06-19 09:43:16 user1527491

使用扭曲回調的異步/等待語法

回答

相關問題