一、背景

在做code review的时候，经常会发现，因为开发习惯问题，很多研发人员喜欢在for循环中查询数据库。今天用一个真实例子，记录一次查询性能优化的例子

二、示例

代码如下:

@classmethod
@fn_performance()
def get_task_inst_info(cls, username, ids):"""获取巡检任务使用实例信息"""permission_host_list = list()check_task_queryset = CheckTask.objects.filter(id__in=ids, task__is_deleted=False)for check_task in check_task_queryset:instance_permission_dict = {"instance": check_task.id, "name": check_task.task.name, "permission": True}instance_info = list()obj_ids = list(CheckTaskInst.objects.filter(check_task=check_task).values_list("obj_id", flat=True))for obj_id in obj_ids:check_task_inst_queryset = CheckTaskInst.objects.filter(check_task=check_task, obj_id=obj_id)host_instance_list = CheckTaskInstIAMSerializer(check_task_inst_queryset, many=True).datapermission_host = HostIamService.screening_hosts_permissions(username, obj_id, host_instance_list, True)for _instance in permission_host:if not len(_instance["auth_actions"]):instance_permission_dict.update({"permission": False})instance_info.append({"permission": False,"inst_id": _instance.get("bk_host_id", _instance.get("bk_inst_id")),"inst_name": _instance.get("bk_host_name", _instance.get("bk_inst_name")),})instance_permission_dict.update({"instance_info": instance_info})permission_host_list.append(instance_permission_dict)return permission_host_list

以上代码，存在最大的问题是，在循环中查询数据库
实例数量33个，执行耗时3s左右

check_task_service.get_task_inst_info("admin",[108])
The [func_name: get_task_inst_info] [func: <function CheckTaskService.get_task_inst_info at 0x000001F5108FA620>]
[args:(<class 'apps.general_check.service.check_task_service.CheckTaskService'>, 'admin', [108])]
[kwargs: {}]
[timer:3.47244s] [threshold:1s], please timely optimize.

三、代码优化

优化思路：不在循环中查询数据库
优化之后效果显著，从3.4s降低到0.17s

check_task_service.get_task_inst_info("admin",[108])
The [func_name: get_task_inst_info] [func: <function CheckTaskService.get_task_inst_info at 0x0000028FA5BB2620>]
[args:(<class 'apps.general_check.service.check_task_service.CheckTaskService'>, 'admin', [108])]
[kwargs: {}]
[timer:0.17271s] [threshold:0.1s], please timely optimize.

优化后的代码：

    @classmethod@fn_performance(threshold=0.1)def get_task_inst_info(cls, username, ids):"""获取巡检任务使用实例信息"""permission_host_list = list()check_task_queryset = CheckTask.objects.select_related("task").filter(id__in=ids, task__is_deleted=False)check_task_inst = CheckTaskInst.objects.filter(check_task_id__in=ids)check_task_inst_data = {}for _task_inst in check_task_inst:check_task_inst_data.setdefault(f"{_task_inst.check_task_id}-{_task_inst.obj_id}", []).append(_task_inst)for check_task in check_task_queryset:instance_permission_dict = {"instance": check_task.id, "name": check_task.task.name, "permission": True}instance_info = list()obj_ids = check_task.task.obj_ids.split(",") if check_task.task.obj_ids else []for obj_id in obj_ids:check_task_inst_list = check_task_inst_data.get(f"{check_task.id}-{obj_id}")host_instance_list = CheckTaskInstIAMSerializer(check_task_inst_list, many=True).datapermission_host = HostIamService.screening_hosts_permissions(username, obj_id, host_instance_list, True)for _instance in permission_host:if not len(_instance["auth_actions"]):instance_permission_dict.update({"permission": False})instance_info.append({"permission": False,"inst_id": _instance.get("bk_host_id", _instance.get("bk_inst_id")),"inst_name": _instance.get("bk_host_name", _instance.get("bk_inst_name"))})instance_permission_dict.update({"instance_info": instance_info})permission_host_list.append(instance_permission_dict)return permission_host_list