如何在Python Asyncio中实现Fire & Forget模式（无需装饰器）

不用装饰器的实现方案 #

方法1：直接用线程执行同步请求 #

因为requests.post是同步阻塞操作，最简单的方式就是开一个后台线程执行请求，主线程直接返回结果，完全不用等待：

import threading
import requests

class DummyClass:
    def do_something():
        "doing something...."

        def _send_request():
            try:
                requests.post("send to url")
            except Exception as e:
                # 这里可以添加日志记录等异常处理逻辑
                print(f"请求发送失败: {str(e)}")

        # 启动守护线程，线程随主线程退出而结束
        threading.Thread(target=_send_request, daemon=True).start()

        return "random string"

这种方式逻辑简单，不需要依赖asyncio，适合没有异步场景的项目。

方法2：基于asyncio实现后台任务 #

如果你的项目已经在用asyncio，注意不要用asyncio.run()（它会阻塞到任务完成），而是把请求任务丢到事件循环后台执行：

import asyncio
import requests
import threading

class DummyClass:
    def do_something():
        "doing something...."

        async def send_request():
            try:
                # 把同步的post请求放到线程池执行（因为requests不支持异步）
                await asyncio.get_event_loop().run_in_executor(None, requests.post, "send to url")
            except Exception as e:
                print(f"请求发送失败: {str(e)}")

        # 处理事件循环的两种情况
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            # 当前没有运行的事件循环，创建新循环并启动守护线程运行
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
            loop.create_task(send_request())
            threading.Thread(target=loop.run_forever, daemon=True).start()
        else:
            # 已有运行的事件循环，直接创建后台任务
            loop.create_task(send_request())

        return "random string"

方法3：改用异步HTTP库（推荐异步场景） #

如果项目本身是异步架构，建议用aiohttp替代requests，实现纯异步的后台请求，不需要依赖线程池：

import asyncio
import aiohttp
import threading

class DummyClass:
    def do_something():
        "doing something...."

        async def send_request():
            try:
                async with aiohttp.ClientSession() as session:
                    await session.post("send to url")
            except Exception as e:
                print(f"请求发送失败: {str(e)}")

        # 处理事件循环
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
            loop.create_task(send_request())
            threading.Thread(target=loop.run_forever, daemon=True).start()
        else:
            loop.create_task(send_request())

        return "random string"

内容的提问来源于stack exchange，提问作者A_Lopez