MySQL异步驱动aiomysql

2022/1/28 19:34:21

本文主要是介绍MySQL异步驱动aiomysql,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!

 

  本文介绍异步MySQL异步驱动aiomysql的使用

  1,安装异步模块

  如果没有模块则先使用pip安装模块

?
1 2 pip3 install asyncio pip3 install aiomysql

  2,创建MySQL数据库连接池

  和同步方式不一样的是使用异步不能直接创建数据库连接conn,需要先创建一个数据库连接池对象__pool通过这个数据库连接池对象来创建数据库连接

  数据库配置信息和介绍pymysql同步使用的数据库是一样的

?
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 import asyncio,aiomysql,time # 数据库配置dict db_config = {     'host': 'localhost',     'user': 'www-data',     'password': 'www-data',     'db': 'awesome' }   # 创建数据库连接池协程函数 async def create_pool(**kw):     global __pool     __pool = await aiomysql.create_pool(         host=kw.get('host', 'localhost'),         port=kw.get('port', 3306),         user=kw['user'],         password=kw['password'],         db=kw['db']     )   loop=asyncio.get_event_loop() loop.run_until_complete(create_pool(**db_config)) # 在事件循环中运行了协程函数则生成了全局变量__pool是一个连接池对象 <aiomysql.pool.Pool object at 0x00000244AD1724C8> print(__pool) # <aiomysql.pool.Pool object at 0x00000244AD1724C8>

  3,创建执行sql语句的协程函数

  因为是异步模块,只能在事件循环中通过await关键字调用,使用需要创建执行sql语句的协程函数

  在协程函数内使用全局上一步创建的连接池对象来创建连接conn和浮标对象cur,通过浮标对象来执行sql语句,执行方法和pymysql模块的执行方法是一样的

?
1 2 3 4 5 6 cursor.execute(sql,args) sql # 需要执行的sql语句例如'select * from table_name' args # 替换sql语句的格式化字符串,即sql语句可以使用%s代表一个字符串,然后在args中使用对应的变量或参数替换,args为一个list或元组,即是一个有序的序列需要和sql中的%s一一对应 # 例如sql='select * from table_name where id=%s'  args=['12345'] # 相当于使用args中的参数替换sql中的%s # select * from table_name where id='12345'

  下面分别创建两个协程函数select execute一个用来执行搜索操作,一个用来执行insert,update,delete等修改操作

?
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 # 执行select函数 async def select(sql,args,size=None):     with await __pool as conn:         cur = await conn.cursor(aiomysql.DictCursor)         await cur.execute(sql.replace('?','?s'),args or ())         if size:             rs = await cur.fetchmany(size)         else:             rs = await cur.fetchall()         await cur.close()         return rs     # 执行insert update delete函数 async def execute(sql,args):     with await __pool as conn:         try:             cur = await conn.cursor()             await cur.execute(sql.replace('?','%s'),args)             affetced = cur.rowcount             await conn.commit()             await cur.close()         except BaseException as e:             raise         return affetced

  4,实践执行sql语句

  实践执行sql语句前我们首先在本机创建一个数据库和对应的表用于测试

  数据库对应的主机,用户名,密码,库名,表名如下

?
1 2 3 4 5 host: localhost user: www-data password: www-data db:awesome table_name: users

  创建表名的sql语句如下,需要在数据库中创建好对应的表

?
1 2 3 4 5 6 7 8 9 10 11 12 CREATE TABLE `users` (   `id` varchar(50) NOT NULL,   `email` varchar(50) NOT NULL,   `passwd` varchar(50) NOT NULL,   `admin` tinyint(1) NOT NULL,   `name` varchar(50) NOT NULL,   `image` varchar(500) NOT NULL,   `created_at` double NOT NULL,   PRIMARY KEY (`id`),   UNIQUE KEY `idx_email` (`email`),   KEY `idx_created_at` (`created_at`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8

  创建好的表对应的结构如下

?
1 2 3 4 5 6 7 8 9 10 11 12 13 mysql> desc users; +------------+--------------+------+-----+---------+-------+ | Field      | Type         | Null | Key | Default | Extra | +------------+--------------+------+-----+---------+-------+ | id         | varchar(50)  | NO   | PRI | NULL    |       | | email      | varchar(50)  | NO   | UNI | NULL    |       | | passwd     | varchar(50)  | NO   |     | NULL    |       | | admin      | tinyint(1)   | NO   |     | NULL    |       | | name       | varchar(50)  | NO   |     | NULL    |       | | image      | varchar(500) | NO   |     | NULL    |       | | created_at | double       | NO   | MUL | NULL    |       | +------------+--------------+------+-----+---------+-------+ 7 rows in set (2.68 sec)

  ①执行insert操作

?
1 2 3 4 5 6 7 8 # insert start import time sql = 'insert into `users` (`email`, `passwd`, `admin`, `name`, `image`, `created_at`, `id`) values (?, ?, ?, ?, ?, ?, ?)' args = ['test@qq.com','password',1,'test','about:blank',time.time(),'111111'] async def insert():     await execute(sql,args) loop.run_until_complete(insert()) # insert end

  执行方式和pymysql没有区别,不同的是需要在事件循环中使用关键字await调用

  执行完毕在MySQL中查看插入的数据

?
1 2 3 4 5 6 7 mysql> select * from users; +--------+-------------+----------+-------+------+-------------+------------------+ | id     | email       | passwd   | admin | name | image       | created_at       | +--------+-------------+----------+-------+------+-------------+------------------+ | 111111 | test@qq.com | password |     1 | test | about:blank | 1637738541.48629 | +--------+-------------+----------+-------+------+-------------+------------------+ 1 row in set (0.00 sec)

  ②执行update操作

  直接在loop事件循环中执行execute协程函数也可以

?
1 2 3 4 5 6 # update start import time sql = 'update `users` set `email`=?, `passwd`=?, `admin`=?, `name`=?, `image`=?, `created_at`=? where `id`=?' args = ['test2@qq.com','password',1,'test2','about:blank',time.time(),'111111'] loop.run_until_complete(execute(sql,args)) # update end

  执行以后把email和name都修改了

  ③执行delete操作

?
1 2 3 4 5 # delete start sql = 'delete from `users` where `id`=?' args = ['111111'] loop.run_until_complete(execute(sql,args)) # delete end

  同样根据关键字id指定的值删除了这条数据

  ④执行selete操作

  在执行select操作前我们保证数据库里面至少有一条数据

?
1 2 3 4 5 6 # select start sql = 'select * from users' args = [] rs = loop.run_until_complete(select(sql,args)) print(rs) # select end

  这里直接执行搜索的协程函数select根据函数的定义返回的是所有结果的list,元素是查询结果的字典

  输出为

?
1 [{'id': '111111', 'email': 'test@qq.com', 'passwd': 'password', 'admin': 1, 'name': 'test', 'image': 'about:blank', 'created_at': 1637739212.74493}]

  如果结果有多个则使用list的下标取出

  

  补充

  同步模块pymysql和异步模块aiomysql执行速度对比

  假如我们需要往数据库插入20000条数据,我们分别使用同步模式和异步模式

  首先删除数据库所有测试数据

?
1 delete from users;

  同步的代码

  d:/learn-python3/学习脚本/pymysql/use_pymysql.py

?
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 import pymysql db_config = {     'host': 'localhost',     'user': 'www-data',     'password': 'www-data',     'db': 'awesome' } # 创建连接,相当于把字典内的键值对传递 # 相当于执行pymysql.connect(host='localhost',user='www-data',password='www-data',db='awesome') conn = pymysql.connect(**db_config) # 创建游标 cursor = conn.cursor(pymysql.cursors.DictCursor) sql = 'select * from users' args = [] # 执行查询返回结果数量 # 执行查询 rs=cursor.execute(sql,args) # 获取查询结果 # 获取查询的第一条结果,返回一个dict,dict元素是查询对应的键值对 # 如果查询结果有多条则执行一次,游标移动到下一条数据,在执行一次又返回一条数据 # print(cursor.fetchone()) # print(cursor.fetchone()) # print(cursor.fetchall()) # print(cursor.fetchmany()) # {'id': '111111', 'email': 'test@qq.com', 'passwd': 'password', 'admin': 1, 'name': 'test', 'image': 'about:blank', 'created_at': 1637723578.5734} # 获取查询的所有结果,返回一个list,list元素是dict,dict元素是查询对应的键值对 # print(cursor.fetchall()) # [{'id': '111111', 'email': 'test@qq.com', 'passwd': 'password', 'admin': 1, 'name': 'test', 'image': 'about:blank', 'created_at': 1637723578.5734}] # 获取查询的前几条结果,返回一个dict,dict元素是查询对应的键值对 # print(cursor.fetchmany(1)) # [{'id': '111111', 'email': 'test@qq.com', 'passwd': 'password', 'admin': 1, 'name': 'test', 'image': 'about:blank', 'created_at': 1637723578.5734}] # 执行修改操作 import time # # insert start sql = 'insert into `users` (`email`, `passwd`, `admin`, `name`, `image`, `created_at`, `id`) values (?, ?, ?, ?, ?, ?, ?)' args = ['test1@qq.com','password',1,'test','about:blank',time.time(),'1111121'] # 使用replace 把'?'替换成'%s' # rs = cursor.execute(sql.replace('?','%s'),args) # print(cursor.rowcount) # conn.commit() # print(rs) # insert end   # update start # sql = 'update `users` set `email`=?, `passwd`=?, `admin`=?, `name`=?, `image`=?, `created_at`=? where `id`=?' # args = ['test2@qq.com','password',1,'test2','about:blank',time.time(),'111111'] # print(cursor.execute(sql.replace('?','%s'),args)) # conn.commit() # update end   # delete start # sql = 'delete from `users` where `id`=?' # args = ['111111'] # print(cursor.execute(sql.replace('?','%s'),args)) # conn.commit() # delete end     # 写成函数调用,函数内部使用了数据库连接对象conn # 可以先定义成全局变量global def select(sql,args,size=None):           cursor =  conn.cursor(pymysql.cursors.DictCursor)     cursor.execute(sql.replace('?','%s'),args or ())     if size:         rs = cursor.fetchmany(size)     else:         rs = cursor.fetchall()     cursor.close     # logging.info('rows returned: %s' % len(rs))     return rs    def execute(sql,args):            cursor = conn.cursor(pymysql.cursors.DictCursor)     try:         cursor.execute(sql.replace('?','%s'),args)         # rowcount方法把影响函数返回         rs = cursor.rowcount         cursor.close()         conn.commit()     except:         raise     return rs   start_time = time.time() for n in range(20000):     sql = 'insert into `users` (`email`, `passwd`, `admin`, `name`, `image`, `created_at`, `id`) values (?, ?, ?, ?, ?, ?, ?)'     email = 'test%s@qq.com' %n     args = [email,'password',1,'test','about:blank',time.time(),n]     execute(sql,args) end_time = time.time() # 打印开始和结束时间的差 print(end_time - start_time)

  我们使用一个循环20000次往数据库插入数据

  执行,插入数据比较多需要等待一段时间输出

?
1 2 D:\learn-python3\函数式编程>C:/Python37/python.exe d:/learn-python3/学习脚本/pymysql/use_pymysql.py 77.46903562545776

  可以在数据库查询到这20000条数据,而且这个表的字段created_at存储了创建这条数据的时间戳,我们可以看到,id越往后的时间戳越往后,说明数据是同步按顺序一一插入的

  我们按照字段created_at排序查询

 

 

  下面我们删除所有数据使用异步方式插入

  异步的代码如下

  d:/learn-python3/学习脚本/aiomysql/use_aiomysql.py

?
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 import asyncio,aiomysql,time # 数据库配置dict db_config = {     'host': 'localhost',     'user': 'www-data',     'password': 'www-data',     'db': 'awesome' }   # 创建数据库连接池协程函数 async def create_pool(**kw):     global __pool     __pool = await aiomysql.create_pool(         host=kw.get('host', 'localhost'),         port=kw.get('port', 3306),         user=kw['user'],         password=kw['password'],         db=kw['db']     )   loop=asyncio.get_event_loop() loop.run_until_complete(create_pool(**db_config)) # 在事件循环中运行了协程函数则生成了全局变量__pool是一个连接池对象 <aiomysql.pool.Pool object at 0x00000244AD1724C8> print(__pool) # <aiomysql.pool.Pool object at 0x00000244AD1724C8>   # 执行select函数 async def select(sql,args,size=None):     with await __pool as conn:         cur = await conn.cursor(aiomysql.DictCursor)         await cur.execute(sql.replace('?','?s'),args or ())         if size:             rs = await cur.fetchmany(size)         else:             rs = await cur.fetchall()         await cur.close()         return rs     # 执行insert update delete函数 async def execute(sql,args):     with await __pool as conn:         try:             cur = await conn.cursor()             await cur.execute(sql.replace('?','%s'),args)             affetced = cur.rowcount             await conn.commit()             await cur.close()         except BaseException as e:             raise         return affetced   # insert start # import time # sql = 'insert into `users` (`email`, `passwd`, `admin`, `name`, `image`, `created_at`, `id`) values (?, ?, ?, ?, ?, ?, ?)' # args = ['test@qq.com','password',1,'test','about:blank',time.time(),'111111'] # async def insert(): #     await execute(sql,args) # loop.run_until_complete(insert()) # insert end   # update start # import time # sql = 'update `users` set `email`=?, `passwd`=?, `admin`=?, `name`=?, `image`=?, `created_at`=? where `id`=?' # args = ['test2@qq.com','password',1,'test2','about:blank',time.time(),'111111'] # loop.run_until_complete(execute(sql,args))   # update end   # delete start # sql = 'delete from `users` where `id`=?' # args = ['111111'] # loop.run_until_complete(execute(sql,args)) # delete end   # select start # sql = 'select * from users' # args = [] # rs = loop.run_until_complete(select(sql,args)) # print(rs) # select end   async def insert1():      for n in range(10000):         sql = 'insert into `users` (`email`, `passwd`, `admin`, `name`, `image`, `created_at`, `id`) values (?, ?, ?, ?, ?, ?, ?)'         email = 'test%s@qq.com' %n         args = [email,'password',1,'test','about:blank',time.time(),n]         await execute(sql,args)   async def insert2():      for n in range(10001,20001):         sql = 'insert into `users` (`email`, `passwd`, `admin`, `name`, `image`, `created_at`, `id`) values (?, ?, ?, ?, ?, ?, ?)'         email = 'test%s@qq.com' %n         args = [email,'password',1,'test','about:blank',time.time(),n]         await execute(sql,args)   async def main():     # 需要组合成一个事件才会异步执行即在执行insert1的时候同步执行insert2     await asyncio.gather(insert1(),insert2())   start_time = time.time() loop.run_until_complete(main()) end_time = time.time() print(end_time - start_time)

  这里我们定义了两个协程函数,分别用来执行前10000个数据和后10000个数据的插入,在main()把这两个协程函数组合成一个事件循环

  等待一段时间后执行输出如下,忽略这个warning,可以看到执行时间明显比同步时间短

?
1 2 3 d:/learn-python3/学习脚本/aiomysql/use_aiomysql.py:42: DeprecationWarning: with await pool as conn deprecated, useasync with pool.acquire() as conn instead   with await __pool as conn: 39.794615507125854

  我们去数据库查询一下数据也可以看到id从0开始和id从10001开始几乎是同时插入的

 

 

 



这篇关于MySQL异步驱动aiomysql的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!


扫一扫关注最新编程教程