经过4次优化我把python代码耗时减少95%

Posted 2021-11-12 金色旭光

结论一：事务不仅能够保证数据原子性，合理使用还能有效减少数据库查询耗时 结论二：集合操作的效率非常高，要善于使用集合减少循环 结论三：字典的查找效率高于列表，但是万次级别的操作无法体验优势

背景交代

团队做大学英语四六级考试相关服务。业务中有一个care服务，购买了care服务考试不过可以全额退款，不过有一个前提是要完成care服务的任务，比如坚持背单词N天，完成指定的试卷。

在这个背景下，当2021年6月的四六级考试完成之后，要统计出两种用户数据：

1. 完成care服务的用户
2. 没有完成care的用户

所以简化的逻辑就是要在所有的用户中区分出care完成用户和care未完成用户。

• 目标1：完成care服务
• 目标2：未完成care服务

所有目标用户的数量在2.7w左右，care完成用户在0.4w左右。所以我需要做的是在从数据库中查询出的 2.7w 所有用户，去另一个表区分出care完成用户和care未完成用户。

第一版

第一版：纯粹使用数据库查询。首先查询出所有目标，然后遍历所有用户，在遍历中使用user id从另一张表中查询出care完成用户。
总量在2.7w 左右，所以数据库就查询了2.7w次。

耗时统计：144.7 s

def remind_repurchase():
    # 查询出所有用户
    all_users = cm.UserPlan.select().where(cm.UserPlan.plan_id.in_([15, 16]))
    plan_15_16_not_refund_users = []
    plan_15_16_can_refund_users = []
    for user in all_users:
        # 从另一张表查询用户是否完成care
        user_insurance = cm.UserInsurance.select().where(
            cm.UserInsurance.user_id == user.user_id,
            cm.UserInsurance.plan_id == user.plan_id,
            cm.UserInsurance.status == cm.UserInsurance.STATUS_SUCCESS,
        )

        # care 完成
        if len(user_insurance) == 1:
            # 其他逻辑
            plan_15_16_can_refund_users.append(user.user_id)
        else:
            # care未完成用户
            plan_15_16_not_refund_users.append(user.user_id)
            

主要的耗时操作就在for循环查询数据库。这种耗时肯定是不被允许的，需要提高效率。

第二版

优化点：增加事务
第二版优化思路：对于 2.7w 次的数据库库查询肯定会有 2.7w 次建立连接、事务、查询语句转SQL等。2.7w次的开销也是一个极大的数字。理所当然的想到了减少事务的开销。将所有的数据库查询都放在一个事务中完成，就能够有效减少查询带来的耗时。

耗时统计：100.6 s

def remind_repurchase():

    # 查询出所有用户
    all_users = cm.UserPlan.select().where(cm.UserPlan.plan_id.in_([15, 16]))
    plan_15_16_not_refund_users = []
    plan_15_16_can_refund_users = []

    # 增加事务
    with pwdb.database.atomic():
        for user in all_users:
            # 从另一张表查询用户是否完成care
            user_insurance = cm.UserInsurance.select().where(
                cm.UserInsurance.user_id == user.user_id,
                cm.UserInsurance.plan_id == user.plan_id,
                cm.UserInsurance.status == cm.UserInsurance.STATUS_SUCCESS,
            )
            
            # care 完成
            if len(user_insurance) == 1:
                # 其他逻辑
                plan_15_16_can_refund_users.append(user.user_id)
            else:
                # care未完成用户
                plan_15_16_not_refund_users.append(user.user_id)

增加事务之后减少了44s，相当于缩短了时间30%的时间，由此可以看出事务在数据库中查询是一个比较耗时的操作。

第三版

优化点：将2.7w次的数据库查询转变成对列表的in操作。
第三版提出改进方案：原来的逻辑是循环 2.7w 次，在数据库中查询用户是否完成care服务。2.7w 次的数据库查询是耗时最长的原因，而可以改进的方法是将所有完成care服务的用户先一次性查询出来，放到一个列表中。遍历所有用户时不去查数据库，而是直接使用in操作在列表中查询。这种方法直接将 2.7w 次数据库遍历减少到1次，极大缩短了数据库查询耗时。

耗时统计：11.5 s

def remind_repurchase():
   
    all_users = cm.UserPlan.select().where(cm.UserPlan.plan_id.in_([15, 16]))
    plan_15_16_not_refund_users = []
    plan_15_16_can_refund_users = []
    
    # 所有care完成用户，先将所有用户查询出来放在一个列表中
    user_insurance = cm.UserInsurance.select().where(
        cm.UserInsurance.plan_id.in_([15, 16]),
        cm.UserInsurance.status == cm.UserInsurance.STATUS_SUCCESS,
    )
    user_insurance_list = [user.user_id for user in user_insurance]

   
    for user in all_users:
        user_id = user.user_id
        # care 完成
        if user.user_id in user_insurance_list:
            # 查分数
            plan_15_16_can_refund_users.append(user_id)
        else:
            # care未完成用户 + 非care用户
            plan_15_16_not_refund_users.append(user_id)

这一次优化的效果是非常显著的，可以看出想要提高代码效率要尽量减少数据库查询次数。

第四版

优化点：2.7w 次对列表的in操作变成对字典的in操作
在第三版中已经极大的优化了效率，但是仔细琢磨之后发现还是有提升的空间的。在第三版中 2.7w 次for循环，然后用in操作在列表中查询。众所周知python中对列表的in操作是遍历的，时间复杂度为0(n)，所以效率不高，而对字典的in操作时间复杂度为常数级别0(1)。所以在第四版优化中先查询出的数据不保存为列表，而是保存为字典。key就是原来列表中的值，value可自定义。

耗时统计：11.42 s

def remind_repurchase():
   
    all_users = cm.UserPlan.select().where(cm.UserPlan.plan_id.in_([15, 16]))
    plan_15_16_not_refund_users = []
    plan_15_16_can_refund_users = []
    
    # 所有care完成用户，先将所有用户查询出来放在一个列表中
    user_insurance = cm.UserInsurance.select().where(
        cm.UserInsurance.plan_id.in_([15, 16]),
        cm.UserInsurance.status == cm.UserInsurance.STATUS_SUCCESS,
    )

    user_insurance_dict = {user.user_id:True for user in user_insurance}

   
    for user in all_users:
        user_id = user.user_id
        # care 完成
        if user.user_id in user_insurance_dict:
            # 查分数
            plan_15_16_can_refund_users.append(user_id)
        else:
            # care未完成用户 + 非care用户
            plan_15_16_not_refund_users.append(user_id)

由于2.7w次的in操作数据量并不是很大，并且列表的in操作在python中优化的效率也很好，所以这里的对字典的in操作并没有减少时间消耗。

第五版

优化点：将in操作转变成集合操作。
在前四版的优化下已经将耗时缩短了 133s，减少了近 92.1% 的耗时，想着这个数据看起来还不错了。隔天早上在刷牙时脑子里思绪纷飞就想到这个事情了。这时忽然想到既然我能查询全部用户，又将完成care用户的用户查询到一个列表中，这时不就是相当于两个集合吗？既然是集合，那么使用集合之间的交集和差集是不是比循环 2.7w 次要快呢？上班之后马上动手来验证这个想法。果然，还能够减少时间消耗，将第四版中的11.42 直接减少了一半，缩短到5.78，缩短近50%。

耗时统计： 5.78 s

def remind_repurchase():
    
    all_users = cm.UserPlan.select().where(cm.UserPlan.plan_id.in_([15, 16]))
    all_users_set = set([user.user_id for user in all_users])

    plan_15_16_can_refund_users = []
    received_user_count = 0

    # 所有care完成用户
    user_insurance = cm.UserInsurance.select().where(
        cm.UserInsurance.plan_id.in_([15, 16]),
        cm.UserInsurance.status == cm.UserInsurance.STATUS_SUCCESS,
    )
    user_insurance_set = set([user.user_id for user in user_insurance])

    temp_can_refund_users = all_users_set.intersection(user_insurance_set)

总结

最终优化的结果：
第一版耗时： 144.7 s
最后一版耗时： 5.7 s
优化时间：109 s
优化百分比：95.0%

在各个版本中的优化详细细节如下：

由此可以得出几个结论，帮助减少程序耗时：
结论一：事务不仅能够保证数据原子性，合理使用还能有效减少数据库查询耗时
结论二：集合操作的效率非常高，要善于使用集合减少循环
结论三：字典的查找效率高于列表，但是万次级别的操作无法体验优势

最后的还有一个结论：程序员的灵感似乎在刷牙、上厕所、洗澡、喝水时特别活跃，所以写不出来代码就该去摸摸鱼了。

性能测试平台效率优化的一次经验（python版）

在做性能测试平台的优化过程中，由于启动任务相对其他测试任务比较频繁，而目前30次两个包的交叉对比(30次)测试需要耗时30分钟整，因此打算优先对测试流程做一次优化，将测试时间消耗降低到20分钟。

由于一开始估计乐观，认为启动时间，一台设备理论上启动顶多1s，1*2*30也就60s，加上其他开销，5分钟都够了，能减少到20分钟应该小半天就能做完了。

于是就来到了第一步：

1.review代码流程

（1）把启动流程里相关的sleep全部review一遍

确实有一点效果，因为有一部分sleep在启动任务执行阶段，60倍杠杆放大后很可怕，因此去掉部分sleep，居然就减少到了23分钟了。

第二步一时想不出了，方法耦合嵌套相当多，而且适配多个版本的产品，迁一发动全身，第二步想到的就是将可疑方法监控起来

 

2.将可疑方法的耗时监控起来

为了方便监控，增加了两个个装饰器来统计耗时

def costs(fn):
    def _wrapper(*args, **kwargs):
        start = time.time()
        fn(*args, **kwargs)
        print "%s 函数cost %s 秒" % (fn.__name__, time.time() - start)

    return _wrapper


def costs_with_info(info):
    def _wrapper(fn):
        print "info: " + info
        return costs(fn)
    return _wrapper

当方法需要监控时，则加入@costs或者@costs_with_info("some infomation")

    @costs
    def configureQuickStart(self, pkg_name):
        if self.config.allow_quick_start == "1":
            self.logger.info("Disable quick start: %s" % pkg_name)
            self.disableQuickStartSnapShot(pkg_name)
        else:
            self.logger.info("Quick start is enabled: %s" % pkg_name)

推荐大家不要用这样的方法，真心耗时耗力，而且效果差。花了半天优化了一分钟。

于是想到了Android里的traceview，traceview有方法能拿到整个调用栈的性能消耗，包括耗时，python应该也有这样的方法才对，然后我找到了cProfile，于是便愉快地进入了第三步

3.使用cProfile进行分析

（1）直接将入口加入监控，输出result.prof文件，并在log区打印出tottime（不包含子方法的耗时统计）

import cProfile
import pstats
cProfile.run(‘main()‘, filename=‘result.prof‘, sort="tottime")
p = pstats.Stats(‘result.prof‘)
p.sort_stats(‘time‘).print_stats()

log区打印出的日志如下

一部分是系统方法，一部分是自己的方法，不是很直观，于是又找到了另一个神器graphviz。

首先需要安装：

sudo apt-get install graphviz

然后下载Gprof2Dot，接着运行

python gprof2dot.py -f pstats result.out | dot -Tpng -o result.png

终于我得到了一张启动测试的方法耗时统计图

局部展示如下：

这样就可以很清晰看到各个函数具体消耗的时间了，但令我震惊的是，启动测试中，30分钟里居然有95.29%的时间是在sleep！但是没关系，因为我知道是哪个方法开始引入的sleep,并且可以知道哪些是可以优化的。