Python协程数据库读写怎么加速？peewee-async异步MySQL实战

简介

协程的好处，玩Python的应该都不陌生——单线程内就能实现并发，底层依赖事件循环和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)

注意看，同步和异步两种写法并存，并且通过set_allow_sync(False)来禁止同步操作，是个好习惯。

异步读写

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!

这里用create_or_get和update演示了增改操作，配合get查询，基本上覆盖了日常CRUD。

同时使用同步和异步调用

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())

这个例子展示了如何在一个项目里混合同步和异步代码——同步创建、异步查询，最后再同步删除。注意连接数据库时用了connect_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()


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

注意输出是FOO，说明事务回滚生效了——即使更新了BAR，遇到异常后数据依然保持原样。

速度测试

先安装造数据的库：

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

结果很直观：同步读取10万条数据花了3.10秒，异步只用了1.86秒，几乎快了40%。

遇到的坑

异步在某些场景下反而可能更慢——比如单一查询、小数据量时，协程的上下文切换开销会抵消掉并发优势。另外，如果数据库连接池配置不当，也可能出现连接等待。

总结

peewee-async 让异步操作 peewee 变得非常轻松，尤其适合高并发读写的场景。当然，任何技术都有适用边界，用之前最好先做个小测试，看看实际收益。