Python性能分析

python性能分析装饰器，官方文档

• line profile分析

# line profile
import time
from functools import wraps
from line_profiler import LineProfiler
 
# 查询接口中每行代码执行的时间
def func_line_time(f):
    @wraps(f)
    def decorator(*args, **kwargs):
        func_return = f(*args, **kwargs)
        lp = LineProfiler()
        lp_wrap = lp(f)
        lp_wrap(*args, **kwargs)
        lp.print_stats()
        return func_return
    return decorator

• cprofile分析器

# cprofile
import cProfile
import pstats
import functools
def do_cProfile(do=False, order='tottime'):
    def wrapper(func):
        @functools.wraps(func)
        def profiled_func(*args, **kwargs):
            if do:
                profile = cProfile.Profile()
                profile.enable()
                result = func(*args, **kwargs)
                profile.disable()
                #profile.print_stats()
                ps = pstats.Stats(profile).sort_stats(order).strip_dirs()
                ps.print_stats()
                return result
            else:
                result = func(*args, **kwargs)
                return result
        return profiled_func
    return wrapper

分析结果的可视化

• 工具安装

# macos
brew install qcachegrind
brew install graphviz
pip install pyprof2calltree

https://docs.python.org/zh-cn/3/library/profile.html
https://www.jianshu.com/p/c9287e56f44d
https://zhuanlan.zhihu.com/p/24495603

性能优化

1. 异步IO: 协程

# python3.6
import asyncio
import time

async def async_say(delay, msg):
    await asyncio.sleep(delay)
    print(msg)

async def main():
    loop = asyncio.get_event_loop()
    task1 = loop.create_task(async_say(4, 'hello'))
    task2 = loop.create_task(async_say(6, 'world'))

    print(f"started at {time.strftime('%X')}")
    await task1
    await task2
    print(f"finished at {time.strftime('%X')}")

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

# 或者以下写法
import asyncio
async def async_say(delay, msg):
    await asyncio.sleep(delay)
    print(msg)

async def main():
    loop = asyncio.get_event_loop()
    task1 = loop.create_task(async_say(4, 'hello'))
    task2 = loop.create_task(async_say(6, 'world'))

    print(f"started at {time.strftime('%X')}")
    await task1
    await task2
    print(f"finished at {time.strftime('%X')}")

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# loop = asyncio.get_event_loop()
loop.run_until_complete(main())

注意区分写法：

# v1串行运行10s
import asyncio
import time
async def async_say(delay, msg):
    await asyncio.sleep(delay)
    print(msg)

async def main(loop1):
    loop = asyncio.get_event_loop()
    print(loop1==loop)
    print(f"started at {time.strftime('%X')}")
    event_list = []
    for i in range(5):
        task = loop.create_task(async_say(i, i))
        await task # 等待task执行完毕，造成串行
    print(f"finished at {time.strftime('%X')}")
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop = asyncio.get_event_loop()
loop.run_until_complete(main(loop))

# v2并发运行，4s
import asyncio
import time
async def async_say(delay, msg):
    await asyncio.sleep(delay)
    print(msg)

async def main(loop1):
    loop = asyncio.get_event_loop()
    print(loop1==loop)
    print(f"started at {time.strftime('%X')}")
    event_list = []
    for i in range(5):
        task = loop.create_task(async_say(i, i))
        event_list.append(task)
    for i in range(5):
        res = await event_list[i]
    print(f"finished at {time.strftime('%X')}")
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop = asyncio.get_event_loop()
loop.run_until_complete(main(loop))



# 注意以下注释
import asyncio

async def nested():
    return 42

async def main():
    # Schedule nested() to run soon concurrently
    # with "main()".
    task = asyncio.create_task(nested())

    # "task" can now be used to cancel "nested()", or
    # can simply be awaited to wait until it is complete:
    await task

asyncio.run(main())

https://docs.python.org/zh-cn/3/library/asyncio-task.html
https://cuiqingcai.com/6160.html

性能优化的建议

https://www.jianshu.com/p/2b71b8fd97aa
https://wiki.python.org/moin/PythonSpeed/PerformanceTips

异步request(httpx)

https://zetcode.com/python/httpx/