标签: python-asyncio

我认为我的问题很简单也很愚蠢，但是我读了很多材料，无法想象如何做我想要的事情。

所以，我使用websockets库，并且我有这个算法：

# 1. get connection and start handle it
async def main_request_handler(ws, path):
    proxy = Proxy()
    try:
        await proxy.start(ws, path)

2.在start内部我创建第二个websocket来传递请求ws并接收答案并将其发送到ws

while True:
    request_raw = await self.ws_server.recv()
    await self.process_request_from_server(request_raw)

问题是，我需要为多个 ws客户端使用一个 websocket 服务器连接，并且我需要向每个人传递相同的答案ws_server。现在我只得到一个响应，因为 .recv() 仅返回其中一个“订阅者”的值。如何解决这个问题？请注意，我使用while True和async

我使用 asyncio 没有获得任何加速。此代码片段仍然以与同步作业相同的方式运行。大多数示例使用 asyncio.sleep() 来施加延迟，我的问题是如果部分代码根据输入参数造成延迟怎么办？

async def c(n):
    #this loop is supposed to impose delay
    for i in range(1, n * 40000):
        c *= i
   return n

async def f():
    tasks = [c(i) for i in [2,1,3]]
    r=[]
    completed, pending = await asyncio.wait(tasks)
    for item in completed:
        r.append(item.result())

    return r

if __name__=="__main__":
    loop = asyncio.get_event_loop()
    k=loop.run_until_complete(f())
    loop.close()

我期望得到 [1,2,3] 但我没有（串行运行时也没有时间差）

如何使用 asyncio 并永远运行该函数。我知道有run_until_complete(function_name)，但是我该如何使用run_forever如何调用异步函数？

async def someFunction():
    async with something as some_variable:
        # do something

我不知道如何启动该功能。

我尝试每 X 秒访问一个具有并行和单独会话的网站，然后分析响应中的内容以查看每个会话是否应该继续。但是，一旦代码到达第二个循环，它就会失败。

import asyncio
from aiohttp import ClientSession
import logging
import time

interval = 30
instances = 2
visit_url = 'http://www.example.org'

tasks = []

logging.basicConfig(
    format='%(asctime)s.%(msecs)03d %(message)s',  # Log in format time.milliseconds {message}
    level=logging.INFO,  # Use with logging.info()
    datefmt='%H:%M:%S')  # Display time as Hours:Minutes:Seconds


class StopException(Exception):
    pass


async def quit_app(session, task_, reason):
    logging.info("[{}] {}.".format(task_, reason))
    session.cookies.clear()  # Reset cookies
    session.headers.clear()  # Reset headers
    session.close()  # End HTTP connection
    raise StopException


async def get_status(response):
    if "abow" in response:
        return "success" …

我有以下场景：

• Python 3.6+
• 输入数据是从文件中逐行读取的。
• 协程将数据发送到 API（使用aiohttp）并将调用结果保存到 Mongo（使用motor）。所以有很多 IO 发生。

该代码是使用async/编写的await，对于手动执行的单个调用来说效果很好。

我不知道该怎么做是批量使用输入数据。

asyncio我见过的所有示例都是asyncio.wait通过发送有限列表作为参数来演示的。但我不能简单地向其发送任务列表，因为输入文件可能有数百万行。

我的场景是通过传送带将数据流向消费者。

我还可以做些什么？我希望程序使用它可以聚集的所有资源来处理文件中的数据，但又不会被淹没。

我以前使用的是 requests，但后来我转向 aiohttp + asyncio 来并行运行帐户，但是我在将逻辑放在脑海中时遇到了困难。

class Faked(object):
    def __init__(self):
        self.database = sqlite3.connect('credentials.db')

    async def query_login(self, email):
        print(email)
        cur = self.database.cursor()
        sql_q = """SELECT * from user WHERE email='{0}'""".format(email)
        users = cur.execute(sql_q)
        row = users.fetchone()
        if row is None:
            raise errors.ToineyError('No user was found with email: ' + email + ' in database!')

        self.logger().debug("Logging into account '{0}'!".format(row[0]))
        call_func = await self._api.login(data={'email': row[0],
                                                'password': row[1],
                                                'deviceId': row[2],
                                                'aaid': row[3]})
        return await call_func

    async def send_friend_request(self, uid):
        return await self._api.send_friend_request(uid)


def main(funcs, …

我正在使用基于 aiohttp 的第三方 API 客户端库。该库没有将其客户端实现为上下文管理器，但只要我始终aiohttp.ClientSession手动关闭（对吗？），这应该没有问题。

问题是，即使在仔细执行此操作之后，我仍然收到警告，Unclosed client session但没有任何信息来自它的来源。我正在使用 捕获警告回溯pytest -W error::ResourceWarning，但结果很简单

.Exception ignored in: <bound method ClientSession.__del__ of <aiohttp.client.ClientSession object at 0x10fc15ba8>>
Traceback (most recent call last):
  File "<PROJECT>/venv/lib/python3.6/site-packages/aiohttp/client.py", line 211, in __del__
    **kwargs)
ResourceWarning: Unclosed client session <aiohttp.client.ClientSession object at 0x10fc15ba8>

即单帧的回溯。

有人可以告诉我我在这里做错了什么以及如何获得有关该警告的更多信息吗？

如果您检查Quart库，app.run()只需建立一些配置，然后使用asyncio.run(serve(self, config))，其中 serve 来自from hypercorn.asyncio import serve。

因此，即使您通过运行Quart应用程序python myapp.py，它不是已经在使用Hypercorn服务器了吗？

特别是，this 和 running via 有hypercorn myapp:app什么区别？

https://pgjones.gitlab.io/quart/deployment.html

不建议在生产中直接运行 Quart（通过 run()）。相反，建议使用 Hypercorn 或替代 ASGI 服务器运行 Quart。Hypercorn 与 Quart 一起安装，默认情况下用于处理请求（例如使用 run()）。

所以听起来，即使Hypercorn默认使用 来处理请求run()，也不建议使用run()? 还有人糊涂吗？

我在看import asyncio: Learn Python's AsyncIO #3 - Using Coroutines。老师举了以下例子：

import asyncio
import datetime

async def keep_printing(name):
    while True:
        print(name, end=" ")
        print(datetime.datetime.now())
        await asyncio.sleep(0.5)

async def main():
    group_task = asyncio.gather(
                     keep_printing("First"),
                     keep_printing("Second"),
                     keep_printing("Third")
                 )
    try:
        await asyncio.wait_for(group_task, 3)
    except asyncio.TimeoutError:
        print("Time's up!")


if __name__ == "__main__":
    asyncio.run(main())

输出有一个异常：

First 2020-08-11 14:53:12.079830
Second 2020-08-11 14:53:12.079830
Third 2020-08-11 14:53:12.080828 
First 2020-08-11 14:53:12.580865
Second 2020-08-11 14:53:12.580865
Third 2020-08-11 14:53:12.581901 
First 2020-08-11 14:53:13.081979
Second 2020-08-11 14:53:13.082408
Third 2020-08-11 14:53:13.082408 
First 2020-08-11 14:53:13.583497 …

我正在使用aiohttp、asyncio和codetiming来发出并发请求。我最近将 Python 升级到 3.9.0 版，但出现 RuntimeError: 程序运行后事件循环已关闭。我正在尝试使用队列数据结构发出异步请求...

import asyncio
import aiohttp
from codetiming import Timer
async def task(name, work_queue):
    timer = Timer(text=f"Task {name} elapsed time: {{:.1f}}")
    async with aiohttp.ClientSession() as session:
        while not work_queue.empty():
            url = await work_queue.get()
            print(f"Task {name} getting URL: {url}")
            timer.start()
            async with session.get(url) as response:
                await response.text()
            timer.stop()
async def main():
    # Create the queue of work
    work_queue = asyncio.Queue()
    # Put some work in the queue
    for url in …