Hypercorn是否支持类似Gunicorn的服务器钩子？求实现方案

我了解到Gunicorn提供可监听各类服务器事件的server hooks，Hypercorn受Gunicorn启发，我希望在Hypercorn中实现相同功能，但查阅Hypercorn文档未获有效信息，也未找到相关实现案例。

我的Hypercorn配置与代码 #

Hypercorn配置文件（hypercorn.config.py） #

import multiprocessing as mp
import os
print( (mp.cpu_count() * 2) + 1)

accesslog = "-"
backlog = 500
bind = "0.0.0.0:8000"
statsd_host = os.environ.get("STASD_HOST")
workers = 3
max_requests = 2

应用入口代码 #

"""
This module is the entry point for the API.
This will contain all of the middleware and routes and database initialization
"""

from starlette.applications import Starlette
from starlette.routing import Route

from src.health_check import HealthCheckAPI

routes = [
    Route(f"/health-check", HealthCheckAPI)
]
app = Starlette(debug=True, routes=routes)
# app = AsyncioWSGIMiddleware(app)

if __name__ == "__main__":
    from hypercorn.config import Config

    config = Config()
    config.bind = ["localhost:8077"]
    config.debug = True
    import asyncio

    from hypercorn.asyncio import serve

    asyncio.run(serve(app, config))

启动命令 #

hypercorn -c file:config/hypercorn.config.py  src.application:app

Gunicorn钩子示例配置 #

import multiprocessing as mp
import os
print( (mp.cpu_count() * 2) + 1)

accesslog = "-"
backlog = 500
bind = "0.0.0.0:8000"
statsd_host = os.environ.get("STASD_HOST")
workers = 3
max_requests = 2

def on_starting(server):
    # register some variables here
    print(server)
    print("Starting Flask application")

def on_exit(server):
    # perform some clean up tasks here using variables from the application
    print("Shutting down Flask application")

def worker_exit(server, worker):
    print("Worker exiting")
    print(worker)

def nworkers_changed(server, new_value, old_value):
    print("n worker",new_value)

Hypercorn中实现类似钩子的方法 #

Hypercorn虽无与Gunicorn完全一一对应的钩子，但提供异步事件回调实现类似功能，可直接在配置文件或代码中设置：

1. 在配置文件中添加钩子 #

修改hypercorn.config.py，添加异步回调函数：

import multiprocessing as mp
import os
import asyncio

print( (mp.cpu_count() * 2) + 1)

accesslog = "-"
backlog = 500
bind = "0.0.0.0:8000"
statsd_host = os.environ.get("STASD_HOST")
workers = 3
max_requests = 2

# 对应Gunicorn的on_starting
async def on_starting():
    print("Starting Starlette application")

# 对应Gunicorn的on_exit
async def on_shutdown():
    print("Shutting down Starlette application")

# 对应Gunicorn的worker_exit
async def worker_exit(worker_id):
    print(f"Worker {worker_id} exiting")

# 绑定回调到配置
on_starting = on_starting
on_shutdown = on_shutdown
worker_exit = worker_exit

2. 在代码中配置钩子 #

通过代码启动Hypercorn时，可直接在Config对象上设置：

if __name__ == "__main__":
    from hypercorn.config import Config

    config = Config()
    config.bind = ["localhost:8077"]
    config.debug = True

    # 定义钩子函数
    async def on_starting():
        print("Starting application via code")

    async def on_shutdown():
        print("Shutting down application via code")

    # 绑定到配置
    config.on_starting = on_starting
    config.on_shutdown = on_shutdown

    import asyncio
    from hypercorn.asyncio import serve

    asyncio.run(serve(app, config))

3. 支持的主要回调事件 #

Hypercorn核心异步回调包括：

• on_starting: 服务器启动前触发
• on_shutdown: 服务器关闭时触发
• worker_exit: 工作进程退出时触发（传入worker_id参数）
• worker_ready: 工作进程准备就绪时触发（传入worker_id参数）

所有回调需用async def定义，适配Hypercorn的异步IO运行机制。

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