最快速查询sql的方法. asyncmy

# 异步加线程池. from concurrent.futures import ThreadPoolExecutor import asyncio all_tables=["select * from hdrx.bas_source", "select * from hdrx.bas_station", "select * from hdrx.bas_unit","select * from hdrx.source_data_day_his", "select * from hdrx.source_data_day_his", "select * from hdrx.source_data_day_his", "select * from hdrx.source_data_day_his", "select * from hdrx.source_data_day_his",] def runsql(ttt): import pandas as pd pd.read_sql_query import pandas as pd from sqlalchemy import create_engine user = 'root' mysql_pass = 'root' mysql_ip = '127.0.0.1' engine = create_engine( "mysql+pymysql://{}:{}@{}:3306/hdrx".format(user, mysql_pass, mysql_ip)) cmd_one_line_sql = ttt df = pd.read_sql_query(cmd_one_line_sql, con=engine) return df if 1: import time kaishi=time.time() async def mini(): with ThreadPoolExecutor(max_workers=10) as executor: loop = asyncio.get_event_loop() tasks = [ loop.run_in_executor(executor, runsql, table) for table in all_tables ] results = await asyncio.gather(*tasks, return_exceptions=True) return results a=asyncio.run(mini()) # print(a) print("异步加线程池的速度",time.time()-kaishi) print("异步的速度") kaishi=time.time() async def runsql(ttt): import pandas as pd pd.read_sql_query import pandas as pd from sqlalchemy import create_engine user = 'root' mysql_pass = 'root' mysql_ip = '127.0.0.1' engine = create_engine( "mysql+pymysql://{}:{}@{}:3306/hdrx".format(user, mysql_pass, mysql_ip)) cmd_one_line_sql = ttt df = pd.read_sql_query(cmd_one_line_sql, con=engine) return df async def mini(): tasks = [ runsql(table) for table in all_tables ] results = await asyncio.gather(*tasks, return_exceptions=True) return results a=asyncio.run(mini()) # print(a) print("异步的速度",time.time()-kaishi) import asyncio import time import pandas as pd from sqlalchemy.ext.asyncio import create_async_engine from sqlalchemy.sql import text # 待执行的 SQL 列表 all_sql = [ "select * from hdrx.bas_source", "select * from hdrx.bas_station", "select * from hdrx.bas_unit", "select * from hdrx.source_data_day_his", "select * from hdrx.source_data_day_his", "select * from hdrx.source_data_day_his", "select * from hdrx.source_data_day_his", "select * from hdrx.source_data_day_his", ] # 数据库配置 DB_CONFIG = { "user": "root", "password": "root", "host": "127.0.0.1", "port": 3306, "database": "hdrx" } #====================最快速查询sql的方法. async def runsql_async(ttt): """真异步 SQL 查询函数""" # 创建异步引擎(基于 asyncmy 驱动) async_engine = create_async_engine( f"mysql+asyncmy://{DB_CONFIG['user']}:{DB_CONFIG['password']}@{DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}", echo=False # 关闭 SQL 日志输出 ) async with async_engine.connect() as conn: # 执行 SQL 查询 result = await conn.execute(text(ttt)) # 获取查询结果并转为 DataFrame df = pd.DataFrame(result.fetchall(), columns=result.keys()) # 关闭异步引擎 await async_engine.dispose() return df async def mini_async(): """异步任务调度函数""" tasks = [runsql_async(table) for table in all_sql] results = await asyncio.gather(*tasks, return_exceptions=True) return results # 测试真异步版本速度 if __name__ == "__main__": kaishi = time.time() a = asyncio.run(mini_async()) # print(a) print("真异步(asyncmy)的速度", time.time() - kaishi)