标签: python-asyncio

我从作者 caleb hattingh 的书中摘取了以下代码片段。我尝试运行代码片段并遇到此错误。（练习）

我该如何解决这个问题？

import asyncio

async def f(delay):
    await asyncio.sleep(1 / delay)
    return delay

loop = asyncio.get_event_loop()
for i in range(10):
    loop.create_task(f(i))

print(loop)
pending = asyncio.all_tasks()
group = asyncio.gather(*pending, return_exceptions=True)
results = loop.run_until_complete(group)
print(f'Results: {results}')
loop.close()

通过以下代码片段，我无法弄清楚为什么infiniteTask没有取消（它不断发送垃圾邮件“我仍然站着”）

在调试模式下，我可以看到Task存储unfinished确实被标记为“已取消”，但显然该线程没有被取消/杀死。

为什么取消包装任务时线程没有被杀死？我应该怎么做才能停止线程？

import time
import asyncio

def quickTask():
    time.sleep(1)

def infiniteTask():
    while True:
        time.sleep(1)
        print("I'm still standing")

async def main():
    finished, unfinished = await asyncio.wait({
            asyncio.create_task(asyncio.to_thread(quickTask)),
            asyncio.create_task(asyncio.to_thread(infiniteTask))
        },
        return_when = "FIRST_COMPLETED"
    )

    for task in unfinished:
        task.cancel()
    await asyncio.wait(unfinished)

    print("  finished : " + str(len(finished))) # print '1'
    print("unfinished : " + str(len(unfinished))) # print '1' 

    
asyncio.run(main())

我本质上是在制作一个 pinger，它有一个密钥/webhook 对的二维列表，并在 ping 一个密钥后，将响应发送到 webhook

二维列表如下：

some_list = [["key1", "webhook1"], ["key2", "webhook2"]]

我的程序本质上是一个循环，我不太确定如何some_list在函数中旋转数据。

这是我的脚本的一个小演示：

some_list = [["key1", "webhook1"], ["key2", "webhook2"]]

我试过了：

async def do_ping(some_pair):
    async with aiohttps.ClientSession() as s:
        tasks = await gen_tasks(s, some_pair)
        results = await asyncio.gather(*tasks*)
        sleep(10)
        await do_ping(some_pair)

但由于该do_ping函数是一个自调用循环，它只是一遍又一遍地调用第一个函数，而永远不会调用后面的函数。希望找到解决方案，无论是线程还是类似的，如果您有更好的构造some_list值的方法（我认为是字典），也请随意放弃该反馈

我正在做一些需要异步方法的繁重处理。我的一个方法返回一个字典列表，在将其添加到另一个可等待对象之前需要对其进行大量处理。IE。

def cpu_bound_task_here(record):
    ```some complicated preprocessing of record```
    return record

在好心人给出以下答案后，我的代码现在被卡住了。

async def fun():
print("Socket open")
record_count = 0
symbol = obj.symbol.replace("-", "").replace("/", "")
loop = asyncio.get_running_loop()
await obj.send()

while True:
    try:
        records = await obj.receive()
        if not records:
            continue

        record_count += len(records)
        

因此，上述函数的作用是异步流式传输值，并在无限期地推送到 Redis 之前进行一些繁重的处理。我做了必要的改变，但现在我陷入了困境。

我希望在后台运行繁重的计算任务而不阻塞 IO。这里的问题是我的主函数不依赖于繁重的任务，并且需要在执行繁重的计算任务之前/同时返回值。举个例子：

def main(args):
   transformed_data_list:List[Dict] = translate_request_to_object(args)
   status = insert_data_into_db(transformed_data:Dict)
   if(status)
      # running background task
      asyncio.run(process_background_task(transformed_data_list))

   # Here, I want to return a success response as soon as data inserted into db
   return "data insert into db"

   async process_background_task(transformed_data_list:List[Dict]):
      for data in transformed_data_list:List:
         asyncio.create_task(heavy_computation_task(data))

但上面的代码在完成之前不会返回响应process_background_task。