Python协程数据库读写怎么加速？peewee-async异步MySQL实战
作者：XerCis
还在为数据库读写速度发愁吗？本文带你了解如何用peewee-async库在单线程内通过协程和IO多路复用实现并发,让MySQL查询显著加快,基于aiomysql驱动,从安装、异步读写、事务到速度测试,一应俱全,还贴心总结了异步可能变慢的坑,助你避雷
简介
协程可以在单线程内实现并发，原理是循环和 IO 多路复用，使用协程可以让数据库读取加快
peewee-async 底层基于 aiomysql
本文以 MySQL 为例
安装
pip install peewee-async
不同数据库连接驱动
pip install aiomysql
- PostgreSQL：aiopg
- MySQL：aiomysql
初试
使用本机连接，创建数据库 test
import asyncio
import peewee_async
from peewee import *
database = peewee_async.MySQLDatabase('test', user='root', password='123456', host='127.0.0.1', port=3306)
class TestModel(Model):
text = CharField()
class Meta:
database = database
# 同步
TestModel.create_table(True)
TestModel.create(text='Yo, I can do it sync!')
database.close()
# 异步
objects = peewee_async.Manager(database)
database.set_allow_sync(False)
async def handler():
await objects.create(TestModel, text='Not bad. Watch this, I am async!')
all_objects = await objects.execute(TestModel.select())
for obj in all_objects:
print(obj.text)
loop = asyncio.get_event_loop()
loop.run_until_complete(handler())
loop.close()
# 以同步方式删除表
with objects.allow_sync():
TestModel.drop_table(True)
异步读写
import asyncio
from peewee import Model, CharField, TextField
from peewee_async import MySQLDatabase, Manager
loop = asyncio.new_event_loop()
database = MySQLDatabase('test', user='root', password='123456', host='127.0.0.1', port=3306)
objects = Manager(database, loop=loop)
class PageBlock(Model):
key = CharField(max_length=40, unique=True)
text = TextField(default='')
class Meta:
database = database
PageBlock.create_table(True)
objects.database.allow_sync = False
async def my_async_func():
await objects.create_or_get(PageBlock, key='title', text='Peewee is AWESOME with async!')
title = await objects.get(PageBlock, key='title')
print('Was:', title.text)
title.text = 'Peewee is SUPER awesome with async!'
await objects.update(title)
print('New:', title.text)
loop.run_until_complete(my_async_func())
loop.close()
# Was: Peewee is AWESOME with async!
# New: Peewee is SUPER awesome with async!
同时使用同步和异步调用
import asyncio
import peewee_async
from peewee import Model, CharField
database = peewee_async.MySQLDatabase('test', user='root', password='123456', host='127.0.0.1', port=3306)
loop = asyncio.get_event_loop()
class TestModel(Model):
text = CharField()
class Meta:
database = database
# 同步
TestModel.create_table(True)
database.close()
@asyncio.coroutine
def my_handler():
obj1 = TestModel.create(text='Yo, I can do it sync!')
obj2 = yield from peewee_async.create_object(TestModel, text='Not bad. Watch this, I am async!')
all_objects = yield from peewee_async.execute(TestModel.select())
for obj in all_objects:
print(obj.text)
obj1.delete_instance()
yield from peewee_async.delete_object(obj2)
loop.run_until_complete(database.connect_async(loop=loop))
loop.run_until_complete(my_handler())
事务
import asyncio
import peewee_async
from peewee import Model, CharField
database = peewee_async.MySQLDatabase('test', user='root', password='123456', host='127.0.0.1', port=3306)
loop = asyncio.get_event_loop()
class TestModel(Model):
text = CharField()
class Meta:
database = database
# 同步
TestModel.create_table(True)
database.close()
async def test():
obj = await peewee_async.create_object(TestModel, text='FOO')
obj_id = obj.id
try:
async with database.atomic_async():
obj.text = 'BAR'
await peewee_async.update_object(obj)
raise Exception('Fake error')
except:
res = await peewee_async.get_object(TestModel, TestModel.id == obj_id)
print(res.text)
loop.run_until_complete(test())
# FOO
速度测试
安装
pip install faker
代码
import time
import asyncio
import peewee_async
from peewee import *
from faker import Faker
database = peewee_async.MySQLDatabase('test', user='root', password='123456', host='127.0.0.1', port=3306)
class Student(Model):
id = PrimaryKeyField()
name = CharField()
birthday = DateField()
chinese = IntegerField()
math = IntegerField()
english = IntegerField()
class Meta:
database = database
Student.create_table(True)
# 数据准备
faker = Faker('zh_CN')
students = []
batch_size = 100
for _ in range(100000): # 10w条数据
student = Student(name=faker.name(), birthday=faker.date(), chinese=faker.random_int(min=0, max=100),
math=faker.random_int(min=0, max=100), english=faker.random_int(min=0, max=100))
students.append(student)
if len(students) >= batch_size:
Student.bulk_create(students, batch_size=batch_size)
students.clear()
print('数据准备完成')
def sync_read():
"""同步读取"""
begin = time.time()
students1 = Student.select().where(Student.birthday >= '2000-01-01')
students2 = Student.select().where(Student.chinese >= 70)
students3 = Student.select().where(Student.math >= 80)
students4 = Student.select().where(Student.english >= 90)
lengths = len(students1), len(students2), len(students3), len(students4)
print('sync_read: {:.2f}s'.format(time.time() - begin))
return lengths
async def async_read():
"""异步读取"""
begin = time.time()
students1 = await manager.execute(Student.select().where(Student.birthday >= '2000-01-01'))
students2 = await manager.execute(Student.select().where(Student.chinese >= 70))
students3 = await manager.execute(Student.select().where(Student.math >= 80))
students4 = await manager.execute(Student.select().where(Student.english >= 90))
lengths = len(students1), len(students2), len(students3), len(students4)
print('async_read: {:.2f}s'.format(time.time() - begin))
return lengths
if __name__ == '__main__':
sync_read()
manager = peewee_async.Manager(database)
database.set_allow_sync(False)
loop = asyncio.get_event_loop()
loop.run_until_complete(async_read())
# 数据准备完成
# sync_read: 3.10s
# async_read: 1.86s
遇到的坑
异步某些情况可能会慢些
总结
以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。