aiomysql、asyncpg、aioredis

Python 目前已经进化到了 3.8 版本，对操作数据库也提供了相应的异步支持。当我们做一个 Web 服务时，性能的瓶颈绝大部分都在数据库上，如果一个请求从数据库中读数据的时候能够自动切换、去处理其它请求的话，是不是就能提高并发量了呢。

下面我们来看看如何使用 Python 异步操作 MySQL、PostgreSQL 以及 Redis，以上几个可以说是最常用的数据库了。至于 SQLServer、Oracle，本人没有找到相应的异步驱动，有兴趣可以自己去探索一下。

而操作数据库无非就是增删改查，下面我们来看看如何异步实现它们。

异步操作 MySQL

下面先来看看如何查询记录。

import asyncio
import aiomysql.sa as aio_sa


async def main():
    # 创建一个异步引擎
    engine = await aio_sa.create_engine(host="xx.xxx.xx.xxx",
                                        port=3306,
                                        user="root",
                                        password="root",
                                        db="_hanser",
                                        connect_timeout=10)

    # 通过 engine.acquire() 获取一个连接
    async with engine.acquire() as conn:
        # 异步执行, 返回一个  对象
        result = await conn.execute("SELECT * FROM girl")
        # 通过 await result.fetchone() 可以获取满足条件的第一条记录, 一个  对象
        data = await result.fetchone()

        # 可以将  对象想象成一个字典
        print(data.keys())  # KeysView((1, '古明地觉', 16, '地灵殿'))
        print(list(data.keys()))  # ['id', 'name', 'age', 'place']

        print(data.values())  # ValuesView((1, '古明地觉', 16, '地灵殿'))
        print(list(data.values()))  # [1, '古明地觉', 16, '地灵殿']

        print(data.items())  # ItemsView((1, '古明地觉', 16, '地灵殿'))
        print(list(data.items()))  # [('id', 1), ('name', '古明地觉'), ('age', 16), ('place', '地灵殿')]

        # 直接转成字典也是可以的
        print(dict(data))  # {'id': 1, 'name': '古明地觉', 'age': 16, 'place': '地灵殿'}
    
    # 最后别忘记关闭引擎, 当然你在创建引擎的时候也可以通过 async with aio_sa.create_engine 的方式创建
    # async with 语句结束后会自动执行下面两行代码
    engine.close()
    await engine.wait_closed()
    

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

怎么样，是不是很简单呢，和同步库的操作方式其实是类似的。但是很明显，我们在获取记录的时候不会只获取一条，而是会获取多条，获取多条的话使用 await result.fetchall() 即可。

import asyncio
from pprint import pprint
import aiomysql.sa as aio_sa


async def main():
    # 通过异步上下文管理器的方式创建, 会自动帮我们关闭引擎
    async with aio_sa.create_engine(host="xx.xxx.xx.xxx",
                                    port=3306,
                                    user="root",
                                    password="root",
                                    db="_hanser",
                                    connect_timeout=10) as engine:
        async with engine.acquire() as conn:
            result = await conn.execute("SELECT * FROM girl")
            # 此时的 data 是一个列表, 列表里面是  对象
            data = await result.fetchall()
            # 将里面的元素转成字典
            pprint(list(map(dict, data)))
            """
            [{'age': 16, 'id': 1, 'name': '古明地觉', 'place': '地灵殿'},
             {'age': 16, 'id': 2, 'name': '雾雨魔理沙', 'place': '魔法森林'},
             {'age': 400, 'id': 3, 'name': '芙兰朵露', 'place': '红魔馆'}]
            """


loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

除了 fetchone、fetchall 之外，还有一个 fetchmany，可以获取指定记录的条数。

import asyncio
from pprint import pprint
import aiomysql.sa as aio_sa


async def main():
    # 通过异步上下文管理器的方式创建, 会自动帮我们关闭引擎
    async with aio_sa.create_engine(host="xx.xxx.xx.xxx",
                                    port=3306,
                                    user="root",
                                    password="root",
                                    db="_hanser",
                                    connect_timeout=10) as engine:
        async with engine.acquire() as conn:
            result = await conn.execute("SELECT * FROM girl")
            # 默认是获取一条, 得到的仍然是一个列表
            data = await result.fetchmany(2)
            pprint(list(map(dict, data)))
            """
            [{'age': 16, 'id': 1, 'name': '古明地觉', 'place': '地灵殿'},
             {'age': 16, 'id': 2, 'name': '雾雨魔理沙', 'place': '魔法森林'}]
            """


loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

以上就是通过 aiomysql 查询数据库中的记录，没什么难度。但是值得一提的是，await conn.execute 里面除了可以传递一个原生的 SQL 语句之外，我们还可以借助 SQLAlchemy。

import asyncio
from pprint import pprint
import aiomysql.sa as aio_sa
from sqlalchemy.sql.selectable import Select
from sqlalchemy import text


async def main():
    async with aio_sa.create_engine(host="xx.xxx.xx.xxx",
                                    port=3306,
                                    user="root",
                                    password="root",
                                    db="_hanser",
                                    connect_timeout=10) as engine:
        async with engine.acquire() as conn:
            sql = Select([text("id, name, place")], whereclause=text("id != 1"), from_obj=text("girl"))
            result = await conn.execute(sql)
            data = await result.fetchall()
            pprint(list(map(dict, data)))
            """
            [{'id': 2, 'name': '雾雨魔理沙', 'place': '魔法森林'},
             {'id': 3, 'name': '芙兰朵露', 'place': '红魔馆'}]
            """


loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

添加记录

修改记录和添加记录是类似的，我们来看一下。

import asyncio
from pprint import pprint
import aiomysql.sa as aio_sa
from sqlalchemy import Table, MetaData, create_engine, text


async def main():
    async with aio_sa.create_engine(host="xx.xx.xx.xxx",
                                    port=3306,
                                    user="root",
                                    password="root",
                                    db="_hanser",
                                    connect_timeout=10) as engine:
        async with engine.acquire() as conn:
            s_engine = create_engine("mysql+pymysql://root:root@xx.xx.xx.xxx:3306/_hanser")
            tbl = Table("girl", MetaData(bind=s_engine), autoload=True)
            update_sql = tbl.update().where(text("name = '古明地觉'")).values({"place": "东方地灵殿"})

            # 同样需要开启一个事务
            async with conn.begin():
                result = await conn.execute(update_sql)
                print(result.lastrowid)  # 0
                print(result.rowcount)   # 1
		
        # 查询结果
        async with engine.acquire() as conn:
            data = await (await conn.execute("select * from girl where name = '古明地觉'")).fetchall()
            data = list(map(dict, data))
            pprint(data)
            """
            [{'age': 16, 'id': 1, 'name': '古明地觉', 'place': '东方地灵殿'}]
            """


loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

可以看到，记录被成功的修改了。

删除记录

异步操作 PostgreSQL 的话，我们有两个选择，一个是 asyncpg 库，另一个是 aiopg 库。

asyncpg 是自己实现了一套连接驱动，而 aiopg 则是对 psycopg2 进行了封装，个人更推荐 asyncpg，性能和活跃度都比 aiopg 要好。

下面来看看如何使用 asyncpg，首先是安装，直接 pip install asyncpg 即可。

查询记录

然后是添加记录，我们看看如何往库里面添加数据。

import asyncio
from pprint import pprint
import asyncpg
from sqlalchemy.sql.selectable import Select
from sqlalchemy import text


async def main():
    conn = await asyncpg.connect("postgres://postgres:zgghyys123@localhost:5432/postgres")
    # 执行 insert 语句我们可以使用 execute
    row = await conn.execute("insert into girl(name, age, place) values ($1, $2, $3)",
                             '十六夜咲夜', 17, '红魔馆')
    pprint(row)  # INSERT 0 1
    pprint(type(row))  # 

    await conn.close()


if __name__ == '__main__':
    asyncio.run(main())

通过 execute 可以插入单条记录，同时返回相关信息，但是说实话这个信息没什么太大用。除了 execute 之外，还有 executemany，用来执行多条插入语句。

import asyncio
import asyncpg

async def main():
    conn = await asyncpg.connect("postgres://postgres:zgghyys123@localhost:5432/postgres")
    # executemany：第一条参数是一个模板，第二条命令是包含多个元组的列表
    # 执行多条记录的话，返回的结果为 None
    rows = await conn.executemany("insert into girl(name, age, place) values ($1, $2, $3)",
                                  [('十六夜咲夜', 17, '红魔馆'), ('琪露诺', 60, '雾之湖')])
    print(rows)  # None

    # 关闭连接
    await conn.close()


if __name__ == '__main__':
    asyncio.run(main())

注意：如果是执行大量 insert 语句的话，那么 executemany 要比 execute 快很多，但是 executemany 不具备事务功能。

await conn.executemany("insert into girl(id, name) values($1, $2)",
                       [(7, "八意永琳"), (7, "八意永琳"), (8, "zun")])

我们表中的 id 是主键，不可以重复，这里插入三条记录。显然第二条记录中的 id 和第一条重复了，执行的时候会报错。但是第一条 (7, "八意永琳") 是进入到数据库了的，尽管第二条记录在插入的时候执行失败。当然第二条执行失败，那么第二条之后的也就无法执行了，这一点务必注意。那如果我们想要实现事务的话该怎么做呢？

import asyncio
import asyncpg


async def main():
    conn = await asyncpg.connect("postgres://postgres:zgghyys123@localhost:5432/postgres")

    # 调用 conn.transaction() 会开启一个事务，当然这里建议使用异步上下文管理
    async with conn.transaction():
        await conn.executemany("insert into girl(id, name) values($1, $2)",
                               [(999, "太田顺也"), (999, "zun")]
                               )

    # 关闭连接
    await conn.close()


if __name__ == '__main__':
    asyncio.run(main())

两条记录主键重复，最终这两条记录都没进入到数据库中。因此插入记录的话，个人建议直接开启一个事务，然后通过 executemany 即可。

因此 fetchrow、fetch 是专门针对 select 语句，而 execute 和 executemant 是针对 select 语句之外的其它语句。

但是问题来了，无论是 execute 还是 executemany，它们都没有返回插入记录之后的具体信息。比如：插入一条记录之后，我们希望返回插入记录的自增 id，这个时候该怎么做呢？答案是依旧使用 fetch。

import asyncio
import asyncpg

async def main():
    conn = await asyncpg.connect("postgres://postgres:zgghyys123@localhost:5432/postgres")
    async with conn.transaction():
        # 对于 fetch 而言, 我们不能像 executemany 那样, 传递一个包含元组的列表
        # 因此想插入多条记录的话, 只能先拼接好 SQL 语句, 并且想返回对应的自增 id 的话, 需要在语句结尾加上 returning id
        # 当然这个是数据库的语法, 否则得到的就是一个 None
        rows = await conn.fetch("insert into girl(name, age, place) values ('太田顺也', 43, 'japan'), ('zun', 43, 'japan') returning id")

    print(rows)  # [, ]
    # 关闭连接
    await conn.close()


if __name__ == '__main__':
    asyncio.run(main())

因此如果只是关注记录是否插入成功，那么推荐使用 executemany 加事务的方式，这样我们只需要传递一个包含元组的列表即可。如果还需要获取插入记录的自增 id，那么需要使用 fetch 加事务的方式（如果是多条 SQL 的话），但是需要事先将 SQL 语句拼接好才行，并且还要使用 PostgreSQL 提供的 returning 字句，否则是不会返回信息的（只能得到一个 None）。

至于 fetchrow 和 execute，它们只针对于单条，因此建议直接使用 fetch 和 executemany 即可。

修改记录

没什么可说的了，直接看代码吧。

import asyncio
import asyncpg


async def main():
    conn = await asyncpg.connect("postgres://postgres:zgghyys123@localhost:5432/postgres")
    async with conn.transaction():
        row = await conn.execute("delete from girl where name in ($1, $2)", "BAKA9", "古明地恋")
        print(row)  # DELETE 2

        rows = await conn.executemany("delete from girl where id = $1",
                                      # 嵌套元组的列表, 即使是一个值也要写成元组
                                      # 会将每一个元组里面的值和占位符进行逐一匹配
                                      [(2,), (3,)])
        print(rows)  # None

    # 关闭连接
    await conn.close()


if __name__ == '__main__':
    asyncio.run(main())

同理如果想获取返回的自增 id的话，也是可以的，方面和上面一样。

import asyncio
import asyncpg


async def main():
    conn = await asyncpg.connect("postgres://postgres:zgghyys123@localhost:5432/postgres")
    async with conn.transaction():
        # 虽然不支持像 executemany 那样传递一个列表, 但是下面这种方式还是可以的
        # 会将元素和占位符逐一替换
        rows = await conn.fetch("delete from girl where name in ($1, $2) returning id", "太田顺也", "zun")
        print(rows)  # [, , , , , ]
    # 关闭连接
    await conn.close()


if __name__ == '__main__':
    asyncio.run(main())

以上就是常见的增删改查操作，虽然没有 SQLAlchemy 那么强大，但是也足够用，当然 SQLAlchemy 内部提供的一些属性也是通过执行 SQL 语句获取的，然后封装成一个属性给你。如果需要的话，我们也可以手动实现执行 SQL 来获取。

连接池

我们之所以使用 asyncpg，无非是为了效率，那么 asyncpg 和传统的 psycopg2 相比，在效率上究竟有多少差距呢？我们来测试一下。

SELECT count(*) FROM interface; -- 8459729

我数据库中有一张表叫做 interface，是之前工作的时候从对方接口获取的，我们就用它来进行测试吧。

先使用同步版本的来访问，看看用多长时间。

import time
from sqlalchemy import create_engine, text

engine = create_engine("postgres://postgres:zgghyys123@localhost:5432/postgres")

with engine.begin() as conn:
    start = time.perf_counter()
    for _ in range(20):
        res = conn.execute(
            text('select * from interface where "ProwlerPersonID" = :arg'),
            {"arg": "c9fcbed8-fa47-481a-9d73-5fd1dd344f19"})
        print(f"满足条件的记录有：{len(res.fetchall())}条")
    end = time.perf_counter()
    print("总耗时:", end - start)  # 50.0419027
"""
满足条件的记录有：228186条
满足条件的记录有：228186条
满足条件的记录有：228186条
满足条件的记录有：228186条
满足条件的记录有：228186条
满足条件的记录有：228186条
满足条件的记录有：228186条
满足条件的记录有：228186条
满足条件的记录有：228186条
满足条件的记录有：228186条
满足条件的记录有：228186条
满足条件的记录有：228186条
满足条件的记录有：228186条
满足条件的记录有：228186条
满足条件的记录有：228186条
满足条件的记录有：228186条
满足条件的记录有：228186条
满足条件的记录有：228186条
满足条件的记录有：228186条
满足条件的记录有：228186条
总耗时: 50.0419027
"""

我们看到记录有 20 多万条，所以就不打印记录了，因为执行的是同一条 SQL，所以结果是一样的，然后我们看到花了 50 秒钟。

再来看看使用异步版本用多长时间。

import time
import asyncio
import asyncpg


async def run_sql(conn, query_list):
    result = []
    for query in query_list:
        result.append(await conn.fetch(*query))
    await conn.close()
    return [f"满足条件的记录有：{len(_)}条" for _ in result]


async def main():
    async with asyncpg.create_pool("postgres://postgres:zgghyys123@localhost:5432/postgres") as pool:
        query_list = [('select * from interface where "ProwlerPersonID" = $1',
                       "c9fcbed8-fa47-481a-9d73-5fd1dd344f19")
                      for _ in range(20)]

        # 我们要创建5个连接异步访问
        count = len(query_list) // 5
        # 将 20 个任务分成 5 份
        query_list = [query_list[c * 4: (c + 1) * 4] for c in range(count + 1)]

        tasks = []
        for q in query_list:
            conn = await pool.acquire()
            tasks.append(run_sql(conn, q))
        results = await asyncio.gather(*tasks)
        return results

if __name__ == '__main__':
    start = time.perf_counter()
    loop = asyncio.get_event_loop()
    results = loop.run_until_complete(main())
    end = time.perf_counter()
    for result in results:
        for _ in result:
            print(_)
    print("总耗时:", end - start)
    """
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    满足条件的记录有：228186条
    总耗时: 9.8730488
    """

我们看到花了将近十秒钟，正好是同步版本的五分之一，因为我们使用了连接池中的五个连接。

注意：如果是 SQLAlchemy，即便你给每一个 SQL 开一个连接，如果不使用多线程，只是同步访问的话，那么耗时还是 50 秒左右，因为它是同步访问的。

而使用异步的模式访问的话，每个连接都可以进行异步访问，那么我们创建的 5 个连接中，如果一个连接阻塞了，会切换到其它的连接去执行。所以耗时为五分之一，不过这里可能有人会觉得困惑，不知道我上面的代码做了些什么，这里来解释一下。

async def f1():
    for _ in [耗时协程1, 耗时协程2, 耗时协程3]:
        await _
        
def f2():
    for _ in [耗时函数1, 耗时函数2, 耗时函数3]:
        _()

我们上面的两段代码，如果函数和协程里面的代码做了相同的事情的话，那么这两个 for 循环耗时基本是一致的。首先函数无需解释，关键是协程为什么会这样。

我们 await 协程，会等待这个协程完成，对于一个 for 循环来说，不可能说当前循环还没执行完毕就去执行下一层循环。所以无论是协程还是普通的函数，都要经历三轮循环，所以它们的耗时是基本一致的。

如果想解决这个问题，那么就要使用 asyncio.gather(*协程列表)，如果是 await asyncio.gather(*[耗时协程1, 耗时协程2, 耗时协程3])，那么时间相比普通函数来说，就会有明显的缩短。因为此时这三个协程是同时发出的。

我们上面使用 asyncio.gather 的目的就在于此，但是问题是我们为什么要创建多个连接、为什么要把 20 个任务分成 5 份呢。

首先对于数据库来讲，一个连接只能同时执行一个 SQL，如果你使用多线程、但是每个线程用的却是同一个连接的话，你会发现耗时和原来基本没区别。虽然线程阻塞会自动切换，但是你使用的连接已经被别人用了，所以请求同时只能发一个。如果是 asyncpg 的话，一个连接同时执行多了 SQL，那么会直接报错。

import asyncio
import asyncpg


async def main():
    async with asyncpg.create_pool("postgres://postgres:zgghyys123@localhost:5432/postgres") as pool:
        async with pool.acquire() as conn:
            # 将20个请求同时发出，理论上可以，但是对于数据库来说是不同意的
            # 因为我们这里的conn都是同一个连接
            query_list = [conn.fetch('select * from interface where "ProwlerPersonID" = $1',
                                     "c9fcbed8-fa47-481a-9d73-5fd1dd344f19")
                          for _ in range(20)]
            await asyncio.gather(*query_list)


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
"""
  File "asyncpg\protocol\protocol.pyx", line 301, in query
  File "asyncpg\protocol\protocol.pyx", line 664, in asyncpg.protocol.protocol.BaseProtocol._check_state
asyncpg.exceptions._base.InterfaceError: cannot perform operation: another operation is in progress
"""

阅读报错信息，提示我们：无法执行操作，另一个操作已经在执行了。

说明我们的20个请求都是同一个连接发出的，第一个连接在执行的时候，第二个连接理论上应该处于阻塞状态的，但是这里直接报错了。但不管怎么样，都不是我们想要的。

所以我们只能使用上面说的那样，使用 for 循环里面写 await 协程 的方式，但是这样和同步又没什么区别。因此我们创建 5 个连接，对 5 个连接使用 asyncio.gather，也就是让这五个连接同时执行。尽管每个连接内部执行的逻辑是同步的，但是这 5 个连接整体是异步的，因为它们彼此没有关联，是不同的连接，因此异步耗时为同步的五分之一。

异步操作 Redis

高并发在现如今是一个主流，任何服务都需要考虑到并发量的问题，如果能够很轻松地支持并发的话，那么肯定是非常受欢迎的。而 Go 语言之所以这么火，很大一部分原因就是它在语言层面就支持并发，当然其它语言也不甘落后。于是 Python 在 3.5 的时候也引入了原生协程，而紧接着围绕着相关生态的异步库自然随之诞生。