fastapi - 同步接口也不会有IO阻塞!

之前一直以为FastAPI对于同步IO会发生服务阻塞,直到今天看到了这一块代码,原来同步的函数会开线程去处理

fastapi/routing.py

async def run_endpoint_function(
    *, dependant: Dependant, values: Dict[str, Any], is_coroutine: bool
) -> Any:
    # Only called by get_request_handler. Has been split into its own function to
    # facilitate profiling endpoints, since inner functions are harder to profile.
    assert dependant.call is not None, "dependant.call must be a function"

    if is_coroutine:
        return await dependant.call(**values)
    else:
        return await run_in_threadpool(dependant.call, **values)

starlette/concurrency.py

async def run_in_threadpool(
    func: typing.Callable[P, T], *args: P.args, **kwargs: P.kwargs
) -> T:
    if kwargs:  # pragma: no cover
        # run_sync doesn't accept 'kwargs', so bind them in here
        func = functools.partial(func, **kwargs)
    return await anyio.to_thread.run_sync(func, *args)

这么一来的话其实即使是同步IO也不至于严重影响FastAPI的性能,那么对于并发要求不是很高的项目也可以用传统的同步IO库去开发了 ^-

专门做了一个anyio.to_thread.run_sync的测试

import time
import asyncio
from threading import current_thread

from anyio import to_thread, run


def work(i):
    time.sleep(3)
    print(f"{i}(index) {int(time.time())}(timestamp) {current_thread().native_id}(thread_id)")


async def main():
    coroutines = []
    for i in range(10):
        coroutines.append(to_thread.run_sync(work, i))
    await asyncio.gather(*coroutines)


run(main)

0(index) 1715408207(timestamp) 5642751(thread_id)
5(index) 1715408207(timestamp) 5642756(thread_id)
1(index) 1715408207(timestamp) 5642752(thread_id)
2(index) 1715408207(timestamp) 5642753(thread_id)
4(index) 1715408207(timestamp) 5642755(thread_id)
6(index) 1715408207(timestamp) 5642757(thread_id)
8(index) 1715408207(timestamp) 5642759(thread_id)
7(index) 1715408207(timestamp) 5642758(thread_id)
9(index) 1715408207(timestamp) 5642760(thread_id)
3(index) 1715408207(timestamp) 5642754(thread_id)

确实可以并发!