Python3.7中的Fire、Forget和返回值

Posted 2023-03-30

【中文标题】Python3.7中的Fire、Forget和返回值

【英文标题】：Fire, Forget, and Return Value in Python3.7

【发布时间】：2019-08-08 07:15:14

【问题描述】：

我有以下情况： 我有一个python服务器，在收到请求后，需要解析一些信息，尽快将结果返回给用户，然后自行清理。 我尝试使用以下逻辑来设计它：

Consumer: *==*   (wait for result)   *====(continue running)=====...
              \                     / return
Producer:      *======(prase)====*=*
                                  \
Cleanup:                           *==========*

我一直在尝试使用异步任务和协程来使这种情况有效，但无济于事。我尝试的一切最终要么是生产者在返回之前等待清理完成，要么是返回杀死清理。 理论上我可以让消费者在向用户显示结果后调用清理，但我拒绝相信 Python 不知道如何“即发即弃”并返回。

例如这段代码：

import asyncio

async def Slowpoke():
    print("I see you shiver with antici...")
    await asyncio.sleep(3)
    print("...pation!")

async def main():
    task = asyncio.create_task(Slowpoke())
    return "Hi!"

if __name__ == "__main__":
    print(asyncio.run(main()))
    while True:
        pass

返回：

I see you shiver with antici...
Hi!

永远不会到达...pation。

我错过了什么？

【问题讨论】：

当您可能想要 run_forever()

时，您正在运行主循环以完成

【参考方案1】：

我设法使用线程而不是 asyncio 让它工作：

import threading
import time

def Slowpoke():
    print("I see you shiver with antici...")
    time.sleep(3)
    print("...pation")

def Rocky():
    t = threading.Thread(name="thread", target=Slowpoke)
    t.setDaemon(True)
    t.start()
    time.sleep(1)
    return "HI!"

if __name__ == "__main__":
    print(Rocky())
    while True:
        time.sleep(1)

【讨论】：

哎呀，我来晚了哈哈

我添加了一个简单的解释，但您可能想研究一下线程以了解为什么您的任务之前会死掉

【参考方案2】：

asyncio 似乎并不特别适合这个问题。你可能想要简单的线程：

这样做的原因是您的任务在父级完成时被杀死。通过抛出一个daemon 线程，您的任务将继续运行，直到它完成，或者直到程序退出。

import threading
import time

def Slowpoke():
    try:
        print("I see you shiver with antici...")
        time.sleep(3)
        print("...pation!")
    except:
        print("Yup")
        raise Exception()

def main():
    task = threading.Thread(target=Slowpoke)
    task.daemon = True
    task.start()
    return "Hi!"

if __name__ == "__main__":
    print(main())
    while True:
        pass

【参考方案3】：

asyncio.run ...

[...] 创建一个新的事件循环并在最后关闭它。 [...]

包裹在task 中的coro 在main 执行期间没有机会完成。 如果您返回 Task 对象并打印它，您会看到它处于取消状态：

async def main():
    task = asyncio.create_task(Slowpoke())
    # return "Hi!"
    return task

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

# I see you shiver with antici...
# <Task cancelled coro=<Slowpoke() done, defined at [...]>>

当main 在创建和调度任务（并打印“Hi！”）后结束时，事件循环关闭，这将导致其中所有正在运行的任务被取消。

您需要保持事件循环运行，直到任务完成，例如awaitmain：

async def main():
    task = asyncio.create_task(Slowpoke())
    await task
    return task

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

# I see you shiver with antici...
# ...pation!
# <Task finished coro=<Slowpoke() done, defined at [..]> result=None>

【讨论】：

OP想要不等待就返回任务

是的，重点是返回值。不过，我确实找到了解决方案 - 使用 therading。

【参考方案4】：

（我希望我确实正确理解了您的问题。ASCII 图像和文本描述在我的脑海中并不完全对应。"Hi!" 是结果，"Antici..pation" 是清理，对吧？我也喜欢那个音乐剧, 顺便说一句）

一种可能的基于异步的解决方案是尽快返回结果。 return 终止了任务，这就是为什么有必要对清理进行一劳永逸。它必须伴随着等待所有清理完成的关闭代码。

import asyncio

async def Slowpoke():
    print("I see you shiver with antici...")
    await asyncio.sleep(3)
    print("...pation!")

async def main():
    result = "Hi!"
    asyncio.create_task(Slowpoke())
    return result

async def start_stop():
    # you can create multiple tasks to serve multiple requests
    task = asyncio.create_task(main())
    print(await task)

    # after the last request wait for cleanups to finish
    this_task = asyncio.current_task()
    all_tasks = [ 
        task for task in asyncio.all_tasks()
        if task is not this_task]
    await asyncio.wait(all_tasks)

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

---

另一种解决方案是使用其他方法（不返回）将结果传递给等待任务，以便在解析后立即开始清理。 Future 被认为是低级的，但无论如何这里是一个示例。

import asyncio

async def main(fut):
    fut.set_result("Hi!")
    # result delivered, continue with cleanup
    print("I see you shiver with antici...")
    await asyncio.sleep(3)
    print("...pation!")

async def start_stop():
    fut = asyncio.get_event_loop().create_future()
    task = asyncio.create_task(main(fut))
    print(await fut)

    this_task = asyncio.current_task()
    all_tasks = [ 
        task for task in asyncio.all_tasks()
        if task is not this_task]
    await asyncio.wait(all_tasks)

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