AIOHTTP - Application.make_handler(...) 已弃用 - 添加多处理
AIOHTTP - Application.make_handler(...) is deprecated - Adding Multiprocessing
我踏上了 "How much performance can I squeeze out of a Python web-server?" 的旅程,这让我了解了 AIOHTTP 和 uvloop。不过,我可以看出 AIOHTTP 并没有充分发挥我的 CPU 的潜力。我开始使用 AIOHTTP 进行多处理。我了解到有一个 Linux 内核功能允许多个进程共享同一个 TCP 端口。这导致我开发了以下代码(效果非常好):
import asyncio
import os
import socket
import time
from aiohttp import web
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import cpu_count
CPU_COUNT = cpu_count()
print("CPU Count:", CPU_COUNT)
def mk_socket(host="127.0.0.1", port=8000, reuseport=False):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if reuseport:
SO_REUSEPORT = 15
sock.setsockopt(socket.SOL_SOCKET, SO_REUSEPORT, 1)
sock.bind((host, port))
return sock
async def handle(request):
name = request.match_info.get('name', "Anonymous")
pid = os.getpid()
text = "{:.2f}: Hello {}! Process {} is treating you\n".format(
time.time(), name, pid)
#time.sleep(5) # intentionally blocking sleep to simulate CPU load
return web.Response(text=text)
def start_server():
host = "127.0.0.1"
port=8000
reuseport = True
app = web.Application()
sock = mk_socket(host, port, reuseport=reuseport)
app.add_routes([web.get('/', handle),
web.get('/{name}', handle)])
loop = asyncio.get_event_loop()
coro = loop.create_server(
protocol_factory=app.make_handler(),
sock=sock,
)
srv = loop.run_until_complete(coro)
loop.run_forever()
if __name__ == '__main__':
with ProcessPoolExecutor() as executor:
for i in range(0, CPU_COUNT):
executor.submit(start_server)
应用此代码之前我网站的 wrk 基准测试:
Running 30s test @ http://127.0.0.1:8000/
12 threads and 400 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 54.33ms 6.54ms 273.24ms 89.95%
Req/Sec 608.68 115.97 2.27k 83.63%
218325 requests in 30.10s, 41.23MB read
Non-2xx or 3xx responses: 218325
Requests/sec: 7254.17
Transfer/sec: 1.37MB
在以下时间之后进行基准测试:
Running 30s test @ http://127.0.0.1:8000/
12 threads and 400 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 15.96ms 7.27ms 97.29ms 84.78%
Req/Sec 2.11k 208.30 4.45k 75.50%
759290 requests in 30.08s, 153.51MB read
Requests/sec: 25242.39
Transfer/sec: 5.10MB
哇!但是有一个问题:
DeprecationWarning: Application.make_handler(...) is deprecated, use AppRunner API instead
protocol_factory=app.make_handler()
所以我尝试了这个:
import asyncio
import os
import socket
import time
from aiohttp import web
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import cpu_count
CPU_COUNT = cpu_count()
print("CPU Count:", CPU_COUNT)
def mk_socket(host="127.0.0.1", port=8000, reuseport=False):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if reuseport:
SO_REUSEPORT = 15
sock.setsockopt(socket.SOL_SOCKET, SO_REUSEPORT, 1)
sock.bind((host, port))
return sock
async def handle(request):
name = request.match_info.get('name', "Anonymous")
pid = os.getpid()
text = "{:.2f}: Hello {}! Process {} is treating you\n".format(
time.time(), name, pid)
#time.sleep(5) # intentionally blocking sleep to simulate CPU load
return web.Response(text=text)
async def start_server():
host = "127.0.0.1"
port=8000
reuseport = True
app = web.Application()
sock = mk_socket(host, port, reuseport=reuseport)
app.add_routes([web.get('/', handle),
web.get('/{name}', handle)])
coro = loop.create_server(
protocol_factory=app.make_handler(),
sock=sock,
)
runner = web.AppRunner(app)
await runner.setup()
srv = web.TCPSite(runner, 'localhost', 8000)
await srv.start()
print('Server started at http://127.0.0.1:8000')
return coro, app, runner
async def finalize(srv, app, runner):
sock = srv.sockets[0]
app.loop.remove_reader(sock.fileno())
sock.close()
#await handler.finish_connections(1.0)
await runner.cleanup()
srv.close()
await srv.wait_closed()
await app.finish()
def init():
loop = asyncio.get_event_loop()
srv, app, runner = loop.run_until_complete(init)
try:
loop.run_forever()
except KeyboardInterrupt:
loop.run_until_complete((finalize(srv, app, runner)))
if __name__ == '__main__':
with ProcessPoolExecutor() as executor:
for i in range(0, CPU_COUNT):
executor.submit(init)
这显然是不完整的,因为 coro
没有被使用。我不确定在哪里将套接字与 AppRunner 集成。答案应显示修改为使用 App Runner 的原始示例。
因为这是我第一次使用协同程序和 aiohttp,所以我可能错了,但它似乎可以与 SockSite 一起使用:
#!/usr/bin/env python
import asyncio
import os
import socket
import time
import traceback
from aiohttp import web
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import cpu_count
CPU_COUNT = cpu_count()
print("CPU Count:", CPU_COUNT)
def mk_socket(host="127.0.0.1", port=9090, reuseport=False):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if reuseport:
SO_REUSEPORT = 15
sock.setsockopt(socket.SOL_SOCKET, SO_REUSEPORT, 1)
sock.bind((host, port))
return sock
async def handle(request):
name = request.match_info.get('name', "Anonymous")
pid = os.getpid()
text = "{:.2f}: Hello {}! Process {} is treating you\n".format(
time.time(), name, pid)
#time.sleep(5) # intentionally blocking sleep to simulate CPU load
return web.Response(text=text)
async def start_server():
try:
host = "127.0.0.1"
port=9090
reuseport = True
app = web.Application()
app.add_routes([web.get('/', handle),
web.get('/{name}', handle)])
runner = web.AppRunner(app)
await runner.setup()
sock = mk_socket(host, port, reuseport=reuseport)
srv = web.SockSite(runner, sock)
await srv.start()
print('Server started at http://127.0.0.1:9090')
return srv, app, runner
except Exception:
traceback.print_exc()
raise
async def finalize(srv, app, runner):
sock = srv.sockets[0]
app.loop.remove_reader(sock.fileno())
sock.close()
#await handler.finish_connections(1.0)
await runner.cleanup()
srv.close()
await srv.wait_closed()
await app.finish()
def init():
loop = asyncio.get_event_loop()
srv, app, runner = loop.run_until_complete(start_server())
try:
loop.run_forever()
except KeyboardInterrupt:
loop.run_until_complete((finalize(srv, app, runner)))
if __name__ == '__main__':
with ProcessPoolExecutor() as executor:
for i in range(0, CPU_COUNT):
executor.submit(init)
最终:
>curl http://127.0.0.1:9090
1580741746.47: Hello Anonymous! Process 54623 is treating you
>curl http://127.0.0.1:9090
1580741747.05: Hello Anonymous! Process 54620 is treating you
>curl http://127.0.0.1:9090
1580741747.77: Hello Anonymous! Process 54619 is treating you
>curl http://127.0.0.1:9090
1580741748.36: Hello Anonymous! Process 54621 is treating you
我还在 finalize 例程中添加了一个日志,它似乎正确触发了。
编辑:而且,出于好奇,我在旧内核上试了一下,我确认启用该选项时它不起作用(它适用于 False)。
我踏上了 "How much performance can I squeeze out of a Python web-server?" 的旅程,这让我了解了 AIOHTTP 和 uvloop。不过,我可以看出 AIOHTTP 并没有充分发挥我的 CPU 的潜力。我开始使用 AIOHTTP 进行多处理。我了解到有一个 Linux 内核功能允许多个进程共享同一个 TCP 端口。这导致我开发了以下代码(效果非常好):
import asyncio
import os
import socket
import time
from aiohttp import web
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import cpu_count
CPU_COUNT = cpu_count()
print("CPU Count:", CPU_COUNT)
def mk_socket(host="127.0.0.1", port=8000, reuseport=False):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if reuseport:
SO_REUSEPORT = 15
sock.setsockopt(socket.SOL_SOCKET, SO_REUSEPORT, 1)
sock.bind((host, port))
return sock
async def handle(request):
name = request.match_info.get('name', "Anonymous")
pid = os.getpid()
text = "{:.2f}: Hello {}! Process {} is treating you\n".format(
time.time(), name, pid)
#time.sleep(5) # intentionally blocking sleep to simulate CPU load
return web.Response(text=text)
def start_server():
host = "127.0.0.1"
port=8000
reuseport = True
app = web.Application()
sock = mk_socket(host, port, reuseport=reuseport)
app.add_routes([web.get('/', handle),
web.get('/{name}', handle)])
loop = asyncio.get_event_loop()
coro = loop.create_server(
protocol_factory=app.make_handler(),
sock=sock,
)
srv = loop.run_until_complete(coro)
loop.run_forever()
if __name__ == '__main__':
with ProcessPoolExecutor() as executor:
for i in range(0, CPU_COUNT):
executor.submit(start_server)
应用此代码之前我网站的 wrk 基准测试:
Running 30s test @ http://127.0.0.1:8000/
12 threads and 400 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 54.33ms 6.54ms 273.24ms 89.95%
Req/Sec 608.68 115.97 2.27k 83.63%
218325 requests in 30.10s, 41.23MB read
Non-2xx or 3xx responses: 218325
Requests/sec: 7254.17
Transfer/sec: 1.37MB
在以下时间之后进行基准测试:
Running 30s test @ http://127.0.0.1:8000/
12 threads and 400 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 15.96ms 7.27ms 97.29ms 84.78%
Req/Sec 2.11k 208.30 4.45k 75.50%
759290 requests in 30.08s, 153.51MB read
Requests/sec: 25242.39
Transfer/sec: 5.10MB
哇!但是有一个问题:
DeprecationWarning: Application.make_handler(...) is deprecated, use AppRunner API instead
protocol_factory=app.make_handler()
所以我尝试了这个:
import asyncio
import os
import socket
import time
from aiohttp import web
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import cpu_count
CPU_COUNT = cpu_count()
print("CPU Count:", CPU_COUNT)
def mk_socket(host="127.0.0.1", port=8000, reuseport=False):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if reuseport:
SO_REUSEPORT = 15
sock.setsockopt(socket.SOL_SOCKET, SO_REUSEPORT, 1)
sock.bind((host, port))
return sock
async def handle(request):
name = request.match_info.get('name', "Anonymous")
pid = os.getpid()
text = "{:.2f}: Hello {}! Process {} is treating you\n".format(
time.time(), name, pid)
#time.sleep(5) # intentionally blocking sleep to simulate CPU load
return web.Response(text=text)
async def start_server():
host = "127.0.0.1"
port=8000
reuseport = True
app = web.Application()
sock = mk_socket(host, port, reuseport=reuseport)
app.add_routes([web.get('/', handle),
web.get('/{name}', handle)])
coro = loop.create_server(
protocol_factory=app.make_handler(),
sock=sock,
)
runner = web.AppRunner(app)
await runner.setup()
srv = web.TCPSite(runner, 'localhost', 8000)
await srv.start()
print('Server started at http://127.0.0.1:8000')
return coro, app, runner
async def finalize(srv, app, runner):
sock = srv.sockets[0]
app.loop.remove_reader(sock.fileno())
sock.close()
#await handler.finish_connections(1.0)
await runner.cleanup()
srv.close()
await srv.wait_closed()
await app.finish()
def init():
loop = asyncio.get_event_loop()
srv, app, runner = loop.run_until_complete(init)
try:
loop.run_forever()
except KeyboardInterrupt:
loop.run_until_complete((finalize(srv, app, runner)))
if __name__ == '__main__':
with ProcessPoolExecutor() as executor:
for i in range(0, CPU_COUNT):
executor.submit(init)
这显然是不完整的,因为 coro
没有被使用。我不确定在哪里将套接字与 AppRunner 集成。答案应显示修改为使用 App Runner 的原始示例。
因为这是我第一次使用协同程序和 aiohttp,所以我可能错了,但它似乎可以与 SockSite 一起使用:
#!/usr/bin/env python
import asyncio
import os
import socket
import time
import traceback
from aiohttp import web
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import cpu_count
CPU_COUNT = cpu_count()
print("CPU Count:", CPU_COUNT)
def mk_socket(host="127.0.0.1", port=9090, reuseport=False):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if reuseport:
SO_REUSEPORT = 15
sock.setsockopt(socket.SOL_SOCKET, SO_REUSEPORT, 1)
sock.bind((host, port))
return sock
async def handle(request):
name = request.match_info.get('name', "Anonymous")
pid = os.getpid()
text = "{:.2f}: Hello {}! Process {} is treating you\n".format(
time.time(), name, pid)
#time.sleep(5) # intentionally blocking sleep to simulate CPU load
return web.Response(text=text)
async def start_server():
try:
host = "127.0.0.1"
port=9090
reuseport = True
app = web.Application()
app.add_routes([web.get('/', handle),
web.get('/{name}', handle)])
runner = web.AppRunner(app)
await runner.setup()
sock = mk_socket(host, port, reuseport=reuseport)
srv = web.SockSite(runner, sock)
await srv.start()
print('Server started at http://127.0.0.1:9090')
return srv, app, runner
except Exception:
traceback.print_exc()
raise
async def finalize(srv, app, runner):
sock = srv.sockets[0]
app.loop.remove_reader(sock.fileno())
sock.close()
#await handler.finish_connections(1.0)
await runner.cleanup()
srv.close()
await srv.wait_closed()
await app.finish()
def init():
loop = asyncio.get_event_loop()
srv, app, runner = loop.run_until_complete(start_server())
try:
loop.run_forever()
except KeyboardInterrupt:
loop.run_until_complete((finalize(srv, app, runner)))
if __name__ == '__main__':
with ProcessPoolExecutor() as executor:
for i in range(0, CPU_COUNT):
executor.submit(init)
最终:
>curl http://127.0.0.1:9090
1580741746.47: Hello Anonymous! Process 54623 is treating you
>curl http://127.0.0.1:9090
1580741747.05: Hello Anonymous! Process 54620 is treating you
>curl http://127.0.0.1:9090
1580741747.77: Hello Anonymous! Process 54619 is treating you
>curl http://127.0.0.1:9090
1580741748.36: Hello Anonymous! Process 54621 is treating you
我还在 finalize 例程中添加了一个日志,它似乎正确触发了。
编辑:而且,出于好奇,我在旧内核上试了一下,我确认启用该选项时它不起作用(它适用于 False)。