Ezio's Blog

record


  • Home

  • Categories

  • Archives

  • Tags

  • About

Django源码分析之执行入口

Posted on 2016-05-20   |  

魔法门

一般我们启动django,最简单的方法是进入project 目录,这时目录结构是这样的

然后我们执行python manage.py runserver,程序就开始执行了。

那django是如何从一个命令就启动整个server,启动的流程是如何的?

踏门而入

打开目录下的manage.py,内容是这样的:

#!/usr/bin/env python
import os
import sys

if __name__ == "__main__":
    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "django_learning.settings")

    from django.core.management import execute_from_command_line

    execute_from_command_line(sys.argv)

看来manage.py只是把命令行参数传给django.core.management模块中的execute_from_command_line 函数。

查看execute_from_command_line函数,可以发现实际执行的是ManagementUtility类的excute方法:

def execute(self):
        """
        Given the command-line arguments, this figures out which subcommand is
        being run, creates a parser appropriate to that command, and runs it.
        """
        try:
            subcommand = self.argv[1]
        except IndexError:
            subcommand = 'help'  # Display help if no arguments were given.

        # Preprocess options to extract --settings and --pythonpath.
        # These options could affect the commands that are available, so they
        # must be processed early.
        parser = CommandParser(None, usage="%(prog)s subcommand [options] [args]", add_help=False)
        parser.add_argument('--settings')
        parser.add_argument('--pythonpath')
        parser.add_argument('args', nargs='*')  # catch-all
        try:
            options, args = parser.parse_known_args(self.argv[2:])
            handle_default_options(options)
        except CommandError:
            pass  # Ignore any option errors at this point.

        no_settings_commands = [
            'help', 'version', '--help', '--version', '-h',
            'compilemessages', 'makemessages',
            'startapp', 'startproject',
        ]

        try:
            settings.INSTALLED_APPS
        except ImproperlyConfigured as exc:
            self.settings_exception = exc
            # A handful of built-in management commands work without settings.
            # Load the default settings -- where INSTALLED_APPS is empty.
            if subcommand in no_settings_commands:
                settings.configure()

        if settings.configured:
            # Start the auto-reloading dev server even if the code is broken.
            # The hardcoded condition is a code smell but we can't rely on a
            # flag on the command class because we haven't located it yet.
            if subcommand == 'runserver' and '--noreload' not in self.argv:
                try:
                    autoreload.check_errors(django.setup)()
                except Exception:
                    # The exception will be raised later in the child process
                    # started by the autoreloader. Pretend it didn't happen by
                    # loading an empty list of applications.
                    apps.all_models = defaultdict(OrderedDict)
                    apps.app_configs = OrderedDict()
                    apps.apps_ready = apps.models_ready = apps.ready = True

            # In all other cases, django.setup() is required to succeed.
            else:
                django.setup()

        self.autocomplete()

        if subcommand == 'help':
            if '--commands' in args:
                sys.stdout.write(self.main_help_text(commands_only=True) + '\n')
            elif len(options.args) < 1:
                sys.stdout.write(self.main_help_text() + '\n')
            else:
                self.fetch_command(options.args[0]).print_help(self.prog_name, options.args[0])
        # Special-cases: We want 'django-admin --version' and
        # 'django-admin --help' to work, for backwards compatibility.
        elif subcommand == 'version' or self.argv[1:] == ['--version']:
            sys.stdout.write(django.get_version() + '\n')
        elif self.argv[1:] in (['--help'], ['-h']):
            sys.stdout.write(self.main_help_text() + '\n')
        else:
            self.fetch_command(subcommand).run_from_argv(self.argv)

其中

parser = CommandParser(None, usage="%(prog)s subcommand [options] [args]", add_help=False)
        parser.add_argument('--settings')
        parser.add_argument('--pythonpath')
        parser.add_argument('args', nargs='*')  # catch-all
        try:
            options, args = parser.parse_known_args(self.argv[2:])
            handle_default_options(options)
        except CommandError:
            pass  # Ignore any option errors at this point.

CommandParser其实类似于Argparse的一个解析命令行参数的类,从代码里可以看出我们可以直接在命令行指定settings文件和pythonpath。

no_settings_commands = [
            'help', 'version', '--help', '--version', '-h',
            'compilemessages', 'makemessages',
            'startapp', 'startproject',
        ]
try:
            settings.INSTALLED_APPS
except ImproperlyConfigured as exc:
            self.settings_exception = exc
            # A handful of built-in management commands work without settings.
            # Load the default settings -- where INSTALLED_APPS is empty.
            if subcommand in no_settings_commands:
                settings.configure()

这块代码就可以解释我们执行python manage.py start project 时django在背后会调用settings.configure方法,这里的settings是指django.conf.LazySettings的一个实例,configure方法其实就是使用django.conf.global_settings.py中的默认设置创建一份新的配置文件,作为我们新创建的project的settings.py

if settings.configured:
            # Start the auto-reloading dev server even if the code is broken.
            # The hardcoded condition is a code smell but we can't rely on a
            # flag on the command class because we haven't located it yet.
            if subcommand == 'runserver' and '--noreload' not in self.argv:
                try:
                    autoreload.check_errors(django.setup)()
                except Exception:
                    # The exception will be raised later in the child process
                    # started by the autoreloader. Pretend it didn't happen by
                    # loading an empty list of applications.
                    apps.all_models = defaultdict(OrderedDict)
                    apps.app_configs = OrderedDict()
                    apps.apps_ready = apps.models_ready = apps.ready = True

            # In all other cases, django.setup() is required to succeed.
            else:
                django.setup()

autoreload.check_errors(django.setup)()其实也是调用django.setup方法,而django.setup方法

def setup():
    """
    Configure the settings (this happens as a side effect of accessing the
    first setting), configure logging and populate the app registry.
    """
    from django.apps import apps
    from django.conf import settings
    from django.utils.log import configure_logging

    configure_logging(settings.LOGGING_CONFIG, settings.LOGGING)
    apps.populate(settings.INSTALLED_APPS)

负责初始化日志模块以及所有应用.

抽丝剥茧

剩下的代码最重要的就是这一句:

self.fetch_command(subcommand).run_from_argv(self.argv)

fetch_command会根据subcommand(这是我们执行python manage.py rumserver时传入的第二个参数:runserver),去django.core.management.commands中查找对应的command类,然后把所有命令行参数传给run_from_argv方法并执行,在runserver这个示例中,最终会调用django.utils.autoreload中的python_reloader或者jython_reloader新开一个线程:

def python_reloader(main_func, args, kwargs):
    if os.environ.get("RUN_MAIN") == "true":
        thread.start_new_thread(main_func, args, kwargs)
        try:
            reloader_thread()
        except KeyboardInterrupt:
            pass
    else:
        try:
            exit_code = restart_with_reloader()
            if exit_code < 0:
                os.kill(os.getpid(), -exit_code)
            else:
                sys.exit(exit_code)
        except KeyboardInterrupt:
            pass

这里的main_func是commands/runserver.py中的inner_run方法:

def inner_run(self, *args, **options):
        # If an exception was silenced in ManagementUtility.execute in order
        # to be raised in the child process, raise it now.
        autoreload.raise_last_exception()

        threading = options.get('use_threading')
        shutdown_message = options.get('shutdown_message', '')
        quit_command = 'CTRL-BREAK' if sys.platform == 'win32' else 'CONTROL-C'

        self.stdout.write("Performing system checks...\n\n")
        self.check(display_num_errors=True)
        self.check_migrations()
        now = datetime.now().strftime('%B %d, %Y - %X')
        if six.PY2:
            now = now.decode(get_system_encoding())
        self.stdout.write(now)
        self.stdout.write((
            "Django version %(version)s, using settings %(settings)r\n"
            "Starting development server at http://%(addr)s:%(port)s/\n"
            "Quit the server with %(quit_command)s.\n"
        ) % {
            "version": self.get_version(),
            "settings": settings.SETTINGS_MODULE,
            "addr": '[%s]' % self.addr if self._raw_ipv6 else self.addr,
            "port": self.port,
            "quit_command": quit_command,
        })

        try:
            handler = self.get_handler(*args, **options)
            run(self.addr, int(self.port), handler,
                ipv6=self.use_ipv6, threading=threading)
        except socket.error as e:
            # Use helpful error messages instead of ugly tracebacks.
            ERRORS = {
                errno.EACCES: "You don't have permission to access that port.",
                errno.EADDRINUSE: "That port is already in use.",
                errno.EADDRNOTAVAIL: "That IP address can't be assigned to.",
            }
            try:
                error_text = ERRORS[e.errno]
            except KeyError:
                error_text = force_text(e)
            self.stderr.write("Error: %s" % error_text)
            # Need to use an OS exit because sys.exit doesn't work in a thread
            os._exit(1)
        except KeyboardInterrupt:
            if shutdown_message:
                self.stdout.write(shutdown_message)
            sys.exit(0)

最关键的是这两条语句:

handler = self.get_handler(*args, **options)
run(self.addr, int(self.port), handler,ipv6=self.use_ipv6, threading=threading)

get_handler会返回django.core.servers.basehttp中定义的一个application(其实就是我们project下的wigs.py中定义的application)

这是run函数的内容

def run(addr, port, wsgi_handler, ipv6=False, threading=False):
    server_address = (addr, port)
    if threading:
        httpd_cls = type(str('WSGIServer'), (socketserver.ThreadingMixIn, WSGIServer), {})
    else:
        httpd_cls = WSGIServer
    httpd = httpd_cls(server_address, WSGIRequestHandler, ipv6=ipv6)
    if threading:
        # ThreadingMixIn.daemon_threads indicates how threads will behave on an
        # abrupt shutdown; like quitting the server by the user or restarting
        # by the auto-reloader. True means the server will not wait for thread
        # termination before it quits. This will make auto-reloader faster
        # and will prevent the need to kill the server manually if a thread
        # isn't terminating correctly.
        httpd.daemon_threads = True
    httpd.set_app(wsgi_handler)
    httpd.serve_forever()

可以看出run函数其实就是启动一个WSGIServer实例(WSGIServer继承python内置类simple_server.WSGIServer),并把handler设置为前面get_handler的返回值

水落石出

这样,一条python manage.py runserver命令的执行生命周期就一览无余了。
接下来,server就开始接收请求了。

_

Django源码分析之server

Posted on 2016-05-10   |  

乍见

Django内置的server基本包括两部分:django.core.servers和django.core.handlers

相识

servers.basehttp是Django自身提供的一个用于开发测试的server模块,其中提供的WSGIServer、ServerHandler、WSGIRequestHandler其实都是属于WSGI server,django只不过是对python内置的WSGI模块simple_server做的一层包装。

handlers package包括base.py和wsgi.py两个模块。
base.py中只定义了一个BaseHandler类,它复杂一些基础的功能,比如加载中间件,处理异常,获取响应数据等

wsgi.py才是主角,其中最重要的就是WSGIHandler类,它继承了base.py中的BaseHandler,只添加了一个__call__方法,那为什么添加这个方法呢?

django.core.wsgi

import django
from django.core.handlers.wsgi import WSGIHandler


def get_wsgi_application():
    """
    The public interface to Django's WSGI support. Should return a WSGI
    callable.

    Allows us to avoid making django.core.handlers.WSGIHandler public API, in
    case the internal WSGI implementation changes or moves in the future.
    """
    django.setup()
    return WSGIHandler()

可见我们启动server时传入的application其实是WSGIHandler的一个实例,而根据WSGI规范,这个application必须要是callable,python中的callable包括函数、方法以及任何定义了__call__方法的对象

回到handlers.wsgi

class WSGIHandler(base.BaseHandler):
    initLock = Lock()
    request_class = WSGIRequest

    def __call__(self, environ, start_response):
        # Set up middleware if needed. We couldn't do this earlier, because
        # settings weren't available.
        if self._request_middleware is None:
            with self.initLock:
                try:
                    # Check that middleware is still uninitialized.
                    if self._request_middleware is None:
                        self.load_middleware()
                except:
                    # Unload whatever middleware we got
                    self._request_middleware = None
                    raise

        set_script_prefix(get_script_name(environ))
        signals.request_started.send(sender=self.__class__, environ=environ)
        try:
            request = self.request_class(environ)
        except UnicodeDecodeError:
            logger.warning('Bad Request (UnicodeDecodeError)',
                exc_info=sys.exc_info(),
                extra={
                    'status_code': 400,
                }
            )
            response = http.HttpResponseBadRequest()
        else:
            response = self.get_response(request)

        response._handler_class = self.__class__

        status = '%s %s' % (response.status_code, response.reason_phrase)
        response_headers = [(str(k), str(v)) for k, v in response.items()]
        for c in response.cookies.values():
            response_headers.append((str('Set-Cookie'), str(c.output(header=''))))
        start_response(force_str(status), response_headers)
        if getattr(response, 'file_to_stream', None) is not None and environ.get('wsgi.file_wrapper'):
            response = environ['wsgi.file_wrapper'](response.file_to_stream)
        return response

由于__call__是一个请求的入口,它需要调用BaseHandler中定义的方法去执行加载中间件等一系列操作和异常处理,除此之外,WSGIHandler还会处理cookie、触发signal等。

至于如何返回响应,具体可看handlers.base模块,大概就是找出请求的path,通过匹配路由,找到并执行用户定义的view方法,执行中间件处理方法,最后返回响应。当然,还有一系列的异常处理。

回想

这部分的内容需要对WSGi协议有个大致的了解,可以参考:http://xiaorui.cc/2016/04/16/%E6%89%93%E9%80%A0mvc%E6%A1%86%E6%9E%B6%E4%B9%8Bwsgi%E5%8D%8F%E8%AE%AE%E7%9A%84%E4%BC%98%E7%BC%BA%E7%82%B9%E5%8F%8A%E6%8E%A5%E5%8F%A3%E5%AE%9E%E7%8E%B0/

排查实时tail功能cpu占用过高问题

Posted on 2016-04-08   |  

白日惊雷

“你的python应用cpu占用快90%了!!!”,良哥朝我眨了眨布满血丝的眼睛
“不会吧”,我心想:我这是好的啊

没接触过kafka的同学可以先了解下:(http://www.jasongj.com/2015/03/10/KafkaColumn1/)

疑云重重

SSH到远程机器上,运行top命令看一下,果然平常4%不到的cpu占用,现在飙升到90%左右了。

这是一个简单的应用:server端从kafka读消息,通过websocket发送到client端,整个server端代码也就几百行。

那就直接看代码吧。

由于线上环境的kafka没有开放端口,我是在本地搭的kafka环境,所以为了快速开发,就在server端直接写了一个kafka生产者,向kafka提交消息。因此,

猜想1: 难道是不小心把kafka生产者的代码也提交上去了?
check最新代码,生产者的代码明显被注释掉了,并没生效。

猜想2:websocket与server端链接太多了?
不可能,这个功能目前还没开始公测,而且只有这一个server cpu占比过高。

那还是server和kafka之间出现了问题。

初现端倪

server端会在每个请求到来时创建一个websocket连接,同时创建一个kafka消费者线程,用来监听特定topic的消息。

client端与server端的websocket会在用户刷新页面或者关闭页面时断开连接,这个不会有问题。那问题有可能出在:创建的kafka消费者线程没有正确地退出。

每个python线程会有一个daemon属性,默认为False。python主线程会在所有daemon为False的线程退出后才终止,而daemon为True的线程(也就是后台线程)会在主线程退出时一起退出。

所以无论如何,每次发布时都是重启整个进程,不会有资源回收失败的问题。

那就是在server运行时创建了过多的kafka消费者线程。

验证一下,打开pycharm的并发状态检测开关并启动server,新开一个页面,连续刷新几次,pycharm里就可以看到刚才创建的线程活的好好的!!!

水落石出

那来看看kafka消费者线程在干什么。

kafka消费者线程负责:

连接kafka
获取消息
向websocket连接写入消息

祭出debug神器:断点

根据多年打断点的经验,果断滴选择了“获取消息”。

经过一段时间的调试,原来是kafka-python从kafka获取消息时会进入无限while循环,从而阻塞线程。

解决方案

既然弄明白了问题的来龙去脉,解决起来就容易了。

首先看kaka-python有没有异步api,在官方文档里找了一圈,并没有,最后发现

,利用异常我们可以跳出while循环,从而有机会结束当前线程。大致代码如下:

class ConsumerThread(Threading.thread):
    ...

    def fetchMsg(self):
        for message in self.consumer:

            if self.stopThread:
                break

                message_value = message.value

                socket.pubsub(message_value)

            else:

                logger.error('consumer timeout')

                if not self.stopThread:
                    self.fetchMsg()
                else:
                    self.consumer.close()

一些感想

连接kafka的kafka-python竟然没做成事件驱动,反而是阻塞式,这不明显是挖坑让人跳么?

定时爬虫抓当日免费应用:Scrapy + Tkinter + LaunchControl

Posted on 2016-03-20   |  

花了个周末学了下Scrapy,正好一直想买mindnode,于是顺手做了个爬虫,抓取爱范儿每天的限免应用信息。

Thinking

大概思路就是使用LaunchControl每天定时(比如早上9点50,这时正好刚到公司不久)跑一下爬虫脚本,如果找到感兴趣的应用在限免,就使用Tkinter弹出提示。当然,也可以直接用Scrapy做定时任务,以后再说。

Coding

Scrapy + Tkinter

``` bash

-- coding: utf-8 --

import scrapy
import Tkinter
from scrapy.shell import inspect_response
import json

设置感兴趣的app名称

I_want_apps = set([‘mindnode pro’, ‘u.memory’])

class XianmianSpider(scrapy.Spider):
user_agent = ‘Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/49.0.2623.112 Safari/537.36’
name = “xianmian”
allowed_domains = [“app.so”]
start_urls = (
‘http://app.so/api/v1.1/appso/discount/?platform=web&limit=10‘,
)

def parse(self, response):

    jsonresponse = json.loads(response.body_as_unicode())

    apps = jsonresponse['objects']

    appTitles = {item['display_name'].lower() for item in apps}

    self.logger.info('today\' apps are: ' + str(appTitles))

    the_apps = appTitles & I_want_apps
    if the_apps:
        self.showMsg('found the apps: {}'.format(list(the_apps)))

def showMsg(self, msg):
    import Tkinter
    root = Tkinter.Tk()
    root.title('福利到!')
    label = Tkinter.Label(root, text=msg)
    label.pack()
    center_window(root, 300, 240)
    root.maxsize(600, 400)
    root.minsize(300, 240)
    Tkinter.mainloop()

def get_screen_size(window):
return window.winfo_screenwidth(),window.winfo_screenheight()

def get_window_size(window):
return window.winfo_reqwidth(),window.winfo_reqheight()

def center_window(root, width, height):
screenwidth = root.winfo_screenwidth()
screenheight = root.winfo_screenheight()
size = ‘%dx%d+%d+%d’ % (width, height, (screenwidth - width)/2, (screenheight - height)/2)
print(size)
root.geometry(size)

```

LaunchControl

title

LaunchControl用起来比较直观,当然,也可以直接用mac自带的launchctl,具体可参考launchctl使用说明

1…45
Ezio

Ezio

44 posts
© 2017 Ezio
Powered by Hexo
Theme - NexT.Mist