aiohttp 高并发web服务_python后台——asyncio,aiohttp入门教程,多进程+asyncio
全栈工程师开发手册 (作者:栾鹏)
架构系列文章
注意: python的异步返回有时是通过异常expection向上冒泡的, 在异步函数中使用try catch有时无法捕获某些异常的, 异常会作为返回结果直接返回给上一层
很多朋友对异步编程都处于“听说很强大”的认知状态。鲜有在生产项目中使用它。而使用它的同学,则大多数都停留在知道如何使用 Tornado、Twisted、Gevent 这类异步框架上,出现各种古怪的问题难以解决。而且使用了异步框架的部分同学,由于用法不对,感觉它并没牛逼到哪里去,所以很多同学做 Web 后端服务时还是采用 Flask、Django等传统的非异步框架。
从上两届 PyCon 技术大会看来,异步编程已经成了 Python 生态下一阶段的主旋律。如新兴的 Go、Rust、Elixir 等编程语言都将其支持异步和高并发作为主要“卖点”,技术变化趋势如此。Python 生态为不落人后,从2013年起由 Python 之父 Guido 亲自操刀主持了Tulip(asyncio)项目的开发。
异步io的好处在于避免的线程的开销和切换,而且我们都知道python其实是没有多线程的,只是通过底层线层锁实现的多线程。另一个好处在于避免io操作(包含网络传输)的堵塞时间。
asyncio可以实现单线程并发IO操作。如果仅用在客户端,发挥的威力不大。如果把asyncio用在服务器端,例如Web服务器,由于HTTP连接就是IO操作,因此可以用单线程+coroutine实现多用户的高并发支持。
asyncio实现了TCP、UDP、SSL等协议,aiohttp则是基于asyncio实现的HTTP框架。
对于异步io你需要知道的重点,要注意的是,await语法只能出现在通过async修饰的函数中,否则会报SyntaxError错误。而且await后面的对象需要是一个Awaitable,或者实现了相关的协议。
注意:
所有需要异步执行的函数,都需要asyncio中的轮训器去轮训执行,如果函数阻塞,轮训器就会去执行下一个函数。所以所有需要异步执行的函数都需要加入到这个轮训器中。
asyncio
demo:
import requests
import time
import asyncio
# 创建一个异步函数
async def task_func():
await asyncio.sleep(1)
resp = requests.get('http://192.168.2.177:5002/')
print('2222222',time.time(),resp.text)
async def main(loop):
loop=asyncio.get_event_loop() # 获取全局轮训器
task = loop.create_task(task_func()) # 在全局轮训器加入协成,只有加入全局轮训器才能被监督执行
await asyncio.sleep(2) # 等待两秒为了不要立即执行event_loop.close(),项目中event_loop应该是永不停歇的
print('11111111111',time.time())
event_loop = asyncio.get_event_loop()
try:
event_loop.run_until_complete(main(event_loop))
finally:
event_loop.close() # 当轮训器关闭以后,所有没有执行完成的协成将全部关闭
所谓「异步 IO」,就是你发起一个 IO 操作,却不用等它结束,你可以继续做其他事情,当它结束时,你会得到通知。
Asyncio 是并发(concurrency)的一种方式。对 Python 来说,并发还可以通过线程(threading)和多进程(multiprocessing)来实现。
Asyncio 并不能带来真正的并行(parallelism)。当然,因为 GIL(全局解释器锁)的存在,Python 的多线程也不能带来真正的并行。
可交给 asyncio 执行的任务,称为协程(coroutine)。一个协程可以放弃执行,把机会让给其它协程(即 yield from 或 await)。`
定义协程
协程的定义,需要使用 async def 语句。
async def do_some_work(x): pass
do_some_work 便是一个协程。
准确来说,do_some_work 是一个协程函数,可以通过 asyncio.iscoroutinefunction 来验证:
print(asyncio.iscoroutinefunction(do_some_work)) # True
这个协程什么都没做,我们让它睡眠几秒,以模拟实际的工作量 :
async def do_some_work(x):
print("Waiting " + str(x))
await asyncio.sleep(x)
在解释 await 之前,有必要说明一下协程可以做哪些事。协程可以:
`* 等待一个 future 结束
* 等待另一个协程(产生一个结果,或引发一个异常)
* 产生一个结果给正在等它的协程
* 引发一个异常给正在等它的协程
`
asyncio.sleep 也是一个协程,所以 await asyncio.sleep(x) 就是等待另一个协程。可参见 asyncio.sleep 的文档:
sleep(delay, result=None, *, loop=None)
Coroutine that completes after a given time (in seconds).
运行协程
调用协程函数,协程并不会开始运行,只是返回一个协程对象,可以通过 asyncio.iscoroutine 来验证:
print(asyncio.iscoroutine(do_some_work(3))) # True
此处还会引发一条警告:
async1.py:16: RuntimeWarning: coroutine 'do_some_work' was never awaited
print(asyncio.iscoroutine(do_some_work(3)))
要让这个协程对象运行的话,有两种方式:
`* 在另一个已经运行的协程中用 `await` 等待它
* 通过 `ensure_future` 函数计划它的执行
`
简单来说,只有 loop 运行了,协程才可能运行。
下面先拿到当前线程缺省的 loop ,然后把协程对象交给 loop.run_until_complete,协程对象随后会在 loop 里得到运行。
loop = asyncio.get_event_loop()
loop.run_until_complete(do_some_work(3))
run_until_complete 是一个阻塞(blocking)调用,直到协程运行结束,它才返回。这一点从函数名不难看出。
run_until_complete 的参数是一个 future,但是我们这里传给它的却是协程对象,之所以能这样,是因为它在内部做了检查,通过 ensure_future 函数把协程对象包装(wrap)成了 future。所以,我们可以写得更明显一些:
loop.run_until_complete(asyncio.ensure_future(do_some_work(3)))
完整代码:
import asyncio
async def do_some_work(x):
print("Waiting " + str(x))
await asyncio.sleep(x)
loop = asyncio.get_event_loop()
loop.run_until_complete(do_some_work(3))
运行结果:
`Waiting 3
`
回调
假如协程是一个 IO 的读操作,等它读完数据后,我们希望得到通知,以便下一步数据的处理。这一需求可以通过往 future 添加回调来实现。
def done_callback(futu):
print('Done')
futu = asyncio.ensure_future(do_some_work(3))
futu.add_done_callback(done_callback)
loop.run_until_complete(futu)
多个协程
实际项目中,往往有多个协程,同时在一个 loop 里运行。为了把多个协程交给 loop,需要借助 asyncio.gather 函数。
loop.run_until_complete(asyncio.gather(do_some_work(1), do_some_work(3)))
或者先把协程存在列表里:
coros = [do_some_work(1), do_some_work(3)]
loop.run_until_complete(asyncio.gather(*coros))
运行结果:
`Waiting 3
Waiting 1
Done`
这两个协程是并发运行的,所以等待的时间不是1 + 3 = 4 秒,而是以耗时较长的那个协程为准。
参考函数 gather 的文档:
gather(*coros_or_futures, loop=None, return_exceptions=False)
Return 各协程协成的合并结果
发现也可以传futures给它:
futus = [asyncio.ensure_future(do_some_work(1)),
asyncio.ensure_future(do_some_work(3))]
loop.run_until_complete(asyncio.gather(*futus))
如果进程中已经有了loop, 则可以直接等待这几个异步的结果
results = await asyncio.gather(*futus)
返回结果results是一个列表,每个元素是每个异步的返回内容
print(str(results))
gather 起聚合的作用,把多个 futures 包装成单个 future,因为 loop.run_until_complete 只接受单个 future。
run_until_complete 和 run_forever
我们一直通过 run_until_complete 来运行 loop ,等到 future 完成,run_until_complete 也就返回了。
async def do_some_work(x):
print('Waiting ' + str(x))
await asyncio.sleep(x)
print('Done')
loop = asyncio.get_event_loop()
coro = do_some_work(3)
loop.run_until_complete(coro)
输出:
`Waiting 3
Done
`
现在改用 run_forever:
async def do_some_work(x):
print('Waiting ' + str(x))
await asyncio.sleep(x)
print('Done')
loop = asyncio.get_event_loop()
coro = do_some_work(3)
asyncio.ensure_future(coro)
loop.run_forever()
输出:
`Waiting 3
Done
`
三秒钟过后,future 结束,但是程序并不会退出。run_forever 会一直运行,直到 stop 被调用,但是你不能像下面这样调 stop:
loop.run_forever()
loop.stop()
run_forever 不返回,stop 永远也不会被调用。所以,只能在协程中调 stop:
async def do_some_work(loop, x):
print('Waiting ' + str(x))
await asyncio.sleep(x)
print('Done')
loop.stop()
这样并非没有问题,假如有多个协程在 loop 里运行:
asyncio.ensure_future(do_some_work(loop, 1))
asyncio.ensure_future(do_some_work(loop, 3))
loop.run_forever()
第二个协程没结束,loop 就停止了——被先结束的那个协程给停掉的。
要解决这个问题,可以用 gather 把多个协程合并成一个 future,并添加回调,然后在回调里再去停止 loop。
async def do_some_work(loop, x):
print('Waiting ' + str(x))
await asyncio.sleep(x)
print('Done')
def done_callback(loop, futu):
loop.stop()
loop = asyncio.get_event_loop()
futus = asyncio.gather(do_some_work(loop, 1), do_some_work(loop, 3))
futus.add_done_callback(functools.partial(done_callback, loop))
loop.run_forever()
其实这基本上就是 run_until_complete 的实现了,run_until_complete 在内部也是调用 run_forever。
Close Loop?
以上示例都没有调用 loop.close,好像也没有什么问题。所以到底要不要调 loop.close 呢?
简单来说,loop 只要不关闭,就还可以再运行。:
loop.run_until_complete(do_some_work(loop, 1))
loop.run_until_complete(do_some_work(loop, 3))
loop.close()
但是如果关闭了,就不能再运行了:
loop.run_until_complete(do_some_work(loop, 1))
loop.close()
loop.run_until_complete(do_some_work(loop, 3)) # 此处异常
建议调用 loop.close,以彻底清理 loop 对象防止误用。
gather vs. wait
asyncio.gather 和 asyncio.wait 功能相似。
coros = [do_some_work(loop, 1), do_some_work(loop, 3)]
loop.run_until_complete(asyncio.wait(coros))
具体差别可请参见 StackOverflow 的讨论:Asyncio.gather vs asyncio.wait。
Timer
C++ Boost.Asio 提供了 IO 对象 timer,但是 Python 并没有原生支持 timer,不过可以用 asyncio.sleep 模拟。
async def timer(x, cb):
futu = asyncio.ensure_future(asyncio.sleep(x))
futu.add_done_callback(cb)
await futu
t = timer(3, lambda futu: print('Done'))
loop.run_until_complete(t)
将阻塞操作定义为协成
在协成中,如果有一个操作阻塞了,会影响其他的一串协成组中的其他协成,我们可以把容易产生阻塞的协成定义为可调出协成
例如:
import asyncio
import requests
import time,datetime
def my_fun():
time.sleep(4)
return datetime.datetime.now()
async def fetch_async(func, *args):
print('begin')
loop = asyncio.get_event_loop()
future = loop.run_in_executor(None, func, *args)
result = await future
return result
tasks = [
fetch_async(my_fun),
fetch_async(my_fun)
]
loop = asyncio.get_event_loop()
results = loop.run_until_complete(asyncio.gather(*tasks))
print(results)
tasks = [
fetch_async(my_fun),
fetch_async(my_fun)
]
results = loop.run_until_complete(asyncio.gather(*tasks))
print(results)
tasks = [
fetch_async(my_fun),
fetch_async(my_fun)
]
results = loop.run_until_complete(asyncio.gather(*tasks))
print(results)
loop.close()
这两个time.sleep相互不影响
aiohttp服务器
下面是aiohttp作为服务器端的一个简单的demo。
#!/usr/bin/env python3
import argparse
from aiohttp import web
import asyncio
import base64
import logging
import uvloop
import time,datetime
import json
import requests
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
routes = web.RouteTableDef()
@routes.get('/')
async def hello(request):
return web.Response(text="Hello, world")
# 定义一个路由映射,接收网址参数,post方式
@routes.post('/demo1/{name}')
async def demo1(request): # 异步监听,只要一有握手就开始触发,此时网址参数中的name就已经知道了,但是前端可能还没有完全post完数据。
name = request.match_info.get('name', "Anonymous") # 获取name
print(datetime.datetime.now()) # 触发视图函数的时间
data = await request.post() # 等待post数据完成接收,只有接收完成才能进行后续操作.data['key']获取参数
print(datetime.datetime.now()) # 接收post数据完成的时间
logging.info('safety dect request start %s' % datetime.datetime.now())
result = {'name':name,'key':data['key']}
logging.info('safety dect request finish %s, %s' % (datetime.datetime.now(),json.dumps(result)))
return web.json_response(result)
# 定义一个路由映射,设计到io操作
@routes.post('/demo2')
async def demo2(request): # 异步监听,只要一有握手就开始触发,此时网址参数中的name就已经知道了,但是前端可能还没有完全post完数据。
data = await request.post() # 等待post数据完成接收,只有接收完成才能进行后续操作.data['key']获取参数
logging.info('safety dect request start %s' % datetime.datetime.now())
res = requests.post('http://www.baidu.com') # 网路id,会自动切换到其他协成上
logging.info('safety dect request finish %s' % res.test)
return web.Response(text="welcome")
if __name__ == '__main__':
logging.info('server start')
app = web.Application()
app.add_routes(routes)
web.run_app(app,host='0.0.0.0',port=8080)
logging.info('server close')
aiohttp客户端
aiohttp的另一个主要作用是作为异步客户端,用来解决高并发请求的情况。比如现在我要模拟一个高并发请求来测试我的服务器负载情况。所以需要在python里模拟高并发。高并发可以有多种方式,比如多线程,但是由于python本质上是没有多线程的,通过底层线程锁实现的多线程。在模型高并发时,具有线程切换和线程开销的损耗。所以我们就可以使用多协成来实现高并发。
我们就可以使用aiohttp来模拟高并发客户端。demo如下,用来模拟多个客户端向指定服务器post图片。
# 异步并发客户端
class Asyncio_Client(object):
def __init__(self):
self.loop=asyncio.get_event_loop()
self.tasks=[]
# 将异步函数介入任务列表。后续参数直接传给异步函数
def set_task(self,task_fun,num,*args):
for i in range(num):
self.tasks.append(task_fun(*args))
# 运行,获取返回结果
def run(self):
back=[]
try:
f = asyncio.wait(self.tasks) # 创建future
self.loop.run_until_complete(f) # 等待future完成
finally:
pass
# 服务器高并发压力测试
class Test_Load():
total_time=0 # 总耗时
total_payload=0 # 总负载
total_num=0 # 总并发数
all_time=[]
# 创建一个异步任务,本地测试,所以post和接收几乎不损耗时间,可以等待完成,主要耗时为算法模块
async def task_func1(self,session):
begin = time.time()
# print('开始发送:', begin)
file=open(self.image, 'rb')
fsize = os.path.getsize(self.image)
self.total_payload+=fsize/(1024*1024)
data = {"image_id": "2", 'image':file}
r = await session.post(self.url,data=data) #只post,不接收
result = await r.json()
self.total_num+=1
# print(result)
end = time.time()
# print('接收完成:', end,',index=',self.total_num)
self.all_time.append(end-begin)
# 负载测试
def test_safety(self):
print('test begin')
async_client = Asyncio_Client() # 创建客户端
session = aiohttp.ClientSession()
for i in range(10): # 执行10次
self.all_time=[]
self.total_num=0
self.total_payload=0
self.image = 'xxxx.jpg' # 设置测试nayizhang
print('测试图片:', self.image)
begin = time.time()
async_client.set_task(self.task_func1,self.num,session) # 设置并发任务
async_client.run() # 执行任务
end=time.time()
self.all_time.sort(reverse=True)
print(self.all_time)
print('并发数量(个):',self.total_num)
print('总耗时(s):',end-begin)
print('最大时延(s):',self.all_time[0])
print('最小时延(s):', self.all_time[len(self.all_time)-1])
print('top-90%时延(s):', self.all_time[int(len(self.all_time)*0.1)])
print('平均耗时(s/个):',sum(self.all_time)/self.total_num)
print('支持并发率(个/s):',self.total_num/(end-begin))
print('总负载(MB):',self.total_payload)
print('吞吐率(MB/S):',self.total_payload/(end-begin)) # 吞吐率受上行下行带宽,服务器带宽,服务器算法性能诸多影响
time.sleep(3)
session.close()
print('test finish')
aiohttp服务器mvc(静态网页,模板,数据库,log)
aiohttp之添加静态资源路径
所谓静态资源,是指图片、js、css等文件。
以一个小项目来说明,下面是项目的目录结构:
.
├── static
│ ├── css
│ │ ├── base.css
│ │ ├── bootstrap.min.css
│ │ └── font-awesome.min.css
│ ├── font
│ │ ├── FontAwesome.otf
│ │ ├── fontawesome-webfont.eot
│ │ ├── fontawesome-webfont.svg
│ │ ├── fontawesome-webfont.ttf
│ │ └── fontawesome-webfont.woff
│ └── index.html
└── proxy_server.py
在proxy_server.py给2个静态文件目录static/css和static/font添加路由:
app.router.add_static('/css/',
path='static/css',
name='css')
app.router.add_static('/font/',
path='static/font',
name='font')
必需的2个参数:
prefix:是静态文件的url的前缀,以/开始,在浏览器地址栏上显示在网站host之后,也用于index.html静态页面进行引用
path:静态文件目录的路径,可以是相对路径,上面代码使用的static/css就是相对路径——相对于proxy_server.py所在路径。
加载的是index.html,下面是它引用静态资源的代码:
添加font的路径是因为/font-awesome.min.css需要使用:
如果修改前缀:
app.router.add_static('/css2017/',
path='static/css',
name='css')
虽然目录本身还是css,但通过add_static已经将它视为了css2017,在文件和浏览器中要想链接到css下的文件,必须使用css2017/xx.css来链接。
此外,如果加上show_index=True,就可以显示静态资源的目录索引了——默认是禁止访问的:
app.router.add_static('/css2017/',
path='static/css',
name='css',
show_index=True)
多进程+asyncio
由于python本身只能单线程,所以所谓的线程是通过线程锁实现的。现在必须要通过多进程实现更多的并发。
现在demo实现了使用多进程,每个进程都有一个asyncio
import asyncio
import threading
import multiprocessing
from multiprocessing import Queue ,Pool,Process
#import aiohttp
import os
async def hello(name):
print('hello {} {}**********{}'.format(name,os.getpid(),threading.current_thread()))
#await asyncio.sleep(int(name))
await asyncio.sleep(1)
print('end:{} {}'.format(name,os.getpid()))
def process_start(*namelist):
tasks=[]
loop=asyncio.get_event_loop()
for name in namelist:
tasks.append(asyncio.ensure_future(hello(name)))
loop.run_until_complete(asyncio.wait(tasks))
def task_start(namelist):
i=0
lst=[]
flag=10
while namelist:
i+=1
l=namelist.pop()
lst.append(l)
if i==flag:
p=Process(target=process_start,args=lst)
p.start()
#p.join()
lst=[]
i=0
if namelist!=[]:
p=Process(target=process_start,args=lst)
p.start()
#p.join()
if __name__=='__main__':
# 测试使用多个进程来实现函数
namelist=list('0123456789'*10)
print(namelist)
task_start(namelist)
# 测试使用一个异步io来实现全部函数
# loop=asyncio.get_event_loop()
# tasks=[]
# namelist=list('0123456789'*10)
# for i in namelist:
# tasks.append(asyncio.ensure_future(hello(i)))
# loop.run_until_complete(asyncio.wait(tasks))
aiohttp 高并发web服务_python后台——asyncio,aiohttp入门教程,多进程+asyncio相关推荐
- aiohttp 高并发web服务_【Python入门】50.异步IO之 asyncio实现异步操作
摘要:如何通过asyncio实现异步IO:用aiohttp模块编写支持多用户高并发的服务器. *写在前面:为了更好的学习python,博主记录下自己的学习路程.本学习笔记基于廖雪峰的Python教程, ...
- aiohttp 高并发web服务_能看懂的aiohttp
asyncio可以实现单线程并发IO操作.如果仅用在客户端,发挥的威力不大.如果把asyncio用在服务器端,例如Web服务器,由于HTTP连接就是IO操作,因此可以用单线程+coroutine实现多 ...
- aiohttp 高并发web服务_UE4 unity3d连接高并发http服务器
1.HTTP服务器使用c++语言开发,vs2019编辑器.支持windows和linux.服务器主要用于游戏服务器后台,以及APP 应用后台增.删.查.改数据. 2.服务器使用多线程技术,以及使用线程 ...
- 高并发Web服务的演变:节约系统内存和CPU
一.越来越多的并发连接数 现在的Web系统面对的并发连接数在近几年呈现指数增长,高并发成为了一种常态,给Web系统带来不小的挑战.以最简单粗暴的方式解决,就是增加Web系统的机器和升级硬件配置.虽然现 ...
- pyTorch模型部署--高并发web服务c++移动端ncnn
文章目录 1 综述 2 以python web服务的形式进行部署 3 c++调用方式进行部署 3.1 torchscript 3.1.1 模型转换 3.1.1.1 torch.jit.trace 3. ...
- android 高并发服务端,GitHub - android-coco/chat: 支持10万人同时在线 Go语言打造高并发web即时聊天(IM)应用...
IM 支持10万人同时在线 Go语言打造高并发web即时聊天(IM)应用 部署前准备 配置文件 config/config.yml 样例: # 服务端监听配置 service: port: :8181 ...
- PHP_MVC框架开发,nosql学习,mysql优化,以及高并发web架构处理
文章目录 定义入口文件 完成自动加载 路由类 .htaccess explode trim array_slice 实现 控制器 加载控制器 数据库 初始化连接 数据查询函数 视图 extract 其 ...
- 轻松实现基于Heartbeat的高可用web服务集群
高可用集群就是为了保证某项服务能够时时在线,我们可以通过几个9来衡量一个高可用集群提供服务的稳定性,例如5个9的高可用集群必须保证服务一年在线的时间占99.999%,也就是说一年的时间中仅允许服务电线 ...
- nginx简介(轻量级开源高并发web服务器:大陆使用者百度、京东、新浪、网易、腾讯、淘宝等)(并发量5w)(一般网站apache够用了,而且稳定)...
nginx简介(轻量级开源高并发web服务器:大陆使用者百度.京东.新浪.网易.腾讯.淘宝等)(并发量5w)(一般网站apache够用了,而且稳定) 一.总结 1.在连接高并发的情况下,Nginx是A ...
最新文章
- 余承东安卓鸿蒙,鸿蒙亮点多,网友嗨了!华为P50还没定,余承东:最强大操作系统...
- springMVC保存数据到mysql数据库中文乱码问题解决方法
- 比tween更有效的tween包.
- 评测通知 | 2022年全国知识图谱与语义计算大会评测任务发布
- 设计模式---观察者模式介绍与理解
- tortoiseGit的SHH秘钥设置
- Win10系统安装教程
- Mac数据库可视化工具,workbench完全替代Navicat和PowerDesigner,Oracle官方免费可视化工具workbench!!
- 【高速PCB电路设计】8.DDR模块设计实战
- Network 之五 TCP/IP 协议族、工作流程、(TCP、IP、MAC、VLAN)常用协议格式详解
- 01_配置yum源-银河麒麟V10(Kylin Linux Advanced Server V10 (Tercel))操作系统
- 计算机怎么放映文档,如何从Apple TV上的计算机播放视频文件
- A Jupyter widget could not be displayed because the widget state could not be found.
- Android 快递接口
- 淘宝自动回复机器人配置手册——目前2018年淘宝主流自动回复软件一览
- 【系统分析师】操作系统
- 【git 整理提交】git rebase -i 命令详解
- 一根木棒折两次形成一个三角形的概率是多少
- 2021年江苏高考成绩排名查询,2021年江苏高考一分一段表查询排名方法 成绩排名位次什么时候公布...
- 如果你只知道开心网,那你就OUT了。。。。。