标签: python-asyncio

aiohttp自述文件说：

如果您想对 aiohttp 客户端使用超时，请使用标准 asyncio 方法：yield from asyncio.wait_for(client.get(url), 10)

但这并不能处理 DNS 超时，我猜这是由操作系统处理的。而且with aiohttp.Timeout不处理操作系统 DNS 查找。

在asyncio repo上进行了讨论，但没有最终结论，Saghul 已经制作了aiodns，但我不确定如何将其混合到 aiohttp 中以及是否允许asyncio.wait_for功能。

测试用例（在我的 Linux 机器上需要 20 秒）：

async def fetch(url):
    url = 'http://alicebluejewelers.com/'
    with aiohttp.Timeout(0.001):
        resp = await aiohttp.get(url)

长话短说

有没有办法等待多个期货，并在它们按给定顺序完成时从中产生收益？

很长的故事

假设您有两个数据源。一个为您提供id -> name映射，另一个为您提供id -> age映射。你想要计算(name, age) -> number_of_ids_with_that_name_and_age.

数据太多，无法直接加载，但是两个数据源都支持分页/迭代和排序id。

所以你写类似的东西

def iterate_names():
    for page in get_name_page_numbers():
        yield from iterate_name_page(page)  # yields (id, name) pairs

对于年龄也是如此，然后你迭代iterate_names()和iterate_ages()。

这有什么问题吗？发生的情况是：

• 您请求一页姓名和年龄
• 你得到他们
• 您处理数据直到到达页面末尾，比方说，年龄
• 您请求另一页年龄
• 您处理数据直到...

基本上，您在处理数据时不会等待任何请求。

您可以用来asyncio.gather发送所有请求并等待所有数据，但是：

• 当第一页到达时，您仍在等待其他页面
• 你内存不足了

它asyncio.as_completed允许您在获得结果时发送所有请求并处理页面，但是您将获得无序的页面，因此您将无法进行处理。

理想情况下，会有一个函数发出第一个请求，当响应到来时，发出第二个请求并同时从第一个请求中产生结果。

那可能吗？

我正在学习Python asyncio 用于异步编程。我知道事件循环会监视 Future 对象，直到它们准备好，然后恢复适当的协程以在出现 wait 关键字的点继续执行。

当您使用类似的东西时，这是非常可以理解的asyncio.sleep，因为睡眠函数知道它需要多少时间，因此会知道事件循环，但是对于依赖于网络（例如）等待时间未知的东西会发生什么？。

事件循环如何知道资源何时准备就绪，或者从某个源收集数据需要多少时间？

我已经成功导入/安装 Twistedasyncioreactor并执行一个简单的异步函数：

from twisted.internet import asyncioreactor
asyncioreactor.install()
from twisted.internet import task
from twisted.internet.defer import inlineCallbacks
from twisted.internet.defer import ensureDeferred

async def sleepy(reactor):
    print("SLEEPING")
    await task.deferLater(reactor, 3.0, lambda: None)
    print("done sleep")
    return 42

@task.react
def main(reactor):
    d = ensureDeferred(sleepy(reactor))
    d.addCallback(print)
    return d

例如，我想在所述代码中混合异步库asyncio.sleep。我尝试过以下方法：

from twisted.internet import asyncioreactor
asyncioreactor.install()
from twisted.internet import task
from twisted.internet.defer import inlineCallbacks
from twisted.internet.defer import ensureDeferred

import asyncio

async def sleepy(reactor):
    print("SLEEPING")
    await asyncio.sleep(3)
    print("done sleep")
    return 42

@task.react
def main(reactor):
    d = …

考虑一个单线程 Python 程序。名为“first”的协程在 I/O 上被阻止。随后的指令是“等待第二个”。协程“第二个”是否保证立即执行，直到它在 I/O 上阻塞？或者，“first”可以在调用“second”之前恢复执行（由于 I/O 操作完成）吗？

让我们考虑以下包含异步循环和异步协程的类示例：

import asyncio

class Async:
    def __init__(self):
        self.loop=asyncio.get_event_loop()

    async def function(self, word):
        print(word)
        await asyncio.sleep(1.0)

a=Async()
a.loop.run_until_complete(a.function("hello_world"))

这确实有效。
我想创建一个装饰器，以便我可以简化调用代码的function语法

a.function("hello_world")

我尝试了以下方法：

class Async:
    def __init__(self):
        self.loop=asyncio.get_event_loop()

    def async_loop(f):
        def decorated(*args, **kwargs):
            self.loop.run_until_complete(f(*args, **kwargs))

    @async_loop
    async def function(self, word):
        print(word)
        await asyncio.sleep(1.0)

a=Async()
a.function("hello_world")

此时我收到错误：'NoneType' object is not callable。- 我也尝试在类之外使用装饰器函数，但我得到了同样的错误。我不确定装饰器函数最好位于类内部（作为方法）还是外部。我对 python 还很陌生，所以 Asyncio、装饰器和类中的装饰器对我来说仍然很困惑。任何好心人都会知道如何正确地编写该代码吗？

yield from我怀疑这与 b/w &的差异有关await。然而，除了新对象被指定为 an 之外async_generator，我还不清楚它和 a 之间的差异的后果coroutine。（除了我在标题中提出的问题之外，我不知道还能如何问这个问题......）

import asyncio

async def async_generator_spits_out_letters():
    yield 'a'
    yield 'b'
    yield 'c'
    yield 'd'
    await asyncio.sleep(0)

async def coroutine_prints_messages():
    while True:
        print('hi')
        await asyncio.sleep(2)

def test_it():
    print(type(async_generator_spits_out_letters))
    print(type(coroutine_prints_messages))
    # This is how I choose to do newlines....it's easier for me to read. :[
    print(); print()

    print(type(async_generator_spits_out_letters()))
    print(type(coroutine_prints_messages()))

这给出：

<class 'async_generator'>
<class 'coroutine'>


<class 'function'>
<class 'function'>

我无法理解这一点......

我正在尝试将程序从 Tornado 转换为 Asyncio，第一步是使用实际的 asyncio eventloop，如此处所述。

该应用程序在嵌入式 Linux 机器上运行，我通过sysfs/gpio 子系统使用 GPIO ，并且在其中一些 GPIO 上我正在等待中断。我可以通过执行以下操作将其直接集成到 Tornado IOLoop 中：

# Register with the queue
self.io_loop.add_handler(
    self.gpio._value_file, self._handle_interrupt, self.io_loop._EPOLLPRI | self.io_loop.ERROR
)

在代码段中，_value_file是可以读取 GPIO 的文件的文件句柄。只要 GPIO 上有可用中断，就会触发 EPOLLPRI 事件。在龙卷风上，这非常有效。_handle_interrupt它会在中断到来后不久调用该函数。

我的问题是，我无法将其转换为本机异步事件循环。在监视文件描述符的文档中，我只找到了添加读取器和写入器的函数，但没有发现文件描述符上的通用事件掩码。我无法深入研究代码，因为它进入了 C。但是，查看 Tornado 层以将 Tornado IOLoop 的调用转换为 asyncio IOLoop ，情况似乎是这样：

def add_handler(self, fd, handler, events):
    fd, fileobj = self.split_fd(fd)
    if fd in self.handlers:
        raise ValueError("fd %s added twice" % fd)
    self.handlers[fd] = (fileobj, stack_context.wrap(handler)) …

我正在 aiohttp 中编写网络爬虫并遇到 cookie 问题。我尝试抓取的服务器需要身份验证，并且为了获取可供经过身份验证的用户使用的页面，我需要在密钥本身中设置一个带有括号的 cookie。这是一个问题，因为aiohttp.ClientSession.cookie_jar.update_cookies要么忽略任何非法 cookie：

session = ClientSession()
cookie = SimpleCookie("a[b]=1234;")
session.cookie_jar.update_cookies(cookie)
print([f for f in session.cookie_jar])  # empty list, cookie not set

或提出CookieError：

session = ClientSession()
cookie = SimpleCookie()
cookie["a[b]"] = "1234"  # http.cookies.CookieError: Illegal key 'a[b]'
session.cookie_jar.update_cookies(cookie)
print([f for f in session.cookie_jar])


session = ClientSession()
session.cookie_jar.update_cookies([("a[b]", "1234")])  # http.cookies.CookieError: Illegal key 'a[b]'
print([f for f in session.cookie_jar])

可以通过访问http.cookies.Morsel受保护的成员来强制设置cookie _key，即

session = ClientSession()
session.cookie_jar.update_cookies([("__tmp", "1234")])
for cookie in session.cookie_jar:
    if …

我需要使用 AWS IoT MQTT 服务。我目前正在使用https://github.com/aws/aws-iot-device-sdk-python进行一些实验。

我的应用程序将使用 websockets 与另一个服务通信，然后发布/订阅 MQTT 主题来转发/接收消息。

该库是否有可能阻止代码执行？我仍然尝试了解 asyncio，但不确定我应该注意什么。我怎么知道它是否会引起问题？

我相信我只需要使用上面库中的AWSIoTMQTTClient。

这是我的工作代码的摘录：

class AWSIoTClient:

    def __init__():
        ...
        self.client = AWSIoTMQTTClient(...)

    def subscribe(self, callback):
        self.client.subscribe(f'{self.TOPIC}/subscribe/', 0, callback)

    def publish(self, message):
        self.client.publish(self.TOPIC, message, 0)


class MyWSProtocol(WebSocketClientProtocol):

    def set_aws_client(self, client: AWSIoTClient):
        client.subscribe(self.customCallback)
        self.client = client

    def customCallback(self, client, userdata, message):
        # This will be called when we send message from AWS
        if message.payload:
            message = json.loads(message.payload.decode('utf-8').replace("'", '"'))
            message['id'] = self.next_id()
            self.sendMessage(json.dumps(message).encode('utf-8'))

    def onMessage(self, payload, isBinary):
        message = …