jumpserver/apps/ops/celery/beat/schedulers.py

import logging
from celery.utils.log import get_logger
from django.db import close_old_connections
from django.core.exceptions import ObjectDoesNotExist
from django.db.utils import DatabaseError, InterfaceError

from django_celery_beat.schedulers import DatabaseScheduler as DJDatabaseScheduler

logger = get_logger(__name__)
debug, info, warning = logger.debug, logger.info, logger.warning


__all__ = ['DatabaseScheduler']


class DatabaseScheduler(DJDatabaseScheduler):

    def sync(self):
        if logger.isEnabledFor(logging.DEBUG):
            debug('Writing entries...')
        _tried = set()
        _failed = set()
        try:
            close_old_connections()

            while self._dirty:
                name = self._dirty.pop()
                try:
                    # 源码
                    # self.schedule[name].save()
                    # _tried.add(name)

                    """
                    ::Debug Description (2023.07.10)::
                    
                    如果调用 self.schedule 可能会导致 self.save() 方法之前重新获取数据库中的数据, 而不是临时设置的 last_run_at 数据
                    
                    如果这里调用 self.schedule
                    那么可能会导致调用 save 的 self.schedule[name] 的 last_run_at 是从数据库中获取回来的老数据
                    而不是任务执行后临时设置的 last_run_at (在 __next__() 方法中设置的)
                    当 `max_interval` 间隔之后, 下一个任务检测周期还是会再次执行任务
                    
                    ::Demo::
                    任务信息:
                        beat config: max_interval = 60s
                        
                        任务名称: cap
                        任务执行周期: 每 3 分钟执行一次
                        任务最后执行时间: 18:00
                        
                    任务第一次执行: 18:03 (执行时设置 last_run_at = 18:03, 此时在内存中)
                    
                    任务执行完成后, 
                    检测到需要 sync, sync 中调用了 self.schedule,
                    self.schedule 中发现 schedule_changed() 为 True, 需要调用 all_as_schedule()
                    此时，sync 中调用的 self.schedule[name] 的 last_run_at 是 18:00
                    这时候在 self.sync() 进行 self.save()
                    
                    
                    beat: Waking up 60s ...
                    
                    任务第二次执行: 18:04 (因为获取回来的 last_run_at 是 18:00, entry.is_due() == True)
                    
                    ::解决方法::
                    所以这里为了避免从数据库中获取，直接使用 _schedule #
                    """
                    self._schedule[name].save()
                    _tried.add(name)
                except (KeyError, TypeError, ObjectDoesNotExist):
                    _failed.add(name)
        except DatabaseError as exc:
            logger.exception('Database error while sync: %r', exc)
        except InterfaceError:
            warning(
                'DatabaseScheduler: InterfaceError in sync(), '
                'waiting to retry in next call...'
            )
        finally:
            # retry later, only for the failed ones
            self._dirty |= _failed