mysql 最快查询代码

# 异步 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",] if 1: 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", ]*5 # 数据库配置 DB_CONFIG = { "user": "root", "password": "root", "host": "127.0.0.1", "port": 3306, "database": "hdrx" } #====================最快速查询sql的方法. semaphore=asyncio.Semaphore(5) # 加上信号量,可以保证并发不蹦. async def runsql_async(ttt): async with semaphore: """真异步 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) # 信号量100时候: 35秒 # 信号量10时候:33秒 # 信号量20时候:35秒