Python 不阻塞主线程下启动 Web 服务

Quan

December 24, 2021

3350views

No comments

2396 words

代码

## 前言

经常遇到使用Web服务的时候，想再起个定时任务干点其他的事情，但这个时候Web服务就会阻塞掉主线程，让定时任务卡住，不执行。于是就有了下面这段代码

```python
import time
from aiohttp import web
from threading import Thread
import asyncio

class MyThreadedServer(Thread):
    def __init__(self):
        self._loop = None  
        self.hostname = '0.0.0.0'
        self.port = 5000

async def _on_shutdown(self, app):
        for conn in set(self.connections):
            await conn.close()

def run(self):
        self._loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self._loop)
        self.stop_server = asyncio.Event()
        self._loop.run_until_complete(self._run_app())
        self._loop.close()

async def _run_app(self):
        self.app = web.Application()
        self.app.on_shutdown.append(self._on_shutdown)
        runner = web.AppRunner(self.app)
        await runner.setup()
        site = web.TCPSite(runner, self.hostname, self.port)
        await site.start()
        print('Started web server on %s port %d' % (self.hostname, self.port))
        await self.stop_server.wait()
        print('Web server closing down')

def stop(self):
        "Call from any thread"
        # Overrides EventThread.stop() to change implementation
        if self.stop_server:
            self._loop.call_soon_threadsafe(self.stop_server.set)

ws_server = MyThreadedServer()
Thread(target=ws_server.run).start()
time.sleep(1)
ws_server.stop()
print("close")
```

#### 输出

```bash
Started web server on 0.0.0.0 port 5000
close
Web server closing down
```

### FastAPI
```python
import time
import fastapi
import uvicorn

class WebServer():
    def __init__(self,host="0.0.0.0", port=8000,**kwargs):
        self.app = fastapi.FastAPI()
        self.host = host
        self.port = port
        self.kwargs = kwargs
        self.background = None
    
    def start(self):
        uvicorn.run(app=self.app, host=self.host, port=self.port, **self.kwargs)
    
    def start_background(self):
        import threading
        t = threading.Thread(target=self.start, daemon=True)
        t.start()
        self.background = t
        return t

web = WebServer()

@web.app.get("/")
def root():
    return {"message": "Hello World"}

# web.start()
web.start_background()

while True:
    print("b")
    time.sleep(10)
```

Last modification：March 9th, 2022 at 10:44 am

If you think my article is useful to you, please feel free to appreciate

Python 不阻塞主线程下启动 Web 服务