Django递归autosearch函数调用后内存未释放问题排查

问题描述 #

已知Django本身不存在内存泄漏问题，但Web应用调用特定递归函数autosearch（最多迭代7次）后，内存占用大幅上升且无法完全释放。通过AWS的mem_used_perc指标观测到，仅调用该函数时内存占用会升高且无法回到之前的水平。

相关核心代码如下：

递归函数autosearch #

def autosearch(self, phase=1, report="", num = 10):
    """
    This is an ES search following specific rules to identify and populate
    the lead notifications
    """
 
    if phase == 1:
        self.referred_to.clear()
        if self.no_of_providers:
            num = self.no_of_providers
        else:
            num = 10

    sqs = OrganisationDocument.search()

    service_type = None
    # filter by care type
    if self.type_of_care_care_home:
        service_type = "service_care_home"
    elif self.type_of_care_home_care:
        service_type = "service_home_care"
    elif self.type_of_care_live_in_care:
        service_type = "service_live_in_care"
    elif self.type_of_care_retirement_village:
        service_type = "service_retirement_village"

    if service_type == "service_retirement_village":
        sqs = sqs.query(Q("multi_match", query=True, fields=service_type))
    elif service_type:
        sqs = sqs.query(
            Q("multi_match", query=True, fields=service_type)
            & Q("match", care_over_65=True)
        )
    else:
        sqs = sqs.query(Q("match", care_over_65=True))

    if self.budget_type:
        ranges = self.filter_by_budget_range(phase)
        sqs = sqs.query(Q("bool", should=list(ranges)))
    
    # filter on location and distance
    if self.radius:
        radius = self.radius
    else:
        radius = 5

    """Increase radius by 2 or 10% for phase 2, 5, and 6"""
    if phase in [2, 6]:
        if radius < 20:
            radius += 2
        else:
            radius *= 1.1
    sqs = sqs.query(
            "geo_distance",
            distance=f"{radius}mi",
            location={
                "lat": self.searcharea_set.all()[0].lat,
                "lon": self.searcharea_set.all()[0].lng,
            },
        ) 
   
    # Filter by care_category_type
    categories = []
    if self.care_need_category_residential:
        categories.append("care_residential")
    if self.care_need_category_nursing:
        categories.append("care_nursing")
    if self.care_need_category_dementia:
        categories.append("care_dementia")
    if self.care_need_category_nursing_dementia:
        pass
    if self.care_need_category_personal_care:
        categories.append("care_residential")
    if self.care_need_category_respite_care:
        categories.append("care_respite")
    if self.care_need_palliative:
        pass
    if self.care_need_end_of_life:
        pass
    if self.care_need_retirement_housing:
        categories.append("retirement_living_housing")
    if self.care_need_retirement_village:
        categories.append("retirement_living_village")
    if self.care_need_retirement_community:
        categories.append("retirement_living_community")
    if self.care_need_retirement_full_time:
        pass

    query = []
    for category in categories:
        if self.type_of_care_live_in_care or self.type_of_care_home_care:
            if category == "care_residential":
                category = "regulated_personal_care"

        if category == "care_nursing":
            query.append(
                Q(
                    Q("match", regulated_nursing_care=True)
                    | Q("match", care_nursing=True)
                )
            )
        else:
            query.append(Q("match", **{f"{category}": True}))

    if self.type_of_care_retirement_village:
        sqs = sqs.query("bool", should=list(query))
    else:
        sqs = sqs.filter(Q("bool", must=query))

    # CQC Regulator filter
    sqs = sqs.query(
        Q(
            Q("match", cqc_rating_overall=1)
            | Q("match", cqc_rating_overall=2)
            | Q("match", cqc_rating_overall=3)
            | Q("match", cqc_rating_overall=99)
        )
    )

    # filter on profile
    if phase >= 4:
        sqs = sqs.query(Q("match", has_thumbnail_image=1))
        # Exclude Standard profiles in Brand with Premium Profile
        prems = [x.id for x in self.referred_to.all() if x.is_premium]
        sqs = sqs.query(~Q('bool', brand_link=list(prems)))
    else:
        sqs = sqs.query(Q("match", is_premium=1))
    
    # filter on budget
    if self.budget_type:
        ranges = self.filter_by_budget_range(phase)
        sqs = sqs.query(Q("bool", should=list(ranges)))

    # funding method
    if self.funding_method == choices.LOCAL_AUTHORITY:
        sqs = sqs.query(~Q("match", fees_local_authority_funded=False))
        # sqs = sqs.query('match', **{'fees_local_authority_funded': True})
    elif self.funding_method == choices.SELF_FUNDING:
        sqs = sqs.query("match", self_funding_clients=True)
    elif self.funding_method == choices.CONTINUING_HEALTHCARE:
        sqs = sqs.query(Q("match", fees_continuous_health_top_up=True))
    elif self.funding_method == choices.TOP_UP:
        sqs = sqs.query(~Q("match", fees_family_top_up=False))

    # Red crossed
    sqs = sqs.query(~Q("match", autumna_flag=2))
    # amber flagged
    if phase < 7:
        sqs = sqs.query(~Q("match", autumna_flag=1))

    # email only
    if not self.may_contact_provider:
        sqs = sqs.query(~Q("match", leads_accepted=1))
        sqs = sqs.query(~Q("match", leads_accepted=2))

    # no permission
    if not self.may_contact_phone:
        sqs = sqs.query(~Q("match", leads_accepted=1))

    # timescales
    if self.timescales in ["ASAP", "2-4 weeks"]:
        sqs = sqs.query(
            Q(
                Q("match", availability_overall=1)
                | Q("match", availability_overall=2)
            )
        )

    sqs.sort("-is_premium", "-has_thumbnail_image", "-is_claimed", self.sort_geo_location_dict())
    sqs = sqs[:num]

    report += f"<p>Phase {phase}:"
    added_count = 0
    for organisation in sqs.to_queryset():
        if not organisation in self.referred_to.all():
            added_count += 1
            self.referred_to.add(organisation)
            report += f"{organisation}, "
    report += "</p>"

    if added_count >= num or phase >= 7:
        self.autosearch_interim_report = report
        self.save()
    else:
        phase += 1
        num -= added_count
        return self.autosearch(phase, report, num)
    return True

调用代码（Admin页面） #

def autosearch(self, request, pk):
    """
    Populates the lead notifications
    """
    lead = get_object_or_404(models.Lead, pk=pk)

    # first, add organisations to favourites
    result = lead.autosearch()
    logger.debug(f'Auto search result: {result}')

    return redirect(reverse("admin:lead_management_lead_change", args=(pk,)))

缓存配置 #

CACHES = {
    "default": {
        "BACKEND": "django.core.cache.backends.memcached.MemcachedCache",
        "LOCATION": os.environ.get("CACHE_LOCATION", "127.0.0.1:11211"),
    }
}

filter_by_budget_range函数 #

def filter_by_budget_range(self, phase):
    """
    How the function works:
        - case 1: When a given value of min and max budget
                - It simply filter organisations between this range.

        - case 2: When a given value is only max budget
            - From given max budget we derived the min budget
                - min_budget = max_budget - max_budget * 0.2
                - e.g max_budget = 100, then the min_budget = 100 - 100 * 0.2 => 80

        - case 3: When a given value is only min budget
            - From given min budget we derived the max budget
                - max_budget = min_budget + min_budget * 0.2
                - e.g min_budget = 100, then the max_budget = 100 + 100 * 0.2 => 120
    """
   
    _budget_min, _budget_max = self.min_max_handler(phase)        
    if _budget_max is not None:
        if self.type_of_care_live_in_care or self.type_of_care_care_home:
            yield {
                "range": {
                    "fees_weekly_residential_costs_from": {
                        "gte": _budget_min * 100,
                        "lte": _budget_max * 100,
                    }
                }
            }      
        else:
            yield {
                "range": {
                    "fees_hourly_start_from": {
                        "gte": _budget_min * 100,
                        "lte": _budget_max * 100,
                    }
                }
            }

问题根源分析 #

1. 递归栈帧累积：即使最多7次递归，每次调用都会保留当前函数的局部变量（如ES查询对象sqs、report字符串、循环加载的Organisation实例），直到整个递归链结束才会被垃圾回收（GC），期间会占用额外内存。
2. ES查询对象冗余：每次递归都重新创建OrganisationDocument.search()并叠加大量查询条件，这些对象会持有查询上下文和中间状态，未及时释放。
3. ORM实例缓存冗余：
   • 循环中调用sqs.to_queryset()会加载大量Organisation实例，Django ORM默认会缓存这些实例；
   • self.referred_to.all()在循环和递归中多次调用，每次都会查询数据库并缓存结果，导致内存中累积大量重复实例。
4. 字符串拼接低效：report变量用+=拼接，每次都会创建新的字符串对象，递归多次后会产生多个中间字符串，占用额外内存。
5. 重复的预算过滤：代码中两次调用filter_by_budget_range(phase)并执行相同的查询过滤，冗余操作增加了内存占用和计算开销。

具体修复步骤 #

1. 将递归改为迭代循环 #

避免栈帧累积，手动管理循环变量，让局部变量在每次循环结束后可被及时回收：

def autosearch(self):
    phase = 1
    report_parts = []
    num = self.no_of_providers or 10
    self.referred_to.clear()

    while phase <= 7 and num > 0:
        sqs = OrganisationDocument.search()

        # --- 原有的所有查询逻辑（从service_type判断到sqs排序、切片）---
        service_type = None
        if self.type_of_care_care_home:
            service_type = "service_care_home"
        elif self.type_of_care_home_care:
            service_type = "service_home_care"
        elif self.type_of_care_live_in_care:
            service_type = "service_live_in_care"
        elif self.type_of_care_retirement_village:
            service_type = "service_retirement_village"

        if service_type == "service_retirement_village":
            sqs = sqs.query(Q("multi_match", query=True, fields=service_type))
        elif service_type:
            sqs = sqs.query(
                Q("multi_match", query=True, fields=service_type)
                & Q("match", care_over_65=True)
            )
        else:
            sqs = sqs.query(Q("match", care_over_65=True))

        # 合并重复的预算过滤逻辑
        if self.budget_type:
            ranges = self.filter_by_budget_range(phase)
            sqs = sqs.query(Q("bool", should=list(ranges)))
        
        # filter on location and distance
        radius = self.radius or 5
        if phase in [2, 6]:
            if radius < 20:
                radius += 2
            else:
                radius *= 1.1
        sqs = sqs.query(
                "geo_distance",
                distance=f"{radius}mi",
                location={
                    "lat": self.searcharea_set.all()[0].lat,
                    "lon": self.searcharea_set.all()[0].lng,
                },
            ) 
       
        # Filter by care_category_type
        categories = []
        if self.care_need_category_residential:
            categories.append("care_residential")
        if self.care_need_category_nursing:
            categories.append("care_nursing")
        if self.care_need_category_dementia:
            categories.append("care_dementia")
        if self.care_need_category_personal_care:
            categories.append("care_residential")
        if self.care_need_category_respite_care:
            categories.append("care_respite")
        if self.care_need_retirement_housing:
            categories.append("retirement_living_housing")
        if self.care_need_retirement_village:
            categories.append("retirement_living_village")
        if self.care_need_retirement_community:
            categories.append("retirement_living_community")

        query = []
        for category in categories:
            if self.type_of_care_live_in_care or self.type_of_care_home_care:
                if category == "care_residential":
                    category = "regulated_personal_care"

            if category == "care_nursing":
                query.append(
                    Q(
                        Q("match", regulated_nursing_care=True)
                        | Q("match", care_nursing=True)
                    )
                )
            else:
                query.append(Q("match", **{f"{category}": True}))

        if self.type_of_care_retirement_village:
            sqs = sqs.query("bool", should=list(query))
        else:
            sqs = sqs.filter(Q("bool", must=query))

        # CQC Regulator filter
        sqs = sqs.query(
            Q(
                Q("match", cqc_rating_overall=1)
                | Q("match", cqc_rating_overall=2)
                | Q("match", cqc_rating_overall=3)
                | Q("match", cqc_rating_overall=99)
            )
        )

        # filter on profile
        if phase >= 4:
            sqs = sqs.query(Q("match", has_thumbnail_image=1))
            # 预取Premium ID，避免多次查询
            prems = set(self.referred_to.filter(is_premium=True).values_list('id', flat=True))
            sqs = sqs.query(~Q('bool', brand_link=list(prems)))
        else:
            sqs = sqs.query(Q("match", is_premium=1))

        # funding method
        if self.funding_method == choices.LOCAL_AUTHORITY:
            sqs = sqs.query(~Q("match", fees_local_authority_funded=False))
        elif self.funding_method == choices.SELF_FUNDING:
            sqs = sqs.query("match", self_funding_clients=True)
        elif self.funding_method == choices.CONTINUING_HEALTHCARE:
            sqs = sqs.query(Q("match", fees_continuous_health_top_up=True))
        elif self.funding_method == choices.TOP_UP:
            sqs = sqs.query(~Q("match", fees_family_top_up=False))

        # Red crossed
        sqs = sqs.query(~Q("match", autumna_flag=2))
        # amber flagged
        if phase < 7:
            sqs = sqs.query(~Q("match", autumna_flag=1))

        # email only
        if not self.may_contact_provider:
            sqs = sqs.query(~Q("match", leads_accepted=1))
            sqs = sqs.query(~Q("match", leads_accepted=2))

        #