标签: python-asyncio

我有一个 bog 标准同步 python 程序，需要能够从 websockets 读取数据并使用数据更新 GUI。然而，asyncio 的蠕动总是让我绊倒。

我如何制作一个模块：

1. 接受多个来源的多个订阅
2. 每当有数据时就向请求者发送更新
3. 每个 URL 只打开一个 websocket 连接
4. 如果 websocket 关闭则重置它

这是我已经拥有的，但它在很多方面都失败了：

1. run_forever()意味着循环在订阅完成之前被卡住，然后handle()卡在 falseywhile循环中
2. 它似乎不想在套接字关闭时重新启动套接字，因为 websockets 对象没有属性connected（没有 s 的 websocket 有，但我不清楚其中的差异，也无法在网上找到信息）
3. 我绝对不确定我的方法是否正确。

已经为此奋斗了几个星期。希望得到一些指点。

class WSClient():
    subscriptions = set()
    connections = {}
    started = False

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

    def start(self):
        self.started = True
        self.loop.run_until_complete(self.handle())
        self.loop.run_until_forever()  # problematic, because it does not allow new subscribe() events

    async def handle(self):
        while len(self.connections) > 0:
            # …

我想在 Python 代码中重现 javascript 的 Promise.race 行为。我想同时运行一组协程，并在第一个协程完成时返回，获取其结果并取消/丢弃仍在运行的协程的结果。

我正在尝试学习 asyncio。如果我在没有 asyncio 库的情况下正常运行这个程序，那么它需要更少的时间，而以这种方式需要更多的时间，那么这是使用 asyncio 发送邮件的正确方法还是还有其他方法？

import smtplib 
import ssl
import time
import asyncio


async def send_mail(receiver_email):
    try:
        print(f"trying..{receiver_email}")
        server = smtplib.SMTP(smtp_server, port)
        server.ehlo()
        server.starttls(context=context)
        server.ehlo()
        server.login(sender_email, password)
        message = "test"
        await asyncio.sleep(0)
        server.sendmail(sender_email, receiver_email, message)
        print(f"done...{receiver_email}")
    except Exception as e:
        print(e)
    finally:
        server.quit()

async def main():
     t1 = time.time()
     await asyncio.gather(
         send_mail("test@test.com"),
         send_mail("test@test.com"),
         send_mail("test@test.com"),
         send_mail("test@test.com")
     )
    print(f"End in {time.time() - t1}sec")

if __name__ == "__main__":
     smtp_server = "smtp.gmail.com"
     port = 587  # For starttls
     sender_email = "*****"
     password = …

我有个问题。因此，每次用户在我的不和谐服务器上写入聊天消息时，我都会运行一个任务 - 它称为on_message. 所以我的机器人在这个事件中有很多事情要做，我经常遇到这种错误：

Task was destroyed but it is pending!\ntask: <Task pending name='pycord: on_message' coro=<Client._run_event() done, defined at /Bots/gift-bot/discord/client.py:374> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f68a7bdfc10>()]>>\n

所以我想如果我想解决这个问题，我需要加速我的代码。但遗憾的是，我不知道如何修复此错误。

编辑：我集成了计时，这就是我打印的内容：

Task was destroyed but it is pending!\ntask: <Task pending name='pycord: on_message' coro=<Client._run_event() done, defined at /Bots/gift-bot/discord/client.py:374> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f01063f98e0>()]>>\n2 if checks done - 7.867813110351562e-06\n5 if checks done - 0.0061550140380859375\nmysql checks done - 0.010785341262817383\ntask done - 0.13075661659240723\n2 if checks done - 8.344650268554688e-06\n5 if …

我正在尝试使用异步客户端库（该库也是由 Google 提供）调用外部 API（由 Google 提供的 API）。

我尝试调用的异步方法是async list_featurestores()（文档）。它提供了以下示例代码：

from google.cloud import aiplatform_v1

async def sample_list_featurestores():
    # Create a client
    client = aiplatform_v1.FeaturestoreServiceAsyncClient()

    # Initialize request argument(s)
    request = aiplatform_v1.ListFeaturestoresRequest(
        parent="parent_value",
    )

    # Make the request
    page_result = client.list_featurestores(request=request)

    # Handle the response
    async for response in page_result:
        print(response)

（我已经从上面的链接页面复制/粘贴了该代码）。

为了运行该代码，我对其进行了稍微调整：

import asyncio
from google.cloud import aiplatform_v1

async def sample_list_featurestores():
    client = aiplatform_v1.FeaturestoreServiceAsyncClient()
    request = aiplatform_v1.ListFeaturestoresRequest(parent="projects/MY_GCP_PROJECT/locations/europe-west2",)
    page_result  = client.list_featurestores(request=request)
    async for response in page_result:
        print(response)


if …

我最近发现并重现了由于使用asyncio.wait导致的内存泄漏。具体来说，我的程序定期执行某些函数，直到stop_event设置为止。我将程序简化为下面的代码片段（减少超时以更好地演示问题）：

async def main():
  stop_event = asyncio.Event()

  while True:
    # Do stuff here
    await asyncio.wait([stop_event.wait()], timeout=0.0001)

asyncio.run(main())

虽然这对我来说看起来无害，但事实证明这里存在内存泄漏。如果执行上面的代码，您将看到内存使用量在几分钟内增长到数百 MB。这让我很惊讶，并花了很长时间才找到。我期望在超时后，我正在等待的任何内容都会被清理（因为我自己没有保留任何对它的引用）。然而，事实证明并非如此。

使用gc.get_referrers，我能够推断每次调用时asyncio.wait(...)，都会创建一个新任务，该任务保存对返回的对象的引用stop_event.wait()，并且该任务将永远保留。具体来说，len(asyncio.all_tasks())随着时间的推移不断增加。即使超时了，任务仍然存在。只有在调用时，stop_event.set()这些任务才会立即完成，并且内存使用量才会急剧减少。

发现这一点后，文档中的这条注释让我尝试使用asyncio.wait_for代替：

与 wait_for() 不同，wait() 在发生超时时不会取消 future。

事实证明，它的表现确实如我所料。超时后不会保留任何引用，内存使用量和任务数量保持不变。这是没有内存泄漏的代码：

async def main():
  stop_event = asyncio.Event()

  while True:
    # Do stuff here
    try:
      await asyncio.wait_for(event.stop_event(), timeout=0.0001)
    except asyncio.TimeoutError:
      pass

asyncio.run(main())

虽然我很高兴这个问题现在已经解决了，但我不太理解这种行为。如果超过了超时时间，为什么要让这个任务保留一个引用呢？这似乎是造成内存泄漏的一个秘诀。关于不取消期货的说明我也不清楚。如果我们不明确取消 future，但我们只是不让任务在超时后保留引用，该怎么办？那不是也可以吗？

如果有人能对此有所启发，我们将不胜感激。多谢！

目标是在后台提取实时数据（例如每 5 秒）并在需要时提取到仪表板。这是我的代码。它有点工作，但我看到两个问题： 1.如果我将 st.write("TESTING!") 移到最后，由于 while 循环，它永远不会被执行。有办法改进吗？我可以想象，随着仪表板的增长，将会有多个页面/表格等。这不会提供太多的灵活性。2. async 函数中的 return px 行，我对此不太满意，因为我通过反复试验得到了正确的结果。抱歉，我是个新手，但如果有更好的方法，我将非常感激。

谢谢你！

import asyncio
import streamlit as st
import numpy as np

st.set_page_config(layout="wide")

async def data_generator(test):
    while True:
        with test:
            px = np.random.randn(5, 1)
        await asyncio.sleep(1)
        return px

test = st.empty()
st.write("TESTING!")

with test:
    while True:
        px = asyncio.run(data_generator(test))
        st.write(px[0])

我刚刚发现了 Python 3.11 的新功能，如 ExceptionGroup 和 TaskGroup，我对以下 TaskGroup 行为感到困惑：如果组内的一个或多个任务失败，则所有其他正常任务都将被取消，我没有机会更改该行为 示例：

async def f_error():
    raise ValueError()

async def f_normal(arg):
    print('starting', arg)
    await asyncio.sleep(1)
    print('ending', arg)


async with asyncio.TaskGroup() as tg:
    tg.create_task(f_normal(1))
    tg.create_task(f_normal(2))
    tg.create_task(f_error())

# starting 1
# starting 2
#----------
#< traceback of the error here >

在上面的例子中，我无法打印“ending 1”和“ending 2”。asyncio.gather(return_exceptions=True)同时，在发生错误时提供类似选项不取消剩余任务的选项将非常有用。

你可以说“如果你不想要这种取消行为，就不要使用TaskGroup”，但答案是我想使用新的异常组功能，并且它严格绑定到TaskGroup

所以问题是：

1. 我可以以某种方式在 asyncio 中使用异常组，而无需在 TaskGroup 中使用这种全有或全无的取消策略吗？
2. 如果对于前面的答案是“否”：为什么 python 开发人员消除了在 TaskGroup API 中禁用取消的可能性？

我正在使用 Python 3.10，我对asyncio.create_task.

在下面的示例代码中，无论我是否使用asyncio.create_task. 似乎没有什么区别。

如何确定何时使用以及使用与不使用相比asyncio.create_task有何优势？asyncio.create_task

import asyncio
from asyncio import sleep

async def process(index: int):
    await sleep(1)
    print('ok:', index)

async def main1():
    tasks = []
    for item in range(10):
        tasks.append(asyncio.create_task(process(item)))
    await asyncio.gather(*tasks)

async def main2():
    tasks = []
    for item in range(10):
        tasks.append(process(item)) # Without asyncio.create_task
    await asyncio.gather(*tasks)

asyncio.run(main1())
asyncio.run(main2())

我正在学习用 scrapy 进行 python 抓取。我做了和教程教的完全一样的事情。但我得到了一个错误。请帮忙！

我的Python代码：

import scrapy


class BookSpider(scrapy.Spider):
    name = "books"
    allowed_domains = ["books.toscrape.com"]
    start_urls = ["https://books.toscrape.com"]

    def parse(self, response):
        books = response.css("article.product_pod")
                             
        for book in books:
            yield{
                "name":book.css("h3 a::text").get(),
                "price":book.css(".product_price .price_color::text").get(),
                "url": book.css("h3 a").attrib["href"],
            }

终端显示

import scrapy


class BookSpider(scrapy.Spider):
    name = "books"
    allowed_domains = ["books.toscrape.com"]
    start_urls = ["https://books.toscrape.com"]

    def parse(self, response):
        books = response.css("article.product_pod")
                             
        for book in books:
            yield{
                "name":book.css("h3 a::text").get(),
                "price":book.css(".product_price .price_color::text").get(),
                "url": book.css("h3 a").attrib["href"],
            }

ossignal.py 文件：

import signal

signal_names = {}
for signame …