如何在Python 3中处理回调及自定义HTTP LISTEN的非阻塞调用？

哈喽！我来逐个给你解答这两个问题：

回调说白了就是把一个函数当成参数传给另一个函数，让它在某个特定时机自动触发执行。Python对这种模式支持得特别顺手，下面给你举几个常见场景的例子：

基础同步回调 #

最直接的用法就是简单的函数传递，比如你写一个处理数据的函数，把回调函数传进去，等数据处理完就触发它：

def process_data(data, on_complete):
    print(f"正在处理数据: {data}")
    # 模拟耗时操作
    import time
    time.sleep(2)
    # 处理完成后触发回调
    on_complete(f"处理结果: {data.upper()}")

def handle_result(result):
    print(f"收到回调结果: {result}")

# 调用时传入回调函数
process_data("hello world", handle_result)

异步场景下的回调 #

如果是异步IO场景（比如用asyncio），回调也可以配合协程使用，或者用add_done_callback方法处理异步任务的结果：

import asyncio

async def async_process_data(data):
    print(f"异步处理数据: {data}")
    await asyncio.sleep(2)
    return f"异步处理结果: {data.upper()}"

def async_callback(future):
    print(f"收到异步回调结果: {future.result()}")

async def main():
    task = asyncio.create_task(async_process_data("hello async"))
    # 给任务添加回调
    task.add_done_callback(async_callback)
    # 继续执行其他逻辑
    print("主线程可以做其他事情...")
    await task

asyncio.run(main())

类中的回调 #

如果是面向对象的代码，也可以把类的方法作为回调传递：

class DataProcessor:
    def process(self, data, callback):
        print(f"类方法处理数据: {data}")
        callback(f"类处理结果: {data * 2}")

class ResultHandler:
    def handle(self, result):
        print(f"类方法接收回调: {result}")

processor = DataProcessor()
handler = ResultHandler()
processor.process(123, handler.handle)

你的场景是一个阻塞的LISTEN请求，服务器会持续推送更新，这时候要让主代码不被卡住，有两种主流方案：多线程或者异步HTTP客户端，下面分别给出具体实现：

方案一：用多线程把监听逻辑放到子线程 #

这种方法最简单，把阻塞的LISTEN请求放到单独的线程里，主线程该干嘛干嘛：

import requests
import threading

def listen_for_updates(url):
    try:
        # 发送LISTEN请求，stream=True表示开启流式响应（关键！）
        with requests.request("LISTEN", url, stream=True) as response:
            response.raise_for_status()
            # 逐行读取服务器推送的内容（因为每次更新是一段独立的JSON）
            for line in response.iter_lines():
                if line:
                    # 解析JSON
                    import json
                    update = json.loads(line)
                    print(f"收到资源更新: {update}")
                    # 这里可以加你自己的处理逻辑，比如触发其他回调
    except Exception as e:
        print(f"监听出错: {e}")
        # 可选：添加重连逻辑，比如休眠几秒后重新发起请求

def main():
    listen_url = "http://<IP-Address>:<Port>/resource"
    # 创建并启动子线程，daemon=True让子线程随主线程退出而结束
    listen_thread = threading.Thread(target=listen_for_updates, args=(listen_url,), daemon=True)
    listen_thread.start()

    # 主线程继续执行其他逻辑
    print("主线程正在执行其他任务...")
    while True:
        # 模拟主任务持续运行
        import time
        time.sleep(3)
        print("主线程还在工作...")

if __name__ == "__main__":
    main()

方案二：用异步HTTP客户端（aiohttp）实现非阻塞监听 #

如果你的项目本身是异步架构，用aiohttp会更优雅，不需要额外线程，完全基于异步IO：

import aiohttp
import asyncio
import json

async def listen_for_updates(url):
    try:
        async with aiohttp.ClientSession() as session:
            # 发送LISTEN请求，timeout=None表示不设置超时（因为是长期监听）
            async with session.request("LISTEN", url, timeout=None) as response:
                response.raise_for_status()
                # 逐块读取流式响应
                async for line in response.content:
                    if line:
                        update = json.loads(line.decode('utf-8'))
                        print(f"收到资源更新: {update}")
    except Exception as e:
        print(f"监听出错: {e}")
        # 可选：添加重连逻辑

async def main():
    listen_url = "http://<IP-Address>:<Port>/resource"
    # 创建监听任务
    listen_task = asyncio.create_task(listen_for_updates(listen_url))

    # 异步主协程执行其他逻辑
    print("异步主协程正在执行其他任务...")
    while True:
        await asyncio.sleep(3)
        print("异步主协程还在工作...")

    # 如果需要停止监听，可以调用 listen_task.cancel()

asyncio.run(main())

注意事项 #

• 不管用哪种方案，都要做好异常处理（比如服务器断开、网络波动），可以在监听函数里加自动重连的逻辑
• 服务器返回的每一行是独立的JSON，一定要逐行解析，别把多行混在一起，不然会解析出错

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