AIOHTTP绑定端口遇[Errno 10048]错误，如何直接处理？

一、主动捕获绑定错误并强制终止进程 #

由于你的数据库框架会维持事件循环运行，aiohttp绑定端口失败后不会自动终止进程。你需要在启动逻辑中显式捕获OSError(10048)错误，直接退出整个进程：

import asyncio
import sys
from aiohttp import web

async def main():
    try:
        app = web.Application()
        # 注册路由、中间件等操作
        runner = web.AppRunner(app)
        await runner.setup()
        site = web.TCPSite(runner, '0.0.0.0', 8080)
        await site.start()
        print("Server started on 0.0.0.0:8080")
    except OSError as e:
        if e.errno == 10048:
            print(f"端口8080已被占用: {e}", file=sys.stderr)
            # 直接终止进程，避免事件循环被数据库框架维持导致服务僵死
            sys.exit(1)
        else:
            # 其他OSError正常抛出
            raise

# 数据库框架初始化逻辑
async def init_db():
    # 你的异步数据库初始化、长期运行任务逻辑
    pass

async def run_all():
    await init_db()
    await main()

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

二、设置Socket重用选项 #

通过自定义Socket并开启SO_REUSEADDR（Windows/Linux通用）和SO_REUSEPORT（Linux/macOS支持）选项，让端口可以被快速重用，减少端口未释放导致的绑定失败：

import asyncio
import socket
from aiohttp import web

async def main():
    app = web.Application()
    # 注册路由等操作

    # 创建自定义Socket并设置重用参数
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    # Windows下SO_REUSEPORT可能不支持，做兼容处理
    try:
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
    except AttributeError:
        pass
    sock.bind(('0.0.0.0', 8080))

    runner = web.AppRunner(app)
    await runner.setup()
    # 使用自定义Socket启动站点
    site = web.SockSite(runner, sock)
    await site.start()
    print("Server started on 0.0.0.0:8080")

# 数据库初始化和run_all函数同上

三、启动前检测端口占用状态 #

在启动aiohttp之前先检测端口是否被占用，提前终止进程避免进入僵死状态：

import socket
import sys

def is_port_occupied(port):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        # 尝试连接端口，返回0表示已被占用
        return s.connect_ex(('0.0.0.0', port)) == 0

async def main():
    if is_port_occupied(8080):
        print("端口8080已被占用，退出进程", file=sys.stderr)
        sys.exit(1)
    # 后续aiohttp启动逻辑

四、统一管理事件循环，清理后台任务 #

如果数据库框架有长期运行的异步任务，在aiohttp启动失败时主动取消这些任务，让事件循环可以正常退出：

import asyncio
import sys
from aiohttp import web

db_background_task = None

async def long_running_db_task():
    # 数据库后台长期运行任务，比如监听变更
    while True:
        await asyncio.sleep(3600)

async def init_db():
    global db_background_task
    db_background_task = asyncio.create_task(long_running_db_task())
    await db_background_task

async def main():
    try:
        app = web.Application()
        # 注册路由等操作
        runner = web.AppRunner(app)
        await runner.setup()
        site = web.TCPSite(runner, '0.0.0.0', 8080)
        await site.start()
        print("Server started on 0.0.0.0:8080")
    except OSError as e:
        if e.errno == 10048:
            print(f"端口8080被占用: {e}", file=sys.stderr)
            # 取消数据库后台任务，让事件循环可以退出
            if db_background_task:
                db_background_task.cancel()
                try:
                    await db_background_task
                except asyncio.CancelledError:
                    pass
            sys.exit(1)
        else:
            raise

async def run_all():
    await init_db()
    await main()

if __name__ == "__main__":
    try:
        asyncio.run(run_all())
    except asyncio.CancelledError:
        pass

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