标签: python-asyncio

我有 csv 文件，其中一列用作在两个 POST 方法调用中传递的数据。我有三个不同的 URL，需要随后在代码中调用。我使用 aiohttp 定义会话并循环列长度（其单元格值在两个 POST 调用中发送）。从响应中获取的数据存储在一个列表中，并用于具有不同 url 的第三个 POST 调用。

前两个调用似乎执行得很好而且相当快（我打印了它们的响应并进行了验证），但是当执行到第三个调用时，出现此错误：

OSError：[Errno 24]打开的文件太多

我尝试了一些建议connector=aiohttp.TCPConnector(verify_ssl=False)在aiohttp.ClientSession(connector=aiohttp.TCPConnector(verify_ssl=False)) 但这不起作用的解决方案中指定。解决这个问题的最佳方法应该是什么？我当前使用的第二个 URL 是 HTTP URL，即 localhost url，最终将更改为基于 https 云的 url。

下面是描述这种情况的示例代码：

import aiohttp
import pandas as pd
import asyncio
import requests
import json
import time

start_time = time.time()

df = pd.read_csv('Sample.csv', header=None, dtype=str)
RequestBodyvalues = df.iloc[:, [0]].values

async def get_FirstUrlresponse(session, url, requestBody, headers):
    async with session.post(url, data = json.dumps(requestBody), headers = headers) as resp:
        response = await resp.json()
        return …

我试图解决的问题： 我正在向服务器发出许多 api 请求。我试图在异步 api 调用之间创建延迟，以遵守服务器的速率限制策略。

我想要它做什么 我希望它的行为如下：

1. 发出 api 请求 #1
2. 等待0.1秒
3. 发出 api 请求 #2
4. 等待0.1秒...等等...
5. 重复直到所有请求都提出
6. 收集响应并在一个对象中返回结果（结果）

问题： 当我在代码中引入asyncio.sleep()或time.sleep()时，它仍然几乎立即发出 api 请求。它似乎延迟了print()的执行，但没有延迟 api 请求。我怀疑我必须在循环中创建延迟，而不是在 fetch_one() 或 fetch_all() 处，但不知道如何做到这一点。

代码块：

async def fetch_all(loop, urls, delay): 
    results = await asyncio.gather(*[fetch_one(loop, url, delay) for url in urls], return_exceptions=True)
    return results

async def fetch_one(loop, url, delay):

    #time.sleep(delay)
    #asyncio.sleep(delay)

    async with aiohttp.ClientSession(loop=loop) as session:
        async with session.get(url, ssl=SSLContext()) as resp:
            # print("An api call to ", …

我有一个问题，我是 python 异步世界的新手，我编写了一些代码来测试 的功能asyncio，我创建了 10 个包含随机内容的文件，名为file1.txt, file2.txt, ..., file10.txt

这是我的代码：

import asyncio
import aiofiles
import time

async def reader(pack, address):
    async with aiofiles.open(address) as file:
        pack.append(await file.read())

async def main():
    content = []
    await asyncio.gather(*(reader(content, f'./file{_+1}.txt') for _ in range(10)))

    return content

def core():
    content = []
    for number in range(10):
        with open(f'./file{number+1}.txt') as file:
            content.append(file.read())
    
    return content

if __name__ == '__main__':
    # Asynchronous
    s = time.perf_counter()
    content = asyncio.run(main())
    e = time.perf_counter()
    print(f'Take {e - s: …

我知道星号用于解包系统参数等值或将列表解包到变量中。

但我之前在这个 asyncio 示例中没有见过这种语法。

我在这里阅读这篇文章，https://realpython.com/async-io-python/#the-10000-foot-view-of-async-io，但我不明白星号运算符在这做什么语境。

#!/usr/bin/env python3
# rand.py

import asyncio
import random

# ANSI colors
c = (
    "\033[0m",   # End of color
    "\033[36m",  # Cyan
    "\033[91m",  # Red
    "\033[35m",  # Magenta
)

async def makerandom(idx: int, threshold: int = 6) -> int:
    print(c[idx + 1] + f"Initiated makerandom({idx}).")
    i = random.randint(0, 10)
    while i <= threshold:
        print(c[idx + 1] + f"makerandom({idx}) == {i} too low; retrying.")
        await asyncio.sleep(idx + 1)
        i = random.randint(0, 10)
    print(c[idx + …

我是一名 Python 初学者，从https://www.youtube.com/watch?v=iG6fr81xHKA&t=269s获取有关 asyncio 强大功能的信息，我尝试使用所示的示例并将其重新调整用途以执行 10 次。这是一个代码片段

def main(x):
    print("Hello")
    time.sleep(3)
    print("World!")

因此，我尝试以异步方式执行此操作，但它不会异步执行。到目前为止，这是我尝试过的。我究竟做错了什么？

import time
import asyncio


async def main(x):
    print(f"Starting Task {x}")
    await asyncio.sleep(3)
    print(f"Finished Task {x}")


async def async_io():
    for i in range(10):
        await main(i)

if __name__ == "__main__":
    start_time = time.perf_counter()
    asyncio.run(async_io())
    print(f"Took {time.perf_counter() - start_time} secs")

我还尝试在 asyncio 中使用queue_task。

我有一个应用程序，它已经无限运行，异步事件循环永远运行，而且我需要每 10 秒运行一个特定的函数。

def do_something():
   pass

a = asyncio.get_event_loop()
a.run_forever()

我想每 10 秒调用一次函数 do_something 。如何在不使用 while 循环替换 asynctio 事件循环的情况下实现这一目标？

编辑：我可以用下面的代码来实现这一点

def do_something():
   pass
while True:
   time.sleep(10)
   do_something()

但我不想使用 while 循环在我的应用程序中无限运行，而是想使用 asyncio run_forever()。那么如何使用 asyncio 每 10 秒调用相同的函数呢？有没有类似的调度程序不会阻止我正在进行的工作？

我很难理解它是如何AsyncIOScheduler工作的以及如何在主函数内安排一个函数，或者如何是正确的方法来做到这一点。

如何foo每 10 秒运行一次该函数？

假设我有这样的结构：

package/
    controllers.py
    main.py

从文件中controllers.py我调用了一个函数foo，该函数如下所示：

async def foo():
    print('Hello World')

我想foo从文件运行该函数（以及许多其他函数）main.py，该文件如下：

import asyncio
from controllers import foo, bar
from apscheduler.schedulers.asyncio import AsyncIOScheduler

async def main():

    # Init message
    print('\nPress Ctrl-C to quit at anytime!\n' )
    
    await asyncio.create_task(bar())

    scheduler = AsyncIOScheduler()
    scheduler.add_job(await asyncio.create_task(foo()), "interval", seconds=10)
    scheduler.start()

if __name__ == "__main__":
    while True:
        try:
            asyncio.run(main())
        except Exception as e:
            print("Exception: " + str(e))

这样运行调度程序是否正确？或者每次main函数运行时定时器都会被重置？

我尝试了下面的代码，循环间隔有效，但该main …

假设我有这个：

class Test():
    def __init__(self, number):
        self.number = number
        await self.TestPrint()

    async def TestPrint(self):
        print(self.number)

正如您所看到的，这不起作用，因为__init__不是async，而且我无法调用await该函数。

我希望能够在假设我想维护这个函数 async 的情况TestPrint下运行。__init__