Django实现热加载原理（从源码开始分析）-Toy模板网

这篇具有很好参考价值的文章主要介绍了Django实现热加载原理（从源码开始分析）。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

Django实现热加载原理（从源码开始分析）

源码地址

autoreload

代码实现

def run_with_reloader(main_func, *args, **kwargs):
    """
        监听系统的kill命令
        然后启动Django
    """
    signal.signal(signal.SIGTERM, lambda *args: sys.exit(0))
    try:
        """
        这里的判断，一开始 Django_AUTORELOAD_ENV 这个环境变量不能是’true'

        因为他会在restart_with_reloader这里执行一个死循环，然后设置Django_AUTORELOAD_ENV的值是true
        ，然后判断错误代码，如果错误代码不是3就return了，
        如果是3就继续循环
        """
        if os.environ.get(DJANGO_AUTORELOAD_ENV) == "true":
            reloader = get_reloader()
            logger.info(
                "Watching for file changes with %s", reloader.__class__.__name__
            )
            start_django(reloader, main_func, *args, **kwargs)
        else:
            exit_code = restart_with_reloader()
            sys.exit(exit_code)
    except KeyboardInterrupt:
        pass

这里是通过restart_with_reloader这个方法来实现的，首先我们第一次在python manage.py runserver的时候，DJANGO_AUTORELOAD_ENV 这个环境变量的值是None，当我们在执行完restart_with_reloader的时候，这个环境会变成true。

restart_with_reloader方法内部

创建了一个新的环境，然后使用子进程，并把新的环境变量进去，然后获取我们之前命令行的参数。然后判断结束吗等于3，如果不是3，则就直接return

      
 def restart_with_reloader():
    new_environ = {**os.environ, DJANGO_AUTORELOAD_ENV: "true"}
    args = get_child_arguments()
    while True:
        p = subprocess.run(args, env=new_environ, close_fds=False)
        if p.returncode != 3:
            return p.returncode

get_reloader

def get_reloader():
    """Return the most suitable reloader for this environment."""
    try:
        WatchmanReloader.check_availability()
    except WatchmanUnavailable:
        return StatReloader()
    return WatchmanReloader()

可以看到，这里是返回了WatchmanReloader或者StatReloader这个类。我们不看WatchmanReloader这个类，这个类是基于pywatchman 这个库实现的。StatReloader这个类是py自己实现的，其中大致逻辑可以可可以在这里看简单代码这个是我模拟Django来自己写的。

start_django

def start_django(reloader, main_func, *args, **kwargs):
    ensure_echo_on()

    main_func = check_errors(main_func)
    django_main_thread = threading.Thread(
        target=main_func, args=args, kwargs=kwargs, name="django-main-thread"
    )
    django_main_thread.daemon = True
    django_main_thread.start()

    while not reloader.should_stop:
        reloader.run(django_main_thread)

这里就开始跑Django项目了。

这里是以线程的方法开了Django服务，然后判断是否需要停止，如果不停止，则就开始监听

reloader

class BaseReloader:
    def __init__(self):
        self.extra_files = set()
        self.directory_globs = defaultdict(set)
        self._stop_condition = threading.Event()

    def watch_dir(self, path, glob):
        path = Path(path)
        try:
            path = path.absolute()
        except FileNotFoundError:
            logger.debug(
                "Unable to watch directory %s as it cannot be resolved.",
                path,
                exc_info=True,
            )
            return
        logger.debug("Watching dir %s with glob %s.", path, glob)
        self.directory_globs[path].add(glob)

    def watched_files(self, include_globs=True):
        """
        Yield all files that need to be watched, including module files and
        files within globs.
        """
        yield from iter_all_python_module_files()
        yield from self.extra_files
        if include_globs:
            for directory, patterns in self.directory_globs.items():
                for pattern in patterns:
                    yield from directory.glob(pattern)

    def wait_for_apps_ready(self, app_reg, django_main_thread):
        """
        Wait until Django reports that the apps have been loaded. If the given
        thread has terminated before the apps are ready, then a SyntaxError or
        other non-recoverable error has been raised. In that case, stop waiting
        for the apps_ready event and continue processing.

        Return True if the thread is alive and the ready event has been
        triggered, or False if the thread is terminated while waiting for the
        event.
        """
        while django_main_thread.is_alive():
            if app_reg.ready_event.wait(timeout=0.1):
                return True
        else:
            logger.debug("Main Django thread has terminated before apps are ready.")
            return False

    def run(self, django_main_thread):
        logger.debug("Waiting for apps ready_event.")
        self.wait_for_apps_ready(apps, django_main_thread)
        from django.urls import get_resolver

        # Prevent a race condition where URL modules aren't loaded when the
        # reloader starts by accessing the urlconf_module property.
        try:
            get_resolver().urlconf_module
        except Exception:
            # Loading the urlconf can result in errors during development.
            # If this occurs then swallow the error and continue.
            pass
        logger.debug("Apps ready_event triggered. Sending autoreload_started signal.")
        autoreload_started.send(sender=self)
        self.run_loop()

    def run_loop(self):
        ticker = self.tick()
        while not self.should_stop:
            try:
                next(ticker)
            except StopIteration:
                break
        self.stop()

    def tick(self):
        """
        This generator is called in a loop from run_loop. It's important that
        the method takes care of pausing or otherwise waiting for a period of
        time. This split between run_loop() and tick() is to improve the
        testability of the reloader implementations by decoupling the work they
        do from the loop.
        """
        raise NotImplementedError("subclasses must implement tick().")

    @classmethod
    def check_availability(cls):
        raise NotImplementedError("subclasses must implement check_availability().")

    def notify_file_changed(self, path):
        results = file_changed.send(sender=self, file_path=path)
        logger.debug("%s notified as changed. Signal results: %s.", path, results)
        if not any(res[1] for res in results):
            trigger_reload(path)

    # These are primarily used for testing.
    @property
    def should_stop(self):
        return self._stop_condition.is_set()

    def stop(self):
        self._stop_condition.set()


class StatReloader(BaseReloader):
    SLEEP_TIME = 1  # Check for changes once per second.

    def tick(self):
        mtimes = {}
        while True:
            for filepath, mtime in self.snapshot_files():
                old_time = mtimes.get(filepath)
                mtimes[filepath] = mtime
                if old_time is None:
                    logger.debug("File %s first seen with mtime %s", filepath, mtime)
                    continue
                elif mtime > old_time:

                    logger.debug(
                        "File %s previous mtime: %s, current mtime: %s",
                        filepath,
                        old_time,
                        mtime,
                    )
                    print('file is change')
                    self.notify_file_changed(filepath)

            time.sleep(self.SLEEP_TIME)
            yield

    def snapshot_files(self):
        # watched_files may produce duplicate paths if globs overlap.
        seen_files = set()
        for file in self.watched_files():
            if file in seen_files:
                continue
            try:
                mtime = file.stat().st_mtime
            except OSError:
                # This is thrown when the file does not exist.
                continue
            seen_files.add(file)
            yield file, mtime

    @classmethod
    def check_availability(cls):
        return True

StatReloader 这个类是继承了BaseReloader，然后自己实现了tick方法，和check_availability方法。

run在BaseReloader这个类里面，可以看到他是运行了run_look，然后里面运行了tick这个方法。

tick方法是在自子类，也就是StatReloader这个里面实现的，可以看到，他这里就是遍历了监听的文件，然后和上一次修改的时间对比，如果大于上次修改的时间，就触发notify_file_changed方法，这个方法就很简单了，他会把进程杀死，饭后返回3。