分类 软件开发 下的文章

你在 Python 中用过异步编程吗?本文中我会告诉你怎样做,而且用一个能工作的例子来展示它:这是一个流行的贪吃蛇游戏,而且是为多人游戏而设计的。

1、简介

在技术和文化领域,大规模多人在线游戏(MMO)毋庸置疑是我们当今世界的潮流之一。很长时间以来,为一个 MMO 游戏写一个服务器这件事总是会涉及到大量的预算与复杂的底层编程技术,不过在最近这几年,事情迅速发生了变化。基于动态语言的现代框架允许在中档的硬件上面处理大量并发的用户连接。同时,HTML5 和 WebSockets 标准使得创建基于实时图形的游戏的直接运行至浏览器上的客户端成为可能,而不需要任何的扩展。

对于创建可扩展的非堵塞性的服务器来说,Python 可能不是最受欢迎的工具,尤其是和在这个领域里最受欢迎的 Node.js 相比而言。但是最近版本的 Python 正在改变这种现状。asyncio 的引入和一个特别的 async/await 语法使得异步代码看起来像常规的阻塞代码一样,这使得 Python 成为了一个值得信赖的异步编程语言,所以我将尝试利用这些新特点来创建一个多人在线游戏。

2、异步

一个游戏服务器应该可以接受尽可能多的用户并发连接,并实时处理这些连接。一个典型的解决方案是创建线程,然而在这种情况下并不能解决这个问题。运行上千的线程需要 CPU 在它们之间不停的切换(这叫做上下文切换),这将导致开销非常大,效率很低下。更糟糕的是使用进程来实现,因为,不但如此,它们还会占用大量的内存。在 Python 中,甚至还有一个问题,Python 的解释器(CPython)并不是针对多线程设计的,相反它主要针对于单线程应用实现最大的性能。这就是为什么它使用 GIL(global interpreter lock),这是一个不允许同时运行多线程 Python 代码的架构,以防止同一个共享对象出现使用不可控。正常情况下,在当前线程正在等待的时候,解释器会转换到另一个线程,通常是等待一个 I/O 的响应(举例说,比如等待 Web 服务器的响应)。这就允许在你的应用中实现非阻塞 I/O 操作,因为每一个操作仅仅阻塞一个线程而不是阻塞整个服务器。然而,这也使得通常的多线程方案变得几近无用,因为它不允许你并发执行 Python 代码,即使是在多核心的 CPU 上也是这样。而与此同时,在一个单一线程中拥有非阻塞 I/O 是完全有可能的,因而消除了经常切换上下文的需要。

实际上,你可以用纯 Python 代码来实现一个单线程的非阻塞 I/O。你所需要的只是标准的 select 模块,这个模块可以让你写一个事件循环来等待未阻塞的 socket 的 I/O。然而,这个方法需要你在一个地方定义所有 app 的逻辑,用不了多久,你的 app 就会变成非常复杂的状态机。有一些框架可以简化这个任务,比较流行的是 tornadetwisted。它们被用来使用回调方法实现复杂的协议(这和 Node.js 比较相似)。这种框架运行在它自己的事件循环中,按照定义的事件调用你的回调函数。并且,这或许是一些情况的解决方案,但是它仍然需要使用回调的方式编程,这使你的代码变得碎片化。与写同步代码并且并发地执行多个副本相比,这就像我们在普通的线程上做的一样。在单个线程上这为什么是不可能的呢?

这就是为什么出现微线程(microthread)概念的原因。这个想法是为了在一个线程上并发执行任务。当你在一个任务中调用阻塞的方法时,有一个叫做“manager” (或者“scheduler”)的东西在执行事件循环。当有一些事件准备处理的时候,一个 manager 会转移执行权给一个任务,并等着它执行完毕。任务将一直执行,直到它遇到一个阻塞调用,然后它就会将执行权返还给 manager。

微线程也称为轻量级线程(lightweight threads)或绿色线程(green threads)(来自于 Java 中的一个术语)。在伪线程中并发执行的任务叫做 tasklets、greenlets 或者协程(coroutines)。

Python 中的微线程最早的实现之一是 Stackless Python。它之所以这么知名是因为它被用在了一个叫 EVE online 的非常有名的在线游戏中。这个 MMO 游戏自称说在一个持久的“宇宙”中,有上千个玩家在做不同的活动,这些都是实时发生的。Stackless 是一个独立的 Python 解释器,它代替了标准的函数栈调用,并且直接控制程序运行流程来减少上下文切换的开销。尽管这非常有效,这个解决方案不如在标准解释器中使用“软”库更流行,像 eventletgevent 的软件包配备了修补过的标准 I/O 库,I/O 函数会将执行权传递到内部事件循环。这使得将正常的阻塞代码转变成非阻塞的代码变得简单。这种方法的一个缺点是从代码上看这并不分明,它的调用是非阻塞的。新版本的 Python 引入了本地协程作为生成器的高级形式。在 Python 的 3.4 版本之后,引入了 asyncio 库,这个库依赖于本地协程来提供单线程并发。但是仅仅到了 Python 3.5 ,协程就变成了 Python 语言的一部分,使用新的关键字 async 和 await 来描述。这是一个简单的例子,演示了使用 asyncio 来运行并发任务。

import asyncio

async def my_task(seconds):
    print("start sleeping for {} seconds".format(seconds))
    await asyncio.sleep(seconds)
    print("end sleeping for {} seconds".format(seconds))

all_tasks = asyncio.gather(my_task(1), my_task(2))
loop = asyncio.get_event_loop()
loop.run_until_complete(all_tasks)
loop.close()    

我们启动了两个任务,一个睡眠 1 秒钟,另一个睡眠 2 秒钟,输出如下:

start sleeping for 1 seconds
start sleeping for 2 seconds
end sleeping for 1 seconds
end sleeping for 2 seconds

正如你所看到的,协程不会阻塞彼此——第二个任务在第一个结束之前启动。这发生的原因是 asyncio.sleep 是协程,它会返回执行权给调度器,直到时间到了。

在下一节中,我们将会使用基于协程的任务来创建一个游戏循环。


via: https://7webpages.com/blog/writing-online-multiplayer-game-with-python-asyncio-getting-asynchronous/

作者:Kyrylo Subbotin 译者:xinglianfly 校对:wxy

本文由 LCTT 原创编译,Linux中国 荣誉推出

(题图来自:deviantart.net

为什么你想要自己构建一个 web 框架呢?我想,原因有以下几点:

  • 你有一个新奇的想法,觉得将会取代其他的框架
  • 你想要获得一些名气
  • 你遇到的问题很独特,以至于现有的框架不太合适
  • 你对 web 框架是如何工作的很感兴趣,因为你想要成为一位更好的 web 开发者。

接下来的笔墨将着重于最后一点。这篇文章旨在通过对设计和实现过程一步一步的阐述告诉读者,我在完成一个小型的服务器和框架之后学到了什么。你可以在这个代码仓库中找到这个项目的完整代码。

我希望这篇文章可以鼓励更多的人来尝试,因为这确实很有趣。它让我知道了 web 应用是如何工作的,而且这比我想的要容易的多!

范围

框架可以处理请求-响应周期、身份认证、数据库访问、模板生成等部分工作。Web 开发者使用框架是因为,大多数的 web 应用拥有大量相同的功能,而对每个项目都重新实现同样的功能意义不大。

比较大的的框架如 Rails 和 Django 实现了高层次的抽象,或者说“自备电池”(“batteries-included”,这是 Python 的口号之一,意即所有功能都自足。)。而实现所有的这些功能可能要花费数千小时,因此在这个项目上,我们重点完成其中的一小部分。在开始写代码前,我先列举一下所需的功能以及限制。

功能:

  • 处理 HTTP 的 GET 和 POST 请求。你可以在这篇 wiki 中对 HTTP 有个大致的了解。
  • 实现异步操作(我喜欢 Python 3 的 asyncio 模块)。
  • 简单的路由逻辑以及参数撷取。
  • 像其他微型框架一样,提供一个简单的用户级 API 。
  • 支持身份认证,因为学会这个很酷啊(微笑)。

限制:

  • 将只支持 HTTP 1.1 的一个小子集,不支持 传输编码 transfer-encoding HTTP 认证 http-auth 内容编码 content-encoding (如 gzip)以及持久化连接等功能。
  • 不支持对响应内容的 MIME 判断 - 用户需要手动指定。
  • 不支持 WSGI - 仅能处理简单的 TCP 连接。
  • 不支持数据库。

我觉得一个小的用例可以让上述内容更加具体,也可以用来演示这个框架的 API:

from diy_framework import App, Router
from diy_framework.http_utils import Response


# GET simple route
async def home(r):
    rsp = Response()
    rsp.set_header('Content-Type', 'text/html')
    rsp.body = '<html><body><b>test</b></body></html>'
    return rsp


# GET route + params
async def welcome(r, name):
    return "Welcome {}".format(name)

# POST route + body param
async def parse_form(r):
    if r.method == 'GET':
        return 'form'
    else:
        name = r.body.get('name', '')[0]
        password = r.body.get('password', '')[0]

       return "{0}:{1}".format(name, password)

# application = router + http server
router = Router()
router.add_routes({
    r'/welcome/{name}': welcome,
    r'/': home,
    r'/login': parse_form,})

app = App(router)
app.start_server()

' 用户需要定义一些能够返回字符串或 Response 对象的异步函数,然后将这些函数与表示路由的字符串配对,最后通过一个函数调用(start_server)开始处理请求。

完成设计之后,我将它抽象为几个我需要编码的部分:

  • 接受 TCP 连接以及调度一个异步函数来处理这些连接的部分
  • 将原始文本解析成某种抽象容器的部分
  • 对于每个请求,用来决定调用哪个函数的部分
  • 将上述部分集中到一起,并为开发者提供一个简单接口的部分

我先编写一些测试,这些测试被用来描述每个部分的功能。几次重构后,整个设计被分成若干部分,每个部分之间是相对解耦的。这样就非常好,因为每个部分可以被独立地研究学习。以下是我上文列出的抽象的具体体现:

  • 一个 HTTPServer 对象,需要一个 Router 对象和一个 http\_parser 模块,并使用它们来初始化。
  • HTTPConnection 对象,每一个对象表示一个单独的客户端 HTTP 连接,并且处理其请求-响应周期:使用 http\_parser 模块将收到的字节流解析为一个 Request 对象;使用一个 Router 实例寻找并调用正确的函数来生成一个响应;最后将这个响应发送回客户端。
  • 一对 Request 和 Response 对象为用户提供了一种友好的方式,来处理实质上是字节流的字符串。用户不需要知道正确的消息格式和分隔符是怎样的。
  • 一个包含“路由:函数”对应关系的 Router 对象。它提供一个添加配对的方法,可以根据 URL 路径查找到相应的函数。
  • 最后,一个 App 对象。它包含配置信息,并使用它们实例化一个 HTTPServer 实例。

让我们从 HTTPConnection 开始来讲解各个部分。

模拟异步连接

为了满足上述约束条件,每一个 HTTP 请求都是一个单独的 TCP 连接。这使得处理请求的速度变慢了,因为建立多个 TCP 连接需要相对高的花销(DNS 查询,TCP 三次握手,慢启动等等的花销),不过这样更加容易模拟。对于这一任务,我选择相对高级的 asyncio-stream 模块,它建立在 asyncio 的传输和协议的基础之上。我强烈推荐你读一读标准库中的相应代码,很有意思!

一个 HTTPConnection 的实例能够处理多个任务。首先,它使用 asyncio.StreamReader 对象以增量的方式从 TCP 连接中读取数据,并存储在缓存中。每一个读取操作完成后,它会尝试解析缓存中的数据,并生成一个 Request 对象。一旦收到了这个完整的请求,它就生成一个回复,并通过 asyncio.StreamWriter 对象发送回客户端。当然,它还有两个任务:超时连接以及错误处理。

你可以在这里浏览这个类的完整代码。我将分别介绍代码的每一部分。为了简单起见,我移除了代码文档。

class HTTPConnection(object):
    def init(self, http_server, reader, writer):
        self.router = http_server.router
        self.http_parser = http_server.http_parser
        self.loop = http_server.loop

        self._reader = reader
        self._writer = writer
        self._buffer = bytearray()
        self._conn_timeout = None
        self.request = Request()

这个 init 方法没啥意思,它仅仅是收集了一些对象以供后面使用。它存储了一个 router 对象、一个 http_parser 对象以及 loop 对象,分别用来生成响应、解析请求以及在事件循环中调度任务。

然后,它存储了代表一个 TCP 连接的读写对,和一个充当原始字节缓冲区的空字节数组_conn_timeout 存储了一个 asyncio.Handle 的实例,用来管理超时逻辑。最后,它还存储了 Request 对象的一个单一实例。

下面的代码是用来接受和发送数据的核心功能:

async def handle_request(self):
    try:
        while not self.request.finished and not self._reader.at_eof():
            data = await self._reader.read(1024)
            if data:
                self._reset_conn_timeout()
                await self.process_data(data)
        if self.request.finished:
            await self.reply()
        elif self._reader.at_eof():
            raise BadRequestException()
    except (NotFoundException,
            BadRequestException) as e:
        self.error_reply(e.code, body=Response.reason_phrases[e.code])
    except Exception as e:
        self.error_reply(500, body=Response.reason_phrases[500])

    self.close_connection()

所有内容被包含在 try-except 代码块中,这样在解析请求或响应期间抛出的异常可以被捕获到,然后一个错误响应会发送回客户端。

while 循环中不断读取请求,直到解析器将 self.request.finished 设置为 True ,或者客户端关闭连接所触发的信号使得 self._reader_at_eof() 函数返回值为 True 为止。这段代码尝试在每次循环迭代中从 StreamReader 中读取数据,并通过调用 self.process_data(data) 函数以增量方式生成 self.request。每次循环读取数据时,连接超时计数器被重置。

这儿有个错误,你发现了吗?稍后我们会再讨论这个。需要注意的是,这个循环可能会耗尽 CPU 资源,因为如果没有读取到东西 self._reader.read() 函数将会返回一个空的字节对象 b''。这就意味着循环将会不断运行,却什么也不做。一个可能的解决方法是,用非阻塞的方式等待一小段时间:await asyncio.sleep(0.1)。我们暂且不对它做优化。

还记得上一段我提到的那个错误吗?只有从 StreamReader 读取数据时,self._reset_conn_timeout() 函数才会被调用。这就意味着,直到第一个字节到达时timeout 才被初始化。如果有一个客户端建立了与服务器的连接却不发送任何数据,那就永远不会超时。这可能被用来消耗系统资源,从而导致拒绝服务式攻击(DoS)。修复方法就是在 init 函数中调用 self._reset_conn_timeout() 函数。

当请求接受完成或连接中断时,程序将运行到 if-else 代码块。这部分代码会判断解析器收到完整的数据后是否完成了解析。如果是,好,生成一个回复并发送回客户端。如果不是,那么请求信息可能有错误,抛出一个异常!最后,我们调用 self.close_connection 执行清理工作。

解析请求的部分在 self.process_data 方法中。这个方法非常简短,也易于测试:

async def process_data(self, data):
    self._buffer.extend(data)

    self._buffer = self.http_parser.parse_into(
        self.request, self._buffer)

每一次调用都将数据累积到 self._buffer 中,然后试着用 self.http_parser 来解析已经收集的数据。这里需要指出的是,这段代码展示了一种称为依赖注入(Dependency Injection)的模式。如果你还记得 init 函数的话,应该知道我们传入了一个包含 http_parser 对象的 http_server 对象。在这个例子里,http_parser 对象是 diy_framework 包中的一个模块。不过它也可以是任何含有 parse_into 函数的类,这个 parse_into 函数接受一个 Request 对象以及字节数组作为参数。这很有用,原因有二:一是,这意味着这段代码更易扩展。如果有人想通过一个不同的解析器来使用 HTTPConnection,没问题,只需将它作为参数传入即可。二是,这使得测试更加容易,因为 http_parser 不是硬编码的,所以使用虚假数据或者 mock 对象来替代是很容易的。

下一段有趣的部分就是 reply 方法了:

async def reply(self):
    request = self.request
    handler = self.router.get_handler(request.path)

    response = await handler.handle(request)

    if not isinstance(response, Response):
        response = Response(code=200, body=response)

    self._writer.write(response.to_bytes())
    await self._writer.drain()

这里,一个 HTTPConnection 的实例使用了 HTTPServer 中的 router 对象来得到一个生成响应的对象。一个路由可以是任何一个拥有 get_handler 方法的对象,这个方法接收一个字符串作为参数,返回一个可调用的对象或者抛出 NotFoundException 异常。而这个可调用的对象被用来处理请求以及生成响应。处理程序由框架的使用者编写,如上文所说的那样,应该返回字符串或者 Response 对象。Response 对象提供了一个友好的接口,因此这个简单的 if 语句保证了无论处理程序返回什么,代码最终都得到一个统一的 Response 对象。

接下来,被赋值给 self._writerStreamWriter 实例被调用,将字节字符串发送回客户端。函数返回前,程序在 await self._writer.drain() 处等待,以确保所有的数据被发送给客户端。只要缓存中还有未发送的数据,self._writer.close() 方法就不会执行。

HTTPConnection 类还有两个更加有趣的部分:一个用于关闭连接的方法,以及一组用来处理超时机制的方法。首先,关闭一条连接由下面这个小函数完成:

def close_connection(self):
    self._cancel_conn_timeout()
    self._writer.close()

每当一条连接将被关闭时,这段代码首先取消超时,然后把连接从事件循环中清除。

超时机制由三个相关的函数组成:第一个函数在超时后给客户端发送错误消息并关闭连接;第二个函数用于取消当前的超时;第三个函数调度超时功能。前两个函数比较简单,我将详细解释第三个函数 _reset_cpmm_timeout()

def _conn_timeout_close(self):
    self.error_reply(500, 'timeout')
    self.close_connection()

def _cancel_conn_timeout(self):
    if self._conn_timeout:
        self._conn_timeout.cancel()

def _reset_conn_timeout(self, timeout=TIMEOUT):
    self._cancel_conn_timeout()
    self._conn_timeout = self.loop.call_later(
        timeout, self._conn_timeout_close)

每当 _reset_conn_timeout 函数被调用时,它会先取消之前所有赋值给 self._conn_timeoutasyncio.Handle 对象。然后,使用 BaseEventLoop.call\_later 函数让 _conn_timeout_close 函数在超时数秒(timeout)后执行。如果你还记得 handle_request 函数的内容,就知道每当接收到数据时,这个函数就会被调用。这就取消了当前的超时并且重新安排 _conn_timeout_close 函数在超时数秒(timeout)后执行。只要接收到数据,这个循环就会不断地重置超时回调。如果在超时时间内没有接收到数据,最后函数 _conn_timeout_close 就会被调用。

创建连接

我们需要创建 HTTPConnection 对象,并且正确地使用它们。这一任务由 HTTPServer 类完成。HTTPServer 类是一个简单的容器,可以存储着一些配置信息(解析器,路由和事件循环实例),并使用这些配置来创建 HTTPConnection 实例:

class HTTPServer(object):
    def init(self, router, http_parser, loop):
        self.router = router
        self.http_parser = http_parser
        self.loop = loop

    async def handle_connection(self, reader, writer):
        connection = HTTPConnection(self, reader, writer)
        asyncio.ensure_future(connection.handle_request(), loop=self.loop)

HTTPServer 的每一个实例能够监听一个端口。它有一个 handle_connection 的异步方法来创建 HTTPConnection 的实例,并安排它们在事件循环中运行。这个方法被传递给 asyncio.start\_server 作为一个回调函数。也就是说,每当一个 TCP 连接初始化时(以 StreamReaderStreamWriter 为参数),它就会被调用。

   self._server = HTTPServer(self.router, self.http_parser, self.loop)
   self._connection_handler = asyncio.start_server(
        self._server.handle_connection,
        host=self.host,
        port=self.port,
        reuse_address=True,
        reuse_port=True,
        loop=self.loop)

这就是构成整个应用程序工作原理的核心:asyncio.start_server 接受 TCP 连接,然后在一个预配置的 HTTPServer 对象上调用一个方法。这个方法将处理一条 TCP 连接的所有逻辑:读取、解析、生成响应并发送回客户端、以及关闭连接。它的重点是 IO 逻辑、解析和生成响应。

讲解了核心的 IO 部分,让我们继续。

解析请求

这个微型框架的使用者被宠坏了,不愿意和字节打交道。它们想要一个更高层次的抽象 —— 一种更加简单的方法来处理请求。这个微型框架就包含了一个简单的 HTTP 解析器,能够将字节流转化为 Request 对象。

这些 Request 对象是像这样的容器:

class Request(object):
    def init(self):
        self.method = None
        self.path = None
        self.query_params = {}
        self.path_params = {}
        self.headers = {}
        self.body = None
        self.body_raw = None
        self.finished = False

它包含了所有需要的数据,可以用一种容易理解的方法从客户端接受数据。哦,不包括 cookie ,它对身份认证是非常重要的,我会将它留在第二部分。

每一个 HTTP 请求都包含了一些必需的内容,如请求路径和请求方法。它们也包含了一些可选的内容,如请求体、请求头,或是 URL 参数。随着 REST 的流行,除了 URL 参数,URL 本身会包含一些信息。比如,"/user/1/edit" 包含了用户的 id 。

一个请求的每个部分都必须被识别、解析,并正确地赋值给 Request 对象的对应属性。HTTP/1.1 是一个文本协议,事实上这简化了很多东西。(HTTP/2 是一个二进制协议,这又是另一种乐趣了)

解析器不需要跟踪状态,因此 http_parser 模块其实就是一组函数。调用函数需要用到 Request 对象,并将它连同一个包含原始请求信息的字节数组传递给 parse_into 函数。然后解析器会修改 Request 对象以及充当缓存的字节数组。字节数组的信息被逐渐地解析到 request 对象中。

http_parser 模块的核心功能就是下面这个 parse_into 函数:

def parse_into(request, buffer):
    _buffer = buffer[:]
    if not request.method and can_parse_request_line(_buffer):
        (request.method, request.path,
         request.query_params) = parse_request_line(_buffer)
        remove_request_line(_buffer)

    if not request.headers and can_parse_headers(_buffer):
        request.headers = parse_headers(_buffer)
        if not has_body(request.headers):
            request.finished = True

        remove_intro(_buffer)

    if not request.finished and can_parse_body(request.headers, _buffer):
        request.body_raw, request.body = parse_body(request.headers, _buffer)
        clear_buffer(_buffer)
        request.finished = True
    return _buffer

从上面的代码中可以看到,我把解析的过程分为三个部分:解析请求行(这行像这样:GET /resource HTTP/1.1),解析请求头以及解析请求体。

请求行包含了 HTTP 请求方法以及 URL 地址。而 URL 地址则包含了更多的信息:路径、url 参数和开发者自定义的 url 参数。解析请求方法和 URL 还是很容易的 - 合适地分割字符串就好了。函数 urlparse.parse 可以用来解析 URL 参数。开发者自定义的 URL 参数可以通过正则表达式来解析。

接下来是 HTTP 头部。它们是一行行由键值对组成的简单文本。问题在于,可能有多个 HTTP 头有相同的名字,却有不同的值。一个值得关注的 HTTP 头部是 Content-Length,它描述了请求体的字节长度(不是整个请求,仅仅是请求体)。这对于决定是否解析请求体有很重要的作用。

最后,解析器根据 HTTP 方法和头部来决定是否解析请求体。

路由!

在某种意义上,路由就像是连接框架和用户的桥梁,用户用合适的方法创建 Router 对象并为其设置路径/函数对,然后将它赋值给 App 对象。而 App 对象依次调用 get_handler 函数生成相应的回调函数。简单来说,路由就负责两件事,一是存储路径/函数对,二是返回需要的路径/函数对

Router 类中有两个允许最终开发者添加路由的方法,分别是 add_routesadd_route。因为 add_routes 就是 add_route 函数的一层封装,我们将主要讲解 add_route 函数:

def add_route(self, path, handler):
    compiled_route = self.class.build_route_regexp(path)
    if compiled_route not in self.routes:
        self.routes[compiled_route] = handler
    else:
        raise DuplicateRoute

首先,这个函数使用 Router.build_router_regexp 的类方法,将一条路由规则(如 '/cars/{id}' 这样的字符串),“编译”到一个已编译的正则表达式对象。这些已编译的正则表达式用来匹配请求路径,以及解析开发者自定义的 URL 参数。如果已经存在一个相同的路由,程序就会抛出一个异常。最后,这个路由/处理程序对被添加到一个简单的字典self.routes中。

下面展示 Router 是如何“编译”路由的:

@classmethod
def build_route_regexp(cls, regexp_str):
    """
    Turns a string into a compiled regular expression. Parses '{}' into
    named groups ie. '/path/{variable}' is turned into
    '/path/(?P<variable>[a-zA-Z0-9_-]+)'.

    :param regexp_str: a string representing a URL path.
    :return: a compiled regular expression.
    """
    def named_groups(matchobj):
        return '(?P<{0}>[a-zA-Z0-9_-]+)'.format(matchobj.group(1))

    re_str = re.sub(r'{([a-zA-Z0-9_-]+)}', named_groups, regexp_str)
    re_str = ''.join(('^', re_str, '$',))
    return re.compile(re_str)

这个方法使用正则表达式将所有出现的 {variable} 替换为 (?P<variable>)。然后在字符串头尾分别添加 ^$ 标记,最后编译正则表达式对象。

完成了路由存储仅成功了一半,下面是如何得到路由对应的函数:

def get_handler(self, path):
    logger.debug('Getting handler for: {0}'.format(path))
    for route, handler in self.routes.items():
        path_params = self.class.match_path(route, path)
        if path_params is not None:
            logger.debug('Got handler for: {0}'.format(path))
            wrapped_handler = HandlerWrapper(handler, path_params)
            return wrapped_handler

    raise NotFoundException()

一旦 App 对象获得一个 Request 对象,也就获得了 URL 的路径部分(如 /users/15/edit)。然后,我们需要匹配函数来生成一个响应或者 404 错误。get_handler 函数将路径作为参数,循环遍历路由,对每条路由调用 Router.match_path 类方法检查是否有已编译的正则对象与这个请求路径匹配。如果存在,我们就调用 HandleWrapper 来包装路由对应的函数。path_params 字典包含了路径变量(如 '/users/15/edit' 中的 '15'),若路由没有指定变量,字典就为空。最后,我们将包装好的函数返回给 App 对象。

如果遍历了所有的路由都找不到与路径匹配的,函数就会抛出 NotFoundException 异常。

这个 Route.match 类方法挺简单:

def match_path(cls, route, path):
    match = route.match(path)
    try:
        return match.groupdict()
    except AttributeError:
        return None

它使用正则对象的 match 方法来检查路由是否与路径匹配。若果不匹配,则返回 None 。

最后,我们有 HandleWraapper 类。它的唯一任务就是封装一个异步函数,存储 path_params 字典,并通过 handle 方法对外提供一个统一的接口。

class HandlerWrapper(object):
    def init(self, handler, path_params):
        self.handler = handler
        self.path_params = path_params
        self.request = None

    async def handle(self, request):
        return await self.handler(request, **self.path_params)

组合到一起

框架的最后部分就是用 App 类把所有的部分联系起来。

App 类用于集中所有的配置细节。一个 App 对象通过其 start_server 方法,使用一些配置数据创建一个 HTTPServer 的实例,然后将它传递给 asyncio.start\_server 函数asyncio.start_server 函数会对每一个 TCP 连接调用 HTTPServer 对象的 handle_connection 方法。

def start_server(self):
    if not self._server:
        self.loop = asyncio.get_event_loop()
        self._server = HTTPServer(self.router, self.http_parser, self.loop)
        self._connection_handler = asyncio.start_server(
            self._server.handle_connection,
            host=self.host,
            port=self.port,
            reuse_address=True,
            reuse_port=True,
            loop=self.loop)

        logger.info('Starting server on {0}:{1}'.format(
            self.host, self.port))
        self.loop.run_until_complete(self._connection_handler)

        try:
            self.loop.run_forever()
        except KeyboardInterrupt:
            logger.info('Got signal, killing server')
        except DiyFrameworkException as e:
            logger.error('Critical framework failure:')
            logger.error(e.traceback)
        finally:
            self.loop.close()
    else:
        logger.info('Server already started - {0}'.format(self))

总结

如果你查看源码,就会发现所有的代码仅 320 余行(包括测试代码的话共 540 余行)。这么少的代码实现了这么多的功能,让我有点惊讶。这个框架没有提供模板、身份认证以及数据库访问等功能(这些内容也很有趣哦)。这也让我知道,像 Django 和 Tornado 这样的框架是如何工作的,而且我能够快速地调试它们了。

这也是我按照测试驱动开发完成的第一个项目,整个过程有趣而有意义。先编写测试用例迫使我思考设计和架构,而不仅仅是把代码放到一起,让它们可以运行。不要误解我的意思,有很多时候,后者的方式更好。不过如果你想给确保这些不怎么维护的代码在之后的几周甚至几个月依然工作,那么测试驱动开发正是你需要的。

我研究了下整洁架构以及依赖注入模式,这些充分体现在 Router 类是如何作为一个更高层次的抽象的(实体?)。Router 类是比较接近核心的,像 http_parserApp 的内容比较边缘化,因为它们只是完成了极小的字符串和字节流、或是中层 IO 的工作。测试驱动开发(TDD)迫使我独立思考每个小部分,这使我问自己这样的问题:方法调用的组合是否易于理解?类名是否准确地反映了我正在解决的问题?我的代码中是否很容易区分出不同的抽象层?

来吧,写个小框架,真的很有趣:)


via: http://mattscodecave.com/posts/simple-python-framework-from-scratch.html

作者:Matt 译者:Cathon 校对:wxy

本文由 LCTT 原创编译,Linux中国 荣誉推出

(题图来自:es-static.us

你可能比较熟悉如何用 Github Pages 来分享你的工作,又或许你看过一堂教你建立你的第一个 Github Pages 网站的教程。近期 Github Pages 的改进使得从不同的数据源来发布您的网站更加的方便,其中的来源之一就是你的仓库的 /docs 目录。

文档的质量是一个软件项目健康发展的标志。对于开源项目来说,维护一个可靠而不出错的知识库、详细说明所有的细节是至关重要的。精心策划的文档可以让增加项目的亲切感,提供一步步的指导并促进各种方式的合作可以推动开源软件开发的协作进程。

在 Web 上托管你的文档是一个消耗时间的挑战,而且对于它的发布和维护也没有省事的办法,然而这是并非不可避免的。面对多种不同的发布工具,又是 FTP 服务器,又是数据库,文件以各种不同的方式存放在不同的位置下,而这些都需要你手动来调整。需要说明的是,传统的 Web 发布方式提供了无与伦比的灵活性和性能,但是在许多情况下,这是以牺牲简单易用为代价的。

当作为文档使用时,麻烦更少的方式显然更容易去维护。

GitHub Pages 可以以指定的方式为你的项目创建网站,这使得它天然地适合发布和维护文档。因为 Github Pages 支持 Jekyll,所以你可以使用纯文本或 Markdown 来书写你的文档,从而降低你维护的成本、减少维护时的障碍。Jekyll 还支持许多有用的工具比如变量、模板、以及自动代码高亮等等,它会给你更多的灵活性而不会增加复杂性,这些你在一些笨重的平台是见不到的。

最重要的是,在 Github 上使用 GitHub Pages 意味着你的文档和代码可以使用诸如 Issues 和 Pull Requests 来确保它得到应有的高水平维护,而且因为 GitHub Pages 允许您发布代码库主分支上的 /docs 目录,这样您就可以在同一分支同时维护你的代码库及其文档。

现在开始!

发布你的第一个文档页面只需要短短几分钟。

  1. 在你的仓库的主分支里创建一个 /docs/index.md 文件。
  2. 把你的内容以 Jekyll 格式添加进去,并提交你的修改。

  1. 查看你的仓库的设置分支然后选择主分支 /docs 目录,将其设置为 GitHub Pages 的源 ,点击保存,你就搞定了。

GitHub Pages 将会从你的 /docs 目录中读取内容,转换 index.md 为 HTML。然后把它发布到你的 GitHub Pages 的 URL 上。

这样将会创建并输出一个最基础的 HTML ,而且你可以使用 Jekyll 的自定义模板、CSS 和其他特性。如果想要看所有的可能,你可以看看 GitHub Pages Showcase


via: https://github.com/blog/2233-publish-your-project-documentation-with-github-pages

作者:loranallensmith 译者:Bestony 校对:wxy

本文由 LCTT 原创编译,Linux中国 荣誉推出

Allison 是 Dropbox 的工程师,在那里她维护着这个世界上最大的 Python 客户端网络之一。在去 Dropbox 之前,她是 Recurse Center 的协调人, 是这个位于纽约的程序员深造机构的作者。她在北美的 PyCon 做过关于 Python 内部机制的演讲,并且她喜欢研究奇怪的 bug。她的博客地址是 akaptur.com

介绍

Byterun 是一个用 Python 实现的 Python 解释器。随着我对 Byterun 的开发,我惊喜地的发现,这个 Python 解释器的基础结构用 500 行代码就能实现。在这一章我们会搞清楚这个解释器的结构,给你足够探索下去的背景知识。我们的目标不是向你展示解释器的每个细节---像编程和计算机科学其他有趣的领域一样,你可能会投入几年的时间去深入了解这个主题。

Byterun 是 Ned Batchelder 和我完成的,建立在 Paul Swartz 的工作之上。它的结构和主要的 Python 实现(CPython)差不多,所以理解 Byterun 会帮助你理解大多数解释器,特别是 CPython 解释器。(如果你不知道你用的是什么 Python,那么很可能它就是 CPython)。尽管 Byterun 很小,但它能执行大多数简单的 Python 程序(这一章是基于 Python 3.5 及其之前版本生成的字节码的,在 Python 3.6 中生成的字节码有一些改变)。

Python 解释器

在开始之前,让我们限定一下“Pyhton 解释器”的意思。在讨论 Python 的时候,“解释器”这个词可以用在很多不同的地方。有的时候解释器指的是 Python REPL,即当你在命令行下敲下 python 时所得到的交互式环境。有时候人们会或多或少的互换使用 “Python 解释器”和“Python”来说明从头到尾执行 Python 代码的这一过程。在本章中,“解释器”有一个更精确的意思:Python 程序的执行过程中的最后一步。

在解释器接手之前,Python 会执行其他 3 个步骤:词法分析,语法解析和编译。这三步合起来把源代码转换成 代码对象 code object ,它包含着解释器可以理解的指令。而解释器的工作就是解释代码对象中的指令。

你可能很奇怪执行 Python 代码会有编译这一步。Python 通常被称为解释型语言,就像 Ruby,Perl 一样,它们和像 C,Rust 这样的编译型语言相对。然而,这个术语并不是它看起来的那样精确。大多数解释型语言包括 Python 在内,确实会有编译这一步。而 Python 被称为解释型的原因是相对于编译型语言,它在编译这一步的工作相对较少(解释器做相对多的工作)。在这章后面你会看到,Python 的编译器比 C 语言编译器需要更少的关于程序行为的信息。

Python 的 Python 解释器

Byterun 是一个用 Python 写的 Python 解释器,这点可能让你感到奇怪,但没有比用 C 语言写 C 语言编译器更奇怪的了。(事实上,广泛使用的 gcc 编译器就是用 C 语言本身写的)你可以用几乎任何语言写一个 Python 解释器。

用 Python 写 Python 既有优点又有缺点。最大的缺点就是速度:用 Byterun 执行代码要比用 CPython 执行慢的多,CPython 解释器是用 C 语言实现的,并做了认真优化。然而 Byterun 是为了学习而设计的,所以速度对我们不重要。使用 Python 最大优势是我们可以仅仅实现解释器,而不用担心 Python 运行时部分,特别是对象系统。比如当 Byterun 需要创建一个类时,它就会回退到“真正”的 Python。另外一个优势是 Byterun 很容易理解,部分原因是它是用人们很容易理解的高级语言写的(Python !)(另外我们不会对解释器做优化 —— 再一次,清晰和简单比速度更重要)

构建一个解释器

在我们考察 Byterun 代码之前,我们需要从高层次对解释器结构有一些了解。Python 解释器是如何工作的?

Python 解释器是一个 虚拟机 virtual machine ,是一个模拟真实计算机的软件。我们这个虚拟机是 栈机器 stack machine ,它用几个栈来完成操作(与之相对的是 寄存器机器 register machine ,它从特定的内存地址读写数据)。

Python 解释器是一个 字节码解释器 bytecode interpreter :它的输入是一些称作 字节码 bytecode 的指令集。当你写 Python 代码时,词法分析器、语法解析器和编译器会生成 代码对象 code object 让解释器去操作。每个代码对象都包含一个要被执行的指令集 —— 它就是字节码 —— 以及还有一些解释器需要的信息。字节码是 Python 代码的一个 中间层表示 intermediate representation :它以一种解释器可以理解的方式来表示源代码。这和汇编语言作为 C 语言和机器语言的中间表示很类似。

微型解释器

为了让说明更具体,让我们从一个非常小的解释器开始。它只能计算两个数的和,只能理解三个指令。它执行的所有代码只是这三个指令的不同组合。下面就是这三个指令:

  • LOAD_VALUE
  • ADD_TWO_VALUES
  • PRINT_ANSWER

我们不关心词法、语法和编译,所以我们也不在乎这些指令集是如何产生的。你可以想象,当你写下 7 + 5,然后一个编译器为你生成那三个指令的组合。如果你有一个合适的编译器,你甚至可以用 Lisp 的语法来写,只要它能生成相同的指令。

假设

7 + 5

生成这样的指令集:

what_to_execute = {
    "instructions": [("LOAD_VALUE", 0),  # the first number
                     ("LOAD_VALUE", 1),  # the second number
                     ("ADD_TWO_VALUES", None),
                     ("PRINT_ANSWER", None)],
    "numbers": [7, 5] }

Python 解释器是一个 栈机器 stack machine ,所以它必须通过操作栈来完成这个加法(见下图)。解释器先执行第一条指令,LOAD_VALUE,把第一个数压到栈中。接着它把第二个数也压到栈中。然后,第三条指令,ADD_TWO_VALUES,先把两个数从栈中弹出,加起来,再把结果压入栈中。最后一步,把结果弹出并输出。

栈机器

LOAD_VALUE这条指令告诉解释器把一个数压入栈中,但指令本身并没有指明这个数是多少。指令需要一个额外的信息告诉解释器去哪里找到这个数。所以我们的指令集有两个部分:指令本身和一个常量列表。(在 Python 中,字节码就是我们所称的“指令”,而解释器“执行”的是代码对象。)

为什么不把数字直接嵌入指令之中?想象一下,如果我们加的不是数字,而是字符串。我们可不想把字符串这样的东西加到指令中,因为它可以有任意的长度。另外,我们这种设计也意味着我们只需要对象的一份拷贝,比如这个加法 7 + 7, 现在常量表 "numbers"只需包含一个[7]

你可能会想为什么会需要除了ADD_TWO_VALUES之外的指令。的确,对于我们两个数加法,这个例子是有点人为制作的意思。然而,这个指令却是建造更复杂程序的轮子。比如,就我们目前定义的三个指令,只要给出正确的指令组合,我们可以做三个数的加法,或者任意个数的加法。同时,栈提供了一个清晰的方法去跟踪解释器的状态,这为我们增长的复杂性提供了支持。

现在让我们来完成我们的解释器。解释器对象需要一个栈,它可以用一个列表来表示。它还需要一个方法来描述怎样执行每条指令。比如,LOAD_VALUE会把一个值压入栈中。

class Interpreter:
    def __init__(self):
        self.stack = []

    def LOAD_VALUE(self, number):
        self.stack.append(number)

    def PRINT_ANSWER(self):
        answer = self.stack.pop()
        print(answer)

    def ADD_TWO_VALUES(self):
        first_num = self.stack.pop()
        second_num = self.stack.pop()
        total = first_num + second_num
        self.stack.append(total)

这三个方法完成了解释器所理解的三条指令。但解释器还需要一样东西:一个能把所有东西结合在一起并执行的方法。这个方法就叫做 run_code,它把我们前面定义的字典结构 what-to-execute 作为参数,循环执行里面的每条指令,如果指令有参数就处理参数,然后调用解释器对象中相应的方法。

    def run_code(self, what_to_execute):
        instructions = what_to_execute["instructions"]
        numbers = what_to_execute["numbers"]
        for each_step in instructions:
            instruction, argument = each_step
            if instruction == "LOAD_VALUE":
                number = numbers[argument]
                self.LOAD_VALUE(number)
            elif instruction == "ADD_TWO_VALUES":
                self.ADD_TWO_VALUES()
            elif instruction == "PRINT_ANSWER":
                self.PRINT_ANSWER()

为了测试,我们创建一个解释器对象,然后用前面定义的 7 + 5 的指令集来调用 run_code

    interpreter = Interpreter()
    interpreter.run_code(what_to_execute)

显然,它会输出 12。

尽管我们的解释器功能十分受限,但这个过程几乎和真正的 Python 解释器处理加法是一样的。这里,我们还有几点要注意。

首先,一些指令需要参数。在真正的 Python 字节码当中,大概有一半的指令有参数。像我们的例子一样,参数和指令打包在一起。注意指令的参数和传递给对应方法的参数是不同的。

第二,指令ADD_TWO_VALUES不需要任何参数,它从解释器栈中弹出所需的值。这正是以基于栈的解释器的特点。

记得我们说过只要给出合适的指令集,不需要对解释器做任何改变,我们就能做多个数的加法。考虑下面的指令集,你觉得会发生什么?如果你有一个合适的编译器,什么代码才能编译出下面的指令集?

    what_to_execute = {
        "instructions": [("LOAD_VALUE", 0),
                         ("LOAD_VALUE", 1),
                         ("ADD_TWO_VALUES", None),
                         ("LOAD_VALUE", 2),
                         ("ADD_TWO_VALUES", None),
                         ("PRINT_ANSWER", None)],
        "numbers": [7, 5, 8] }

从这点出发,我们开始看到这种结构的可扩展性:我们可以通过向解释器对象增加方法来描述更多的操作(只要有一个编译器能为我们生成组织良好的指令集就行)。

变量

接下来给我们的解释器增加变量的支持。我们需要一个保存变量值的指令 STORE_NAME;一个取变量值的指令LOAD_NAME;和一个变量到值的映射关系。目前,我们会忽略命名空间和作用域,所以我们可以把变量和值的映射直接存储在解释器对象中。最后,我们要保证what_to_execute除了一个常量列表,还要有个变量名字的列表。

>>> def s():
...     a = 1
...     b = 2
...     print(a + b)
# a friendly compiler transforms `s` into:
    what_to_execute = {
        "instructions": [("LOAD_VALUE", 0),
                         ("STORE_NAME", 0),
                         ("LOAD_VALUE", 1),
                         ("STORE_NAME", 1),
                         ("LOAD_NAME", 0),
                         ("LOAD_NAME", 1),
                         ("ADD_TWO_VALUES", None),
                         ("PRINT_ANSWER", None)],
        "numbers": [1, 2],
        "names":   ["a", "b"] }

我们的新的实现在下面。为了跟踪哪个名字绑定到哪个值,我们在__init__方法中增加一个environment字典。我们也增加了STORE_NAMELOAD_NAME方法,它们获得变量名,然后从environment字典中设置或取出这个变量值。

现在指令的参数就有两个不同的意思,它可能是numbers列表的索引,也可能是names列表的索引。解释器通过检查所执行的指令就能知道是那种参数。而我们打破这种逻辑 ,把指令和它所用何种参数的映射关系放在另一个单独的方法中。

class Interpreter:
    def __init__(self):
        self.stack = []
        self.environment = {}

    def STORE_NAME(self, name):
        val = self.stack.pop()
        self.environment[name] = val

    def LOAD_NAME(self, name):
        val = self.environment[name]
        self.stack.append(val)

    def parse_argument(self, instruction, argument, what_to_execute):
        """ Understand what the argument to each instruction means."""
        numbers = ["LOAD_VALUE"]
        names = ["LOAD_NAME", "STORE_NAME"]

        if instruction in numbers:
            argument = what_to_execute["numbers"][argument]
        elif instruction in names:
            argument = what_to_execute["names"][argument]

        return argument

    def run_code(self, what_to_execute):
        instructions = what_to_execute["instructions"]
        for each_step in instructions:
            instruction, argument = each_step
            argument = self.parse_argument(instruction, argument, what_to_execute)

            if instruction == "LOAD_VALUE":
                self.LOAD_VALUE(argument)
            elif instruction == "ADD_TWO_VALUES":
                self.ADD_TWO_VALUES()
            elif instruction == "PRINT_ANSWER":
                self.PRINT_ANSWER()
            elif instruction == "STORE_NAME":
                self.STORE_NAME(argument)
            elif instruction == "LOAD_NAME":
                self.LOAD_NAME(argument)

仅仅五个指令,run_code这个方法已经开始变得冗长了。如果保持这种结构,那么每条指令都需要一个if分支。这里,我们要利用 Python 的动态方法查找。我们总会给一个称为FOO的指令定义一个名为FOO的方法,这样我们就可用 Python 的getattr函数在运行时动态查找方法,而不用这个大大的分支结构。run_code方法现在是这样:

    def execute(self, what_to_execute):
        instructions = what_to_execute["instructions"]
        for each_step in instructions:
            instruction, argument = each_step
            argument = self.parse_argument(instruction, argument, what_to_execute)
            bytecode_method = getattr(self, instruction)
            if argument is None:
                bytecode_method()
            else:
                bytecode_method(argument)

真实的 Python 字节码

现在,放弃我们的小指令集,去看看真正的 Python 字节码。字节码的结构和我们的小解释器的指令集差不多,除了字节码用一个字节而不是一个名字来代表这条指令。为了理解它的结构,我们将考察一个函数的字节码。考虑下面这个例子:

>>> def cond():
...     x = 3
...     if x < 5:
...         return 'yes'
...     else:
...         return 'no'
...

Python 在运行时会暴露一大批内部信息,并且我们可以通过 REPL 直接访问这些信息。对于函数对象condcond.__code__是与其关联的代码对象,而cond.__code__.co_code就是它的字节码。当你写 Python 代码时,你永远也不会想直接使用这些属性,但是这可以让我们做出各种恶作剧,同时也可以看看内部机制。

>>> cond.__code__.co_code  # the bytecode as raw bytes
b'd\x01\x00}\x00\x00|\x00\x00d\x02\x00k\x00\x00r\x16\x00d\x03\x00Sd\x04\x00Sd\x00
   \x00S'
>>> list(cond.__code__.co_code)  # the bytecode as numbers
[100, 1, 0, 125, 0, 0, 124, 0, 0, 100, 2, 0, 107, 0, 0, 114, 22, 0, 100, 3, 0, 83, 
 100, 4, 0, 83, 100, 0, 0, 83]

当我们直接输出这个字节码,它看起来完全无法理解 —— 唯一我们了解的是它是一串字节。很幸运,我们有一个很强大的工具可以用:Python 标准库中的dis模块。

dis是一个字节码反汇编器。反汇编器以为机器而写的底层代码作为输入,比如汇编代码和字节码,然后以人类可读的方式输出。当我们运行dis.dis,它输出每个字节码的解释。

>>> dis.dis(cond)
  2           0 LOAD_CONST               1 (3)
              3 STORE_FAST               0 (x)

  3           6 LOAD_FAST                0 (x)
              9 LOAD_CONST               2 (5)
             12 COMPARE_OP               0 (<)
             15 POP_JUMP_IF_FALSE       22

  4          18 LOAD_CONST               3 ('yes')
             21 RETURN_VALUE

  6     >>   22 LOAD_CONST               4 ('no')
             25 RETURN_VALUE
             26 LOAD_CONST               0 (None)
             29 RETURN_VALUE

这些都是什么意思?让我们以第一条指令LOAD_CONST为例子。第一列的数字(2)表示对应源代码的行数。第二列的数字是字节码的索引,告诉我们指令LOAD_CONST在位置 0 。第三列是指令本身对应的人类可读的名字。如果第四列存在,它表示指令的参数。如果第五列存在,它是一个关于参数是什么的提示。

考虑这个字节码的前几个字节:[100, 1, 0, 125, 0, 0]。这 6 个字节表示两条带参数的指令。我们可以使用dis.opname,一个字节到可读字符串的映射,来找到指令 100 和指令 125 代表的是什么:

>>> dis.opname[100]
'LOAD_CONST'
>>> dis.opname[125]
'STORE_FAST'

第二和第三个字节 —— 1 、0 ——是LOAD_CONST的参数,第五和第六个字节 —— 0、0 —— 是STORE_FAST的参数。就像我们前面的小例子,LOAD_CONST需要知道的到哪去找常量,STORE_FAST需要知道要存储的名字。(Python 的LOAD_CONST和我们小例子中的LOAD_VALUE一样,LOAD_FASTLOAD_NAME一样)。所以这六个字节代表第一行源代码x = 3 (为什么用两个字节表示指令的参数?如果 Python 使用一个字节,每个代码对象你只能有 256 个常量/名字,而用两个字节,就增加到了 256 的平方,65536个)。

条件语句与循环语句

到目前为止,我们的解释器只能一条接着一条的执行指令。这有个问题,我们经常会想多次执行某个指令,或者在特定的条件下跳过它们。为了可以写循环和分支结构,解释器必须能够在指令中跳转。在某种程度上,Python 在字节码中使用GOTO语句来处理循环和分支!让我们再看一个cond函数的反汇编结果:

>>> dis.dis(cond)
  2           0 LOAD_CONST               1 (3)
              3 STORE_FAST               0 (x)

  3           6 LOAD_FAST                0 (x)
              9 LOAD_CONST               2 (5)
             12 COMPARE_OP               0 (<)
             15 POP_JUMP_IF_FALSE       22

  4          18 LOAD_CONST               3 ('yes')
             21 RETURN_VALUE

  6     >>   22 LOAD_CONST               4 ('no')
             25 RETURN_VALUE
             26 LOAD_CONST               0 (None)
             29 RETURN_VALUE

第三行的条件表达式if x < 5被编译成四条指令:LOAD_FASTLOAD_CONSTCOMPARE_OPPOP_JUMP_IF_FALSEx < 5对应加载x、加载 5、比较这两个值。指令POP_JUMP_IF_FALSE完成这个if语句。这条指令把栈顶的值弹出,如果值为真,什么都不发生。如果值为假,解释器会跳转到另一条指令。

这条将被加载的指令称为跳转目标,它作为指令POP_JUMP的参数。这里,跳转目标是 22,索引为 22 的指令是LOAD_CONST,对应源码的第 6 行。(dis>>标记跳转目标。)如果X < 5为假,解释器会忽略第四行(return yes),直接跳转到第6行(return "no")。因此解释器通过跳转指令选择性的执行指令。

Python 的循环也依赖于跳转。在下面的字节码中,while x < 5这一行产生了和if x < 10几乎一样的字节码。在这两种情况下,解释器都是先执行比较,然后执行POP_JUMP_IF_FALSE来控制下一条执行哪个指令。第四行的最后一条字节码JUMP_ABSOLUT(循环体结束的地方),让解释器返回到循环开始的第 9 条指令处。当 x < 10变为假,POP_JUMP_IF_FALSE会让解释器跳到循环的终止处,第 34 条指令。

>>> def loop():
...      x = 1
...      while x < 5:
...          x = x + 1
...      return x
...
>>> dis.dis(loop)
  2           0 LOAD_CONST               1 (1)
              3 STORE_FAST               0 (x)

  3           6 SETUP_LOOP              26 (to 35)
        >>    9 LOAD_FAST                0 (x)
             12 LOAD_CONST               2 (5)
             15 COMPARE_OP               0 (<)
             18 POP_JUMP_IF_FALSE       34

  4          21 LOAD_FAST                0 (x)
             24 LOAD_CONST               1 (1)
             27 BINARY_ADD
             28 STORE_FAST               0 (x)
             31 JUMP_ABSOLUTE            9
        >>   34 POP_BLOCK

  5     >>   35 LOAD_FAST                0 (x)
             38 RETURN_VALUE

探索字节码

我希望你用dis.dis来试试你自己写的函数。一些有趣的问题值得探索:

  • 对解释器而言 for 循环和 while 循环有什么不同?
  • 能不能写出两个不同函数,却能产生相同的字节码?
  • elif是怎么工作的?列表推导呢?

到目前为止,我们已经知道了 Python 虚拟机是一个栈机器。它能顺序执行指令,在指令间跳转,压入或弹出栈值。但是这和我们期望的解释器还有一定距离。在前面的那个例子中,最后一条指令是RETURN_VALUE,它和return语句相对应。但是它返回到哪里去呢?

为了回答这个问题,我们必须再增加一层复杂性: frame 。一个帧是一些信息的集合和代码的执行上下文。帧在 Python 代码执行时动态地创建和销毁。每个帧对应函数的一次调用 —— 所以每个帧只有一个代码对象与之关联,而一个代码对象可以有多个帧。比如你有一个函数递归的调用自己 10 次,这会产生 11 个帧,每次调用对应一个,再加上启动模块对应的一个帧。总的来说,Python 程序的每个作用域都有一个帧,比如,模块、函数、类定义。

帧存在于 调用栈 call stack 中,一个和我们之前讨论的完全不同的栈。(你最熟悉的栈就是调用栈,就是你经常看到的异常回溯,每个以"File 'program.py'"开始的回溯对应一个帧。)解释器在执行字节码时操作的栈,我们叫它 数据栈 data stack 。其实还有第三个栈,叫做 块栈 block stack ,用于特定的控制流块,比如循环和异常处理。调用栈中的每个帧都有它自己的数据栈和块栈。

让我们用一个具体的例子来说明一下。假设 Python 解释器执行到下面标记为 3 的地方。解释器正处于foo函数的调用中,它接着调用bar。下面是帧调用栈、块栈和数据栈的示意图。我们感兴趣的是解释器先从最底下的foo()开始,接着执行foo的函数体,然后到达bar

>>> def bar(y):
...     z = y + 3     # <--- (3) ... and the interpreter is here.
...     return z
...
>>> def foo():
...     a = 1
...     b = 2
...     return a + bar(b) # <--- (2) ... which is returning a call to bar ...
...
>>> foo()             # <--- (1) We're in the middle of a call to foo ...
3

调用栈

现在,解释器处于bar函数的调用中。调用栈中有 3 个帧:一个对应于模块层,一个对应函数foo,另一个对应函数bar。(见上图)一旦bar返回,与它对应的帧就会从调用栈中弹出并丢弃。

字节码指令RETURN_VALUE告诉解释器在帧之间传递一个值。首先,它把位于调用栈栈顶的帧中的数据栈的栈顶值弹出。然后把整个帧弹出丢弃。最后把这个值压到下一个帧的数据栈中。

当 Ned Batchelder 和我在写 Byterun 时,很长一段时间我们的实现中一直有个重大的错误。我们整个虚拟机中只有一个数据栈,而不是每个帧都有一个。我们写了很多测试代码,同时在 Byterun 和真正的 Python 上运行,希望得到一致结果。我们几乎通过了所有测试,只有一样东西不能通过,那就是 生成器 generators 。最后,通过仔细的阅读 CPython 的源码,我们发现了错误所在(感谢 Michael Arntzenius 对这个 bug 的洞悉)。把数据栈移到每个帧就解决了这个问题。

回头在看看这个 bug,我惊讶的发现 Python 真的很少依赖于每个帧有一个数据栈这个特性。在 Python 中几乎所有的操作都会清空数据栈,所以所有的帧公用一个数据栈是没问题的。在上面的例子中,当bar执行完后,它的数据栈为空。即使foo公用这一个栈,它的值也不会受影响。然而,对应生成器,它的一个关键的特点是它能暂停一个帧的执行,返回到其他的帧,一段时间后它能返回到原来的帧,并以它离开时的相同状态继续执行。

Byterun

现在我们有足够的 Python 解释器的知识背景去考察 Byterun。

Byterun 中有四种对象。

  • VirtualMachine类,它管理高层结构,尤其是帧调用栈,并包含了指令到操作的映射。这是一个比前面Inteprter对象更复杂的版本。
  • Frame类,每个Frame类都有一个代码对象,并且管理着其他一些必要的状态位,尤其是全局和局部命名空间、指向调用它的整的指针和最后执行的字节码指令。
  • Function类,它被用来代替真正的 Python 函数。回想一下,调用函数时会创建一个新的帧。我们自己实现了Function,以便我们控制新的Frame的创建。
  • Block类,它只是包装了块的 3 个属性。(块的细节不是解释器的核心,我们不会花时间在它身上,把它列在这里,是因为 Byterun 需要它。)

VirtualMachine

每次程序运行时只会创建一个VirtualMachine实例,因为我们只有一个 Python 解释器。VirtualMachine 保存调用栈、异常状态、在帧之间传递的返回值。它的入口点是run_code方法,它以编译后的代码对象为参数,以创建一个帧为开始,然后运行这个帧。这个帧可能再创建出新的帧;调用栈随着程序的运行而增长和缩短。当第一个帧返回时,执行结束。

class VirtualMachineError(Exception):
    pass

class VirtualMachine(object):
    def __init__(self):
        self.frames = []   # The call stack of frames.
        self.frame = None  # The current frame.
        self.return_value = None
        self.last_exception = None

    def run_code(self, code, global_names=None, local_names=None):
        """ An entry point to execute code using the virtual machine."""
        frame = self.make_frame(code, global_names=global_names, 
                                local_names=local_names)
        self.run_frame(frame)

Frame

接下来,我们来写Frame对象。帧是一个属性的集合,它没有任何方法。前面提到过,这些属性包括由编译器生成的代码对象;局部、全局和内置命名空间;前一个帧的引用;一个数据栈;一个块栈;最后执行的指令指针。(对于内置命名空间我们需要多做一点工作,Python 在不同模块中对这个命名空间有不同的处理;但这个细节对我们的虚拟机不重要。)

class Frame(object):
    def __init__(self, code_obj, global_names, local_names, prev_frame):
        self.code_obj = code_obj
        self.global_names = global_names
        self.local_names = local_names
        self.prev_frame = prev_frame
        self.stack = []
        if prev_frame:
            self.builtin_names = prev_frame.builtin_names
        else:
            self.builtin_names = local_names['__builtins__']
            if hasattr(self.builtin_names, '__dict__'):
                self.builtin_names = self.builtin_names.__dict__

        self.last_instruction = 0
        self.block_stack = []

接着,我们在虚拟机中增加对帧的操作。这有 3 个帮助函数:一个创建新的帧的方法(它负责为新的帧找到名字空间),和压栈和出栈的方法。第四个函数,run_frame,完成执行帧的主要工作,待会我们再讨论这个方法。

class VirtualMachine(object):
    [... 删节 ...]

    # Frame manipulation
    def make_frame(self, code, callargs={}, global_names=None, local_names=None):
        if global_names is not None and local_names is not None:
            local_names = global_names
        elif self.frames:
            global_names = self.frame.global_names
            local_names = {}
        else:
            global_names = local_names = {
                '__builtins__': __builtins__,
                '__name__': '__main__',
                '__doc__': None,
                '__package__': None,
            }
        local_names.update(callargs)
        frame = Frame(code, global_names, local_names, self.frame)
        return frame

    def push_frame(self, frame):
        self.frames.append(frame)
        self.frame = frame

    def pop_frame(self):
        self.frames.pop()
        if self.frames:
            self.frame = self.frames[-1]
        else:
            self.frame = None

    def run_frame(self):
        pass
        # we'll come back to this shortly

Function

Function的实现有点曲折,但是大部分的细节对理解解释器不重要。重要的是当调用函数时 —— 即调用 __call__方法 —— 它创建一个新的Frame并运行它。

class Function(object):
    """
    Create a realistic function object, defining the things the interpreter expects.
    """
    __slots__ = [
        'func_code', 'func_name', 'func_defaults', 'func_globals',
        'func_locals', 'func_dict', 'func_closure',
        '__name__', '__dict__', '__doc__',
        '_vm', '_func',
    ]

    def __init__(self, name, code, globs, defaults, closure, vm):
        """You don't need to follow this closely to understand the interpreter."""
        self._vm = vm
        self.func_code = code
        self.func_name = self.__name__ = name or code.co_name
        self.func_defaults = tuple(defaults)
        self.func_globals = globs
        self.func_locals = self._vm.frame.f_locals
        self.__dict__ = {}
        self.func_closure = closure
        self.__doc__ = code.co_consts[0] if code.co_consts else None

        # Sometimes, we need a real Python function.  This is for that.
        kw = {
            'argdefs': self.func_defaults,
        }
        if closure:
            kw['closure'] = tuple(make_cell(0) for _ in closure)
        self._func = types.FunctionType(code, globs, **kw)

    def __call__(self, *args, **kwargs):
        """When calling a Function, make a new frame and run it."""
        callargs = inspect.getcallargs(self._func, *args, **kwargs)
        # Use callargs to provide a mapping of arguments: values to pass into the new 
        # frame.
        frame = self._vm.make_frame(
            self.func_code, callargs, self.func_globals, {}
        )
        return self._vm.run_frame(frame)

def make_cell(value):
    """Create a real Python closure and grab a cell."""
    # Thanks to Alex Gaynor for help with this bit of twistiness.
    fn = (lambda x: lambda: x)(value)
    return fn.__closure__[0]

接着,回到VirtualMachine对象,我们对数据栈的操作也增加一些帮助方法。字节码操作的栈总是在当前帧的数据栈。这些帮助函数让我们的POP_TOPLOAD_FAST以及其他操作栈的指令的实现可读性更高。

class VirtualMachine(object):
    [... 删节 ...]

    # Data stack manipulation
    def top(self):
        return self.frame.stack[-1]

    def pop(self):
        return self.frame.stack.pop()

    def push(self, *vals):
        self.frame.stack.extend(vals)

    def popn(self, n):
        """Pop a number of values from the value stack.
        A list of `n` values is returned, the deepest value first.
        """
        if n:
            ret = self.frame.stack[-n:]
            self.frame.stack[-n:] = []
            return ret
        else:
            return []

在我们运行帧之前,我们还需两个方法。

第一个方法,parse_byte_and_args 以一个字节码为输入,先检查它是否有参数,如果有,就解析它的参数。这个方法同时也更新帧的last_instruction属性,它指向最后执行的指令。一条没有参数的指令只有一个字节长度,而有参数的字节有3个字节长。参数的意义依赖于指令是什么。比如,前面说过,指令POP_JUMP_IF_FALSE,它的参数指的是跳转目标。BUILD_LIST,它的参数是列表的个数。LOAD_CONST,它的参数是常量的索引。

一些指令用简单的数字作为参数。对于另一些,虚拟机需要一点努力去发现它含意。标准库中的dis模块中有一个备忘单,它解释什么参数有什么意思,这让我们的代码更加简洁。比如,列表dis.hasname告诉我们LOAD_NAMEIMPORT_NAMELOAD_GLOBAL,以及另外的 9 个指令的参数都有同样的意义:对于这些指令,它们的参数代表了代码对象中的名字列表的索引。

class VirtualMachine(object):
    [... 删节 ...]

    def parse_byte_and_args(self):
        f = self.frame
        opoffset = f.last_instruction
        byteCode = f.code_obj.co_code[opoffset]
        f.last_instruction += 1
        byte_name = dis.opname[byteCode]
        if byteCode >= dis.HAVE_ARGUMENT:
            # index into the bytecode
            arg = f.code_obj.co_code[f.last_instruction:f.last_instruction+2]  
            f.last_instruction += 2   # advance the instruction pointer
            arg_val = arg[0] + (arg[1] * 256)
            if byteCode in dis.hasconst:   # Look up a constant
                arg = f.code_obj.co_consts[arg_val]
            elif byteCode in dis.hasname:  # Look up a name
                arg = f.code_obj.co_names[arg_val]
            elif byteCode in dis.haslocal: # Look up a local name
                arg = f.code_obj.co_varnames[arg_val]
            elif byteCode in dis.hasjrel:  # Calculate a relative jump
                arg = f.last_instruction + arg_val
            else:
                arg = arg_val
            argument = [arg]
        else:
            argument = []

        return byte_name, argument

下一个方法是dispatch,它查找给定的指令并执行相应的操作。在 CPython 中,这个分派函数用一个巨大的 switch 语句实现,有超过 1500 行的代码。幸运的是,我们用的是 Python,我们的代码会简洁的多。我们会为每一个字节码名字定义一个方法,然后用getattr来查找。就像我们前面的小解释器一样,如果一条指令叫做FOO_BAR,那么它对应的方法就是byte_FOO_BAR。现在,我们先把这些方法当做一个黑盒子。每个指令方法都会返回None或者一个字符串why,有些情况下虚拟机需要这个额外why信息。这些指令方法的返回值,仅作为解释器状态的内部指示,千万不要和执行帧的返回值相混淆。

class VirtualMachine(object):
    [... 删节 ...]

    def dispatch(self, byte_name, argument):
        """ Dispatch by bytename to the corresponding methods.
        Exceptions are caught and set on the virtual machine."""

        # When later unwinding the block stack,
        # we need to keep track of why we are doing it.
        why = None
        try:
            bytecode_fn = getattr(self, 'byte_%s' % byte_name, None)
            if bytecode_fn is None:
                if byte_name.startswith('UNARY_'):
                    self.unaryOperator(byte_name[6:])
                elif byte_name.startswith('BINARY_'):
                    self.binaryOperator(byte_name[7:])
                else:
                    raise VirtualMachineError(
                        "unsupported bytecode type: %s" % byte_name
                    )
            else:
                why = bytecode_fn(*argument)
        except:
            # deal with exceptions encountered while executing the op.
            self.last_exception = sys.exc_info()[:2] + (None,)
            why = 'exception'

        return why

    def run_frame(self, frame):
        """Run a frame until it returns (somehow).
        Exceptions are raised, the return value is returned.
        """
        self.push_frame(frame)
        while True:
            byte_name, arguments = self.parse_byte_and_args()

            why = self.dispatch(byte_name, arguments)

            # Deal with any block management we need to do
            while why and frame.block_stack:
                why = self.manage_block_stack(why)

            if why:
                break

        self.pop_frame()

        if why == 'exception':
            exc, val, tb = self.last_exception
            e = exc(val)
            e.__traceback__ = tb
            raise e

        return self.return_value

Block

在我们完成每个字节码方法前,我们简单的讨论一下块。一个块被用于某种控制流,特别是异常处理和循环。它负责保证当操作完成后数据栈处于正确的状态。比如,在一个循环中,一个特殊的迭代器会存在栈中,当循环完成时它从栈中弹出。解释器需要检查循环仍在继续还是已经停止。

为了跟踪这些额外的信息,解释器设置了一个标志来指示它的状态。我们用一个变量why实现这个标志,它可以是None或者是下面几个字符串之一:"continue""break""excption"return。它们指示对块栈和数据栈进行什么操作。回到我们迭代器的例子,如果块栈的栈顶是一个loop块,why的代码是continue,迭代器就应该保存在数据栈上,而如果whybreak,迭代器就会被弹出。

块操作的细节比这个还要繁琐,我们不会花时间在这上面,但是有兴趣的读者值得仔细的看看。

Block = collections.namedtuple("Block", "type, handler, stack_height")

class VirtualMachine(object):
    [... 删节 ...]

    # Block stack manipulation
    def push_block(self, b_type, handler=None):
        level = len(self.frame.stack)
        self.frame.block_stack.append(Block(b_type, handler, stack_height))

    def pop_block(self):
        return self.frame.block_stack.pop()

    def unwind_block(self, block):
        """Unwind the values on the data stack corresponding to a given block."""
        if block.type == 'except-handler':
            # The exception itself is on the stack as type, value, and traceback.
            offset = 3  
        else:
            offset = 0

        while len(self.frame.stack) > block.level + offset:
            self.pop()

        if block.type == 'except-handler':
            traceback, value, exctype = self.popn(3)
            self.last_exception = exctype, value, traceback

    def manage_block_stack(self, why):
        """ """
        frame = self.frame
        block = frame.block_stack[-1]
        if block.type == 'loop' and why == 'continue':
            self.jump(self.return_value)
            why = None
            return why

        self.pop_block()
        self.unwind_block(block)

        if block.type == 'loop' and why == 'break':
            why = None
            self.jump(block.handler)
            return why

        if (block.type in ['setup-except', 'finally'] and why == 'exception'):
            self.push_block('except-handler')
            exctype, value, tb = self.last_exception
            self.push(tb, value, exctype)
            self.push(tb, value, exctype) # yes, twice
            why = None
            self.jump(block.handler)
            return why

        elif block.type == 'finally':
            if why in ('return', 'continue'):
                self.push(self.return_value)

            self.push(why)

            why = None
            self.jump(block.handler)
            return why
        return why

指令

剩下了的就是完成那些指令方法了:byte_LOAD_FASTbyte_BINARY_MODULO等等。而这些指令的实现并不是很有趣,这里我们只展示了一小部分,完整的实现在 GitHub 上。(这里包括的指令足够执行我们前面所述的所有代码了。)

class VirtualMachine(object):
    [... 删节 ...]

    ## Stack manipulation

    def byte_LOAD_CONST(self, const):
        self.push(const)

    def byte_POP_TOP(self):
        self.pop()

    ## Names
    def byte_LOAD_NAME(self, name):
        frame = self.frame
        if name in frame.f_locals:
            val = frame.f_locals[name]
        elif name in frame.f_globals:
            val = frame.f_globals[name]
        elif name in frame.f_builtins:
            val = frame.f_builtins[name]
        else:
            raise NameError("name '%s' is not defined" % name)
        self.push(val)

    def byte_STORE_NAME(self, name):
        self.frame.f_locals[name] = self.pop()

    def byte_LOAD_FAST(self, name):
        if name in self.frame.f_locals:
            val = self.frame.f_locals[name]
        else:
            raise UnboundLocalError(
                "local variable '%s' referenced before assignment" % name
            )
        self.push(val)

    def byte_STORE_FAST(self, name):
        self.frame.f_locals[name] = self.pop()

    def byte_LOAD_GLOBAL(self, name):
        f = self.frame
        if name in f.f_globals:
            val = f.f_globals[name]
        elif name in f.f_builtins:
            val = f.f_builtins[name]
        else:
            raise NameError("global name '%s' is not defined" % name)
        self.push(val)

    ## Operators

    BINARY_OPERATORS = {
        'POWER':    pow,
        'MULTIPLY': operator.mul,
        'FLOOR_DIVIDE': operator.floordiv,
        'TRUE_DIVIDE':  operator.truediv,
        'MODULO':   operator.mod,
        'ADD':      operator.add,
        'SUBTRACT': operator.sub,
        'SUBSCR':   operator.getitem,
        'LSHIFT':   operator.lshift,
        'RSHIFT':   operator.rshift,
        'AND':      operator.and_,
        'XOR':      operator.xor,
        'OR':       operator.or_,
    }

    def binaryOperator(self, op):
        x, y = self.popn(2)
        self.push(self.BINARY_OPERATORS[op](x, y))

    COMPARE_OPERATORS = [
        operator.lt,
        operator.le,
        operator.eq,
        operator.ne,
        operator.gt,
        operator.ge,
        lambda x, y: x in y,
        lambda x, y: x not in y,
        lambda x, y: x is y,
        lambda x, y: x is not y,
        lambda x, y: issubclass(x, Exception) and issubclass(x, y),
    ]

    def byte_COMPARE_OP(self, opnum):
        x, y = self.popn(2)
        self.push(self.COMPARE_OPERATORS[opnum](x, y))

    ## Attributes and indexing

    def byte_LOAD_ATTR(self, attr):
        obj = self.pop()
        val = getattr(obj, attr)
        self.push(val)

    def byte_STORE_ATTR(self, name):
        val, obj = self.popn(2)
        setattr(obj, name, val)

    ## Building

    def byte_BUILD_LIST(self, count):
        elts = self.popn(count)
        self.push(elts)

    def byte_BUILD_MAP(self, size):
        self.push({})

    def byte_STORE_MAP(self):
        the_map, val, key = self.popn(3)
        the_map[key] = val
        self.push(the_map)

    def byte_LIST_APPEND(self, count):
        val = self.pop()
        the_list = self.frame.stack[-count] # peek
        the_list.append(val)

    ## Jumps

    def byte_JUMP_FORWARD(self, jump):
        self.jump(jump)

    def byte_JUMP_ABSOLUTE(self, jump):
        self.jump(jump)

    def byte_POP_JUMP_IF_TRUE(self, jump):
        val = self.pop()
        if val:
            self.jump(jump)

    def byte_POP_JUMP_IF_FALSE(self, jump):
        val = self.pop()
        if not val:
            self.jump(jump)

    ## Blocks

    def byte_SETUP_LOOP(self, dest):
        self.push_block('loop', dest)

    def byte_GET_ITER(self):
        self.push(iter(self.pop()))

    def byte_FOR_ITER(self, jump):
        iterobj = self.top()
        try:
            v = next(iterobj)
            self.push(v)
        except StopIteration:
            self.pop()
            self.jump(jump)

    def byte_BREAK_LOOP(self):
        return 'break'

    def byte_POP_BLOCK(self):
        self.pop_block()

    ## Functions

    def byte_MAKE_FUNCTION(self, argc):
        name = self.pop()
        code = self.pop()
        defaults = self.popn(argc)
        globs = self.frame.f_globals
        fn = Function(name, code, globs, defaults, None, self)
        self.push(fn)

    def byte_CALL_FUNCTION(self, arg):
        lenKw, lenPos = divmod(arg, 256) # KWargs not supported here
        posargs = self.popn(lenPos)

        func = self.pop()
        frame = self.frame
        retval = func(*posargs)
        self.push(retval)

    def byte_RETURN_VALUE(self):
        self.return_value = self.pop()
        return "return"

动态类型:编译器不知道它是什么

你可能听过 Python 是一种动态语言 —— 它是动态类型的。在我们建造解释器的过程中,已经透露出这样的信息。

动态的一个意思是很多工作是在运行时完成的。前面我们看到 Python 的编译器没有很多关于代码真正做什么的信息。举个例子,考虑下面这个简单的函数mod。它取两个参数,返回它们的模运算值。从它的字节码中,我们看到变量ab首先被加载,然后字节码BINAY_MODULO完成这个模运算。

>>> def mod(a, b):
...    return a % b
>>> dis.dis(mod)
  2           0 LOAD_FAST                0 (a)
              3 LOAD_FAST                1 (b)
              6 BINARY_MODULO
              7 RETURN_VALUE
>>> mod(19, 5)
4

计算 19 % 5 得4,—— 一点也不奇怪。如果我们用不同类的参数呢?

>>> mod("by%sde", "teco")
'bytecode'

刚才发生了什么?你可能在其它地方见过这样的语法,格式化字符串。

>>> print("by%sde" % "teco")
bytecode

用符号%去格式化字符串会调用字节码BUNARY_MODULO。它取栈顶的两个值求模,不管这两个值是字符串、数字或是你自己定义的类的实例。字节码在函数编译时生成(或者说,函数定义时)相同的字节码会用于不同类的参数。

Python 的编译器关于字节码的功能知道的很少,而取决于解释器来决定BINAYR_MODULO应用于什么类型的对象并完成正确的操作。这就是为什么 Python 被描述为 动态类型 dynamically typed :直到运行前你不必知道这个函数参数的类型。相反,在一个静态类型语言中,程序员需要告诉编译器参数的类型是什么(或者编译器自己推断出参数的类型。)

编译器的无知是优化 Python 的一个挑战 —— 只看字节码,而不真正运行它,你就不知道每条字节码在干什么!你可以定义一个类,实现__mod__方法,当你对这个类的实例使用%时,Python 就会自动调用这个方法。所以,BINARY_MODULO其实可以运行任何代码。

看看下面的代码,第一个a % b看起来没有用。

def mod(a,b):
    a % b
    return a %b

不幸的是,对这段代码进行静态分析 —— 不运行它 —— 不能确定第一个a % b没有做任何事。用 %调用__mod__可能会写一个文件,或是和程序的其他部分交互,或者其他任何可以在 Python 中完成的事。很难优化一个你不知道它会做什么的函数。在 Russell Power 和 Alex Rubinsteyn 的优秀论文中写道,“我们可以用多快的速度解释 Python?”,他们说,“在普遍缺乏类型信息下,每条指令必须被看作一个INVOKE_ARBITRARY_METHOD。”

总结

Byterun 是一个比 CPython 容易理解的简洁的 Python 解释器。Byterun 复制了 CPython 的主要结构:一个基于栈的解释器对称之为字节码的指令集进行操作,它们顺序执行或在指令间跳转,向栈中压入和从中弹出数据。解释器随着函数和生成器的调用和返回,动态的创建、销毁帧,并在帧之间跳转。Byterun 也有着和真正解释器一样的限制:因为 Python 使用动态类型,解释器必须在运行时决定指令的正确行为。

我鼓励你去反汇编你的程序,然后用 Byterun 来运行。你很快会发现这个缩短版的 Byterun 所没有实现的指令。完整的实现在 https://github.com/nedbat/byterun,或者你可以仔细阅读真正的 CPython 解释器ceval.c,你也可以实现自己的解释器!

致谢

感谢 Ned Batchelder 发起这个项目并引导我的贡献,感谢 Michael Arntzenius 帮助调试代码和这篇文章的修订,感谢 Leta Montopoli 的修订,以及感谢整个 Recurse Center 社区的支持和鼓励。所有的不足全是我自己没搞好。


via: http://aosabook.org/en/500L/a-python-interpreter-written-in-python.html

作者: Allison Kaptur 译者:qingyunha 校对:wxy

本文由 LCTT 原创翻译,Linux中国 荣誉推出

当你需要保存日期时间数据时,一个问题来了:你应该使用 MySQL 中的什么类型?使用 MySQL 原生的 DATE 类型还是使用 INT 字段把日期和时间保存为一个纯数字呢?

在这篇文章中,我将解释 MySQL 原生的方案,并给出一个最常用数据类型的对比表。我们也将对一些典型的查询做基准测试,然后得出在给定场景下应该使用什么数据类型的结论。

如果你想直接看结论,请翻到文章最下方。

原生的 MySQL Datetime 数据类型

Datetime 数据表示一个时间点。这可以用作日志记录、物联网时间戳、日历事件数据,等等。MySQL 有两种原生的类型可以将这种信息保存在单个字段中:Datetime 和 Timestamp。MySQL 文档中是这么介绍这些数据类型的:

DATETIME 类型用于保存同时包含日期和时间两部分的值。MySQL 以 'YYYY-MM-DD HH:MM:SS' 形式接收和显示 DATETIME 类型的值。

TIMESTAMP 类型用于保存同时包含日期和时间两部分的值。

DATETIME 或 TIMESTAMP 类型的值可以在尾部包含一个毫秒部分,精确度最高到微秒(6 位数)。

TIMESTAMP 和 DATETIME 数据类型提供自动初始化和更新到当前的日期和时间的功能,只需在列的定义中设置 DEFAULT CURRENTTIMESTAMP 和 ON UPDATE CURRENTTIMESTAMP。

作为一个例子:

CREATE TABLE `datetime_example` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `measured_on` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `measured_on` (`measured_on`)
) ENGINE=InnoDB;
CREATE TABLE `timestamp_example` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `measured_on` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `measured_on` (`measured_on`)
) ENGINE=InnoDB;

除了原生的日期时间表示方法,还有另一种常用的存储日期和时间信息的方法。即使用 INT 字段保存 Unix 时间(从1970 年 1 月 1 日协调世界时(UTC)建立所经过的秒数)。

MySQL 也提供了只保存时间信息中的一部分的方式,通过使用 Date、Year 或 Time 类型。由于这篇文章是关于保存准确时间点的最佳方式的,我们没有讨论这些不那么精确的局部类型。

使用 INT 类型保存 Unix 时间

使用一个简单的 INT 列保存 Unix 时间是最普通的方法。使用 INT,你可以确保你要保存的数字可以快速、可靠地插入到表中,就像这样:

INSERT INTO `vertabelo`.`sampletable`
(
 `id`,
 `measured_on` ### INT 类型的列
)
VALUES
(
 1,
 946684801
 ### 至 01/01/2000 @ 12:00am (UTC) 的 UNIX 时间戳 http://unixtimestamp.com
);

这就是关于它的所有内容了。它仅仅是个简单的 INT 列,MySQL 的处理方式是这样的:在内部使用 4 个字节保存那些数据。所以如果你在这个列上使用 SELECT 你将会得到一个数字。如果你想把这个列用作日期进行比较,下面的查询并不能正确工作:

SELECT
    id, measured_on, FROM_UNIXTIME(measured_on)
FROM
    vertabelo.inttimestampmeasures
WHERE
    measured_on > '2016-01-01' ### measured_on 会被作为字符串比较以进行查询
LIMIT 5;

这是因为 MySQL 把 INT 视为数字,而非日期。为了进行日期比较,你必须要么获取( LCTT 译注:从 1970-01-01 00:00:00)到 2016-01-01 经过的秒数,要么使用 MySQL 的 FROM\_UNIXTIME() 函数把 INT 列转为 Date 类型。下面的查询展示了 FROM\_UNIXTIME() 函数的用法:

SELECT
    id, measured_on, FROM_UNIXTIME(measured_on)
FROM
    vertabelo.inttimestampmeasures
WHERE
    FROM_UNIXTIME(measured_on) > '2016-01-01'
LIMIT 5;

这会正确地获取到日期在 2016-01-01 之后的记录。你也可以直接比较数字和 2016-01-01 的 Unix 时间戳表示形式,即 1451606400。这样做意味着不用使用任何特殊的函数,因为你是在直接比较数字。查询如下:

SELECT
    id, measured_on, FROM_UNIXTIME(measured_on)
FROM
    vertabelo.inttimestampmeasures
WHERE
   measured_on > 1451606400
LIMIT 5;

假如这种方式不够高效甚至提前做这种转换是不可行的话,那该怎么办?例如,你想获取 2016 年所有星期三的记录。要做到这样而不使用任何 MySQL 日期函数,你就不得不查出 2016 年每个星期三的开始和结束时间的 Unix 时间戳。然后你不得不写很大的查询,至少要在 WHERE 中包含 104 个比较。(2016 年有 52 个星期三,你不得不考虑一天的开始(0:00 am)和结束(11:59:59 pm)...)

结果是你很可能最终会使用 FROM\_UNIXTIME() 转换函数。既然如此,为什么不试下真正的日期类型呢?

使用 Datetime 和 Timestamp

Datetime 和 Timestamp 几乎以同样的方式工作。两种都保存日期和时间信息,毫秒部分最高精确度都是 6 位数。同时,使用人类可读的日期形式如 "2016-01-01" (为了便于比较)都能工作。查询时两种类型都支持“宽松格式”。宽松的语法允许任何标点符号作为分隔符。例如,"YYYY-MM-DD HH:MM:SS" 和 "YY-MM-DD HH:MM:SS" 两种形式都可以。在宽松格式情况下以下任何一种形式都能工作:

2012-12-31 11:30:45
2012^12^31 11+30+45
2012/12/31 11*30*45
2012@12@31 11^30^45

其它宽松格式也是允许的;你可以在 MySQL 参考手册 找到所有的格式。

默认情况下,Datetime 和 Timestamp 两种类型查询结果都以标准输出格式显示 —— 年-月-日 时:分:秒 (如 2016-01-01 23:59:59)。如果使用了毫秒部分,它们应该以小数值出现在秒后面 (如 2016-01-01 23:59:59.5)。

Timestamp 和 Datetime 的核心不同点主要在于 MySQL 在内部如何表示这些信息:两种都以二进制而非字符串形式存储,但在表示日期/时间部分时 Timestamp (4 字节) 比 Datetime (5 字节) 少使用 1 字节。当保存毫秒部分时两种都使用额外的空间 (1-3 字节)。如果你存储 150 万条记录,这种 1 字节的差异是微不足道的:

150 万条记录 * 每条记录 1 字节 / (1048576 字节/MB) = 1.43 MB

Timestamp 节省的 1 字节是有代价的:你只能存储从 '1970-01-01 00:00:01.000000' 到 '2038-01-19 03:14:07.999999' 之间的时间。而 Datetime 允许你存储从 '1000-01-01 00:00:00.000000' 到 '9999-12-31 23:59:59.999999' 之间的任何时间。

另一个重要的差别 —— 很多 MySQL 开发者没意识到的 —— 是 MySQL 使用服务器的时区转换 Timestamp 值到它的 UTC 等价值再保存。当获取值是它会再次进行时区转换,所以你得回了你“原始的”日期/时间值。有可能,下面这些情况会发生。

理想情况下,如果你一直使用同一个时区,MySQL 会获取到和你存储的同样的值。以我的经验,如果你的数据库涉及时区变换,你可能会遇到问题。例如,服务器变化(比如,你把数据库从都柏林的一台服务器迁移到加利福尼亚的一台服务器上,或者你只是修改了一下服务器的时区)时可能会发生这种情况。不管哪种方式,如果你获取数据时的时区是不同的,数据就会受影响。

Datetime 列不会被数据库改变。无论时区怎样配置,每次都会保存和获取到同样的值。就我而言,我认为这是一个更可靠的选择。

MySQL 文档:

MySQL 把 TIMESTAMP 值从当前的时区转换到 UTC 再存储,获取时再从 UTC 转回当前的时区。(其它类型如 DATETIME 不会这样,它们会“原样”保存。) 默认情况下,每个连接的当前时区都是服务器的时区。时区可以基于连接设置。只要时区设置保持一致,你就能得到和保存的相同的值。如果你保存了一个 TIMESTAMP 值,然后改变了时区再获取这个值,获取到的值和你存储的是不同的。这是因为在写入和查询的会话上没有使用同一个时区。当前时区可以通过系统变量 time\_zone 的值得到。更多信息,请查看 MySQL Server Time Zone Support

对比总结

在深入探讨使用各数据类型的性能差异之前,让我们先看一个总结表格以给你更多了解。每种类型的弱点以红色显示。

特性DatetimeTimestampInt (保存 Unix 时间)
原生时间表示否,所以大多数操作需要先使用转换函数,如 FROM\_UNIXTIME()
能保存毫秒是,最高 6 位精度是,最高 6 位精度
合法范围'1000-01-01 00:00:00.000000' 到 '9999-12-31 23:59:59.999999'1970-01-01 00:00:01.000000' 到 '2038-01-19 03:14:07.999999'若使用 unsigned, '1970-01-01 00:00:01.000000; 理论上最大到 '2106-2-07 06:28:15'
自动初始化(MySQL 5.6.5+)
宽松解释 (MySQL docs否,必须使用正确的格式
值被转换到 UTC 存储
可转换到其它类型是,如果值在合法的 Timestamp 范围中是,总是是,如果值在合法的范围中并使用转换函数
存储需求(MySQL 5.6.4+5 字节(如果使用了毫秒部分,再加最多 3 字节)4 字节 (如果使用了毫秒部分,再加最多 3 字节)4 字节 (不允许毫秒部分)
无需使用函数即可作为真实日期可读否,你必须格式化输出
数据分区是,使用 UNIX\_TIMESTAMP();在 MySQL 5.7 中其它表达式是不允许包含 TIMESTAMP 值的。同时,注意分区裁剪时的这些考虑是,使用 INT 上的任何合法操作

基准测试 INT、Timestamp 和 Datetime 的性能

为了比较这些类型的性能,我会使用我创建的一个天气预报网络的 150 万记录(准确说是 1,497,421)。这个网络每分钟都收集数据。为了让这些测试可复现,我已经删除了一些私有列,所以你可以使用这些数据运行你自己的测试。

基于我原始的表格,我创建了三个版本:

  • datetimemeasures 表在 measured_on 列使用 Datetime 类型,表示天气预报记录的测量时间
  • timestampmeasures 表在 measured_on 列使用 Timestamp 类型
  • inttimestampmeasures 表在 measured_on 列使用 INT (unsigned) 类型

这三个表拥有完全相同的数据;唯一的差别就是 measured_on 字段的类型。所有表都在 measured_on 列上设置了一个索引。

基准测试工具

为了评估这些数据类型的性能,我使用了两种方法。一种基于 Sysbench,它的官网是这么描述的:

... 一个模块化、跨平台和多线程的基准测试工具,用以评估那些对运行高负载数据库的系统非常重要的系统参数。

这个工具是 MySQL 文档中推荐的。

如果你使用 Windows (就像我),你可以下载一个包含可执行文件和我使用的测试查询的 zip 文件。它们基于 一种推荐的基准测试方法

为了执行一个给定的测试,你可以使用下面的命令(插入你自己的连接参数):

sysbench --MySQL-table-engine=innodb --MySQL-db=vertabelo --MySQL-user=root --MySQL-host=localhost --MySQL-password= --test=sysbench_test_file.lua --num-threads=8 --max-requests=100 run

这会正常工作,这里 sysbench_test_file.lua 是测试文件,并包含了各个测试中指向各个表的 SQL 查询。

为了进一步验证结果,我也运行了 mysqlslap。它的官网是这么描述的:

mysqlslap 是一个诊断程序,为模拟 MySQL 服务器的客户端负载并报告各个阶段的用时而设计。它工作起来就像是很多客户端在同时访问服务器。”

记得这些测试中最重要的不是所需的绝对时间。而是在不同数据类型上执行相同查询时的相对时间。这两个基准测试工具的测试时间不一定相同,因为不同工具的工作方式不同。重要的是数据类型的比较,随着我们深入到测试中,这将会变得清楚。

基准测试

我将使用三种可以评估几个性能方面的查询:

  • 时间范围选择

    • 在 Datetime 和 Timestamp 数据类型上这允许我们直接比较而不需要使用任何特殊的日期函数。
    • 同时,我们可以评估在 INT 类型的列上使用日期函数相对于使用简单的数值比较的影响。为了做到这些我们需要把范围转换为 Unix 时间戳数值。
  • 日期函数选择

    • 与前个测试中比较操作针对一个简单的 DATE 值相反,这个测试使得我们可以评估使用日期函数作为 “WHERE” 子句的一部分的性能。
    • 我们还可以测试一个场景,即我们必须使用一个函数将 INT 列转换为一个合法的 DATE 类型然后执行查询。
  • count() 查询

    • 作为对前面测试的补充,这将评估在三种不同的表示类型上进行典型的统计查询的性能。

我们将在这些测试中覆盖一些常见的场景,并看到三种类型上的性能表现。

关于 SQL\_NO\_CACHE

当在查询中使用 SQL\_NO\_CACHE 时,服务器不使用查询缓存。它既不检查查询缓存以确认结果是不是已经在那儿了,也不会保存查询结果。因此,每个查询将反映真实的性能影响,就像每次查询都是第一次被调用。

测试 1:选择一个日期范围中的值

这个查询返回总计 1,497,421 行记录中的 75,706 行。

查询 1 和 Datetime:

SELECT SQL_NO_CACHE
    measured_on
FROM
    vertabelo.datetimemeasures m
WHERE
    m.measured_on > '2016-01-01 00:00:00.0'
      AND m.measured_on < '2016-02-01 00:00:00.0';

性能

响应时间 (ms)Sysbenchmysqlslap
最小152296
最大12613203
平均362809
Sysbench cmd> sysbench --MySQL-table-engine=innodb --MySQL-db=vertabelo --MySQL-user=root --MySQL-host=localhost --MySQL-password= --test=datetime.lua --num-threads=8 --max-requests=100 run
mysqlslap cmd> mysqlslap --query="SELECT SQL_NO_CACHE measured_on FROM vertabelo.datetimemeasures m WHERE m.measured_on > '2016-01-01 00:00:00.0' AND m.measured_on < '2016-02-01 00:00:00.0'" --host=localhost --user=root --concurrency=8 --iterations=100  --no-drop --create-schema=vertabelo

查询 1 和 Timestamp:

SELECT SQL_NO_CACHE
    measured_on
FROM
    vertabelo.timestampmeasures m
WHERE
    m.measured_on > '2016-01-01 00:00:00.0'
        AND m.measured_on < '2016-02-01 00:00:00.0';

性能

响应时间 (ms)Sysbenchmysqlslap
最小214359
最大13893313
平均4311004
Sysbench cmd> sysbench --MySQL-table-engine=innodb --MySQL-db=vertabelo --MySQL-user=root --MySQL-host=localhost --MySQL-password= --test=timestamp.lua --num-threads=8 --max-requests=100 run
mysqlslap cmd> mysqlslap --query="SELECT SQL_NO_CACHE measured_on FROM vertabelo.timestampmeasures m WHERE m.measured_on > '2016-01-01 00:00:00.0' AND m.measured_on < '2016-02-01 00:00:00.0'" --host=localhost --user=root --concurrency=8 --iterations=100  --no-drop --create-schema=vertabelo

查询 1 和 INT:

SELECT SQL_NO_CACHE
    measured_on
FROM
    vertabelo.inttimestampmeasures m
WHERE
    FROM_UNIXTIME(m.measured_on) > '2016-01-01 00:00:00.0'
        AND FROM_UNIXTIME(m.measured_on) < '2016-02-01 00:00:00.0';

性能

响应时间 (ms)Sysbenchmysqlslap
最小24727968
最大655410312
平均41078527
Sysbench cmd> sysbench --MySQL-table-engine=innodb --MySQL-db=vertabelo --MySQL-user=root --MySQL-host=localhost --MySQL-password= --test=int.lua --num-threads=8 --max-requests=100 run
mysqlslap cmd> mysqlslap --query="SELECT SQL_NO_CACHE measured_on FROM vertabelo.inttimestampmeasures m WHERE FROM_UNIXTIME(m.measured_on) > '2016-01-01 00:00:00.0' AND FROM_UNIXTIME(m.measured_on) < '2016-02-01 00:00:00.0'" --host=localhost --user=root --concurrency=8 --iterations=100  --no-drop --create-schema=vertabelo

另一种 INT 上的查询 1:

由于这是个相当直接的范围搜索,而且查询中的日期可以轻易地转为简单的数值比较,我将它包含在了这个测试中。结果证明这是最快的方法 (你大概已经预料到了),因为它仅仅是比较数字而没有使用任何日期转换函数:

SELECT SQL_NO_CACHE
    measured_on
FROM
    vertabelo.inttimestampmeasures m
WHERE
    m.measured_on > 1451617200
        AND m.measured_on < 1454295600;

性能

响应时间 (ms)Sysbenchmysqlslap
最小88171
最大2752157
平均165514
Sysbench cmd> sysbench --MySQL-table-engine=innodb --MySQL-db=vertabelo --MySQL-user=root --MySQL-host=localhost --MySQL-password= --test=basic_int.lua --num-threads=8 --max-requests=100 run
mysqlslap cmd> mysqlslap --query="SELECT SQL_NO_CACHE measured_on FROM vertabelo.inttimestampmeasures m WHERE m.measured_on > 1451617200 AND m.measured_on < 1454295600" --host=localhost --user=root --concurrency=8 --iterations=100  --no-drop --create-schema=vertabelo

测试 1 总结

平均响应时间 (ms)Sysbench相对于 Datetime 的速度mysqlslap相对于 Datetime 的速度
Datetime362-809-
Timestamp431慢 19%1004慢 24%
INT4107慢 1134%8527慢 1054%
另一种 INT 查询165快 55%514快 36%

两种基准测试工具都显示 Datetime 比 Timestamp 和 INT 更快。但 Datetime 没有我们在另一种 INT 查询中使用的简单数值比较快。

测试 2:选择星期一产生的记录

这个查询返回总计 1,497,421 行记录中的 221,850 行。

查询 2 和 Datetime:

SELECT SQL_NO_CACHE measured_on
FROM
    vertabelo.datetimemeasures m
WHERE
    WEEKDAY(m.measured_on) = 0; # MONDAY

性能

响应时间 (ms)Sysbenchmysqlslap
最小18744343
最大61687797
平均31276103
Sysbench cmd> sysbench --MySQL-table-engine=innodb --MySQL-db=vertabelo --MySQL-user=root --MySQL-host=localhost --MySQL-password= --test=datetime_1.lua --num-threads=8 --max-requests=100 run
mysqlslap cmd> mysqlslap --query="SELECT SQL_NO_CACHE measured_on FROM vertabelo.datetimemeasures m WHERE WEEKDAY(m.measured_on) = 0" --host=localhost --user=root --concurrency=8 --iterations=25 --no-drop --create-schema=vertabelo

查询 2 和 Timestamp:

SELECT SQL_NO_CACHE
    measured_on
FROM
    vertabelo.timestampmeasures m
WHERE
    WEEKDAY(m.measured_on) = 0; # MONDAY

性能

响应时间 (ms)Sysbenchmysqlslap
最小26885953
最大666613531
平均36538412
Sysbench cmd> sysbench --MySQL-table-engine=innodb --MySQL-db=vertabelo --MySQL-user=root --MySQL-host=localhost --MySQL-password= --test=timestamp_1.lua --num-threads=8 --max-requests=100 run
mysqlslap cmd> mysqlslap --query="SELECT SQL_NO_CACHE measured_on FROM vertabelo.timestampmeasures m WHERE WEEKDAY(m.measured_on) = 0" --host=localhost --user=root --concurrency=8 --iterations=25 --no-drop --create-schema=vertabelo

查询 2 和 INT:

SELECT SQL_NO_CACHE
    measured_on
FROM
    vertabelo.inttimestampmeasures m
WHERE
    WEEKDAY(FROM_UNIXTIME(m.measured_on)) = 0; # MONDAY

性能

响应时间 (ms)Sysbenchmysqlslap
最小20515844
最大700710469
平均34868088
Sysbench cmd> sysbench --MySQL-table-engine=innodb --MySQL-db=vertabelo --MySQL-user=root --MySQL-host=localhost --MySQL-password= --test=int_1.lua --num-threads=8 --max-requests=100 run
mysqlslap cmd> mysqlslap --query="SELECT SQL_NO_CACHE measured_on FROM vertabelo.inttimestampmeasures m WHERE WEEKDAY(FROM_UNIXTIME(m.measured_on)) = 0" --host=localhost --user=root --concurrency=8 --iterations=25  --no-drop --create-schema=vertabelo

测试 2 总结

平均响应时间 (ms)Sysbench相对于 Datetime 的速度mysqlslap相对于 Datetime 的速度
Datetime3127-6103-
Timestamp3653慢 17%8412慢 38%
INT3486慢 11%8088慢 32%

再次,在两个基准测试工具中 Datetime 比 Timestamp 和 INT 快。但在这个测试中,INT 查询 —— 即使它使用了一个函数以转换日期 —— 比 Timestamp 查询更快得到结果。

测试 3:选择星期一产生的记录总数

这个查询返回一行,包含产生于星期一的所有记录的总数(从总共 1,497,421 行可用记录中)。

查询 3 和 Datetime:

SELECT SQL_NO_CACHE
    COUNT(measured_on)
FROM
    vertabelo.datetimemeasures m
WHERE
    WEEKDAY(m.measured_on) = 0; # MONDAY

性能

响应时间 (ms)Sysbenchmysqlslap
最小17204063
最大45947812
平均27975540
Sysbench cmd> sysbench --MySQL-table-engine=innodb --MySQL-db=vertabelo --MySQL-user=root --MySQL-host=localhost --MySQL-password= --test=datetime_1_count.lua --num-threads=8 --max-requests=100 run
mysqlslap cmd> mysqlslap --query="SELECT SQL_NO_CACHE COUNT(measured_on) FROM vertabelo.datetimemeasures m WHERE WEEKDAY(m.measured_on) = 0" --host=localhost --user=root --concurrency=8 --iterations=25 --no-drop --create-schema=vertabelo

查询 3 和 Timestamp:

SELECT SQL_NO_CACHE
    COUNT(measured_on)
FROM
    vertabelo.timestampmeasures m
WHERE
    WEEKDAY(m.measured_on) = 0; # MONDAY

性能

响应时间 (ms)Sysbenchmysqlslap
最小19074578
最大543710235
平均34087102
Sysbench cmd> sysbench --MySQL-table-engine=innodb --MySQL-db=vertabelo --MySQL-user=root --MySQL-host=localhost --MySQL-password= --test=timestamp_1_count.lua --num-threads=8 --max-requests=100 run
mysqlslap cmd> mysqlslap --query="SELECT SQL_NO_CACHE COUNT(measured_on) FROM vertabelo.timestampmeasures m WHERE WEEKDAY(m.measured_on) = 0" --host=localhost --user=root --concurrency=8 --iterations=25 --no-drop --create-schema=vertabelo

查询 3 和 INT:

SELECT SQL_NO_CACHE
    COUNT(measured_on)
FROM
    vertabelo.inttimestampmeasures m
WHERE
    WEEKDAY(FROM_UNIXTIME(m.measured_on)) = 0; # MONDAY

性能

响应时间 (ms)Sysbenchmysqlslap
最小21085609
最大47649735
平均33077416
Sysbench cmd> sysbench --MySQL-table-engine=innodb --MySQL-db=vertabelo --MySQL-user=root --MySQL-host=localhost --MySQL-password= --test=int_1_count.lua --num-threads=8 --max-requests=100 run
mysqlslap cmd> mysqlslap --query="SELECT SQL_NO_CACHE COUNT(measured_on) FROM vertabelo.inttimestampmeasures m WHERE WEEKDAY(FROM_UNIXTIME(m.measured_on)) = 0" --host=localhost --user=root --concurrency=8 --iterations=25  --no-drop --create-schema=vertabelo

测试 3 总结

平均响应时间 (ms)Sysbench相对于 Datetime 的速度mysqlslap相对于 Datetime 的速度
Datetime2797-5540-
Timestamp3408慢 22%7102慢 28%
INT3307慢 18%7416慢 33%

再一次,两个基准测试工具都显示 Datetime 比 Timestamp 和 INT 快。不能判断 INT 是否比 Timestamp 快,因为 mysqlslap 显示 INT 比 Timestamp 略快而 Sysbench 却相反。

注意: 所有测试都是在一台 Windows 10 机器上本地运行的,这台机器拥有一个双核 i7 CPU,16GB 内存,运行 MariaDB v10.1.9,使用 innoDB 引擎。

结论

基于这些数据,我确信 Datetime 是大多数场景下的最佳选择。原因是:

  • 更快(根据我们的三个基准测试)。
  • 无需任何转换即是人类可读的。
  • 不会因为时区变换产生问题。
  • 只比它的对手们多用 1 字节
  • 支持更大的日期范围(从 1000 年到 9999 年)

如果你只是存储 Unix 时间戳(并且在它的合法日期范围内),而且你真的不打算在它上面使用任何基于日期的查询,我觉得使用 INT 是可以的。我们已经看到,它执行简单数值比较查询时非常快,因为只是在处理简单的数字。

Timestamp 怎么样呢?如果 Datetime 相对于 Timestamp 的优势不适用于你特殊的场景,你最好使用时间戳。阅读这篇文章后,你对三种类型间的区别应该有了更好的理解,可以根据你的需要做出最佳的选择。


via: http://www.vertabelo.com/blog/technical-articles/what-datatype-should-you-use-to-represent-time-in-mysql-we-compare-datetime-timestamp-and-int

作者:Francisco Claria 译者:bianjp 校对:wxy

本文由 LCTT 原创编译,Linux中国 荣誉推出

本文讲述的是 Python 中 Mock 的使用。

如何执行单元测试而不用考验你的耐心

很多时候,我们编写的软件会直接与那些被标记为“垃圾”的服务交互。用外行人的话说:服务对我们的应用程序很重要,但是我们想要的是交互,而不是那些不想要的副作用,这里的“不想要”是在自动化测试运行的语境中说的。例如:我们正在写一个社交 app,并且想要测试一下 "发布到 Facebook" 的新功能,但是不想每次运行测试集的时候真的发布到 Facebook。

Python 的 unittest 库包含了一个名为 unittest.mock 或者可以称之为依赖的子包,简称为 mock —— 其提供了极其强大和有用的方法,通过它们可以 模拟 mock 并去除那些我们不希望的副作用。

注意:mock 最近被收录到了 Python 3.3 的标准库中;先前发布的版本必须通过 PyPI 下载 Mock 库。

恐惧系统调用

再举另一个例子,我们在接下来的部分都会用到它,这是就是系统调用。不难发现,这些系统调用都是主要的模拟对象:无论你是正在写一个可以弹出 CD 驱动器的脚本,还是一个用来删除 /tmp 下过期的缓存文件的 Web 服务,或者一个绑定到 TCP 端口的 socket 服务器,这些调用都是在你的单元测试上下文中不希望产生的副作用。

作为一个开发者,你需要更关心你的库是否成功地调用了一个可以弹出 CD 的系统函数(使用了正确的参数等等),而不是切身经历 CD 托盘每次在测试执行的时候都打开了。(或者更糟糕的是,弹出了很多次,在一个单元测试运行期间多个测试都引用了弹出代码!)

同样,保持单元测试的效率和性能意味着需要让如此多的“缓慢代码”远离自动测试,比如文件系统和网络访问。

对于第一个例子来说,我们要从原始形式换成使用 mock 重构一个标准 Python 测试用例。我们会演示如何使用 mock 写一个测试用例,使我们的测试更加智能、快速,并展示更多关于我们软件的工作原理。

一个简单的删除函数

我们都有过需要从文件系统中一遍又一遍的删除文件的时候,因此,让我们在 Python 中写一个可以使我们的脚本更加轻易完成此功能的函数。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os

def rm(filename):
    os.remove(filename)

很明显,我们的 rm 方法此时无法提供比 os.remove 方法更多的相关功能,但我们可以在这里添加更多的功能,使我们的基础代码逐步改善。

让我们写一个传统的测试用例,即,没有使用 mock

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from mymodule import rm

import os.path
import tempfile
import unittest

class RmTestCase(unittest.TestCase):

    tmpfilepath = os.path.join(tempfile.gettempdir(), "tmp-testfile")

    def setUp(self):
        with open(self.tmpfilepath, "wb") as f:
            f.write("Delete me!")

    def test_rm(self):
        # remove the file
        rm(self.tmpfilepath)
        # test that it was actually removed
        self.assertFalse(os.path.isfile(self.tmpfilepath), "Failed to remove the file.")

我们的测试用例相当简单,但是在它每次运行的时候,它都会创建一个临时文件并且随后删除。此外,我们没有办法测试我们的 rm 方法是否正确地将我们的参数向下传递给 os.remove 调用。我们可以基于以上的测试认为它做到了,但还有很多需要改进的地方。

使用 Mock 重构

让我们使用 mock 重构我们的测试用例:

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from mymodule import rm

import mock
import unittest

class RmTestCase(unittest.TestCase):

    @mock.patch('mymodule.os')
    def test_rm(self, mock_os):
        rm("any path")
        # test that rm called os.remove with the right parameters
        mock_os.remove.assert_called_with("any path")

使用这些重构,我们从根本上改变了测试用例的操作方式。现在,我们有一个可以用于验证其他功能的内部对象。

潜在陷阱

第一件需要注意的事情就是,我们使用了 mock.patch 方法装饰器,用于模拟位于 mymodule.os 的对象,并且将 mock 注入到我们的测试用例方法。那么只是模拟 os 本身,而不是 mymodule.osos 的引用(LCTT 译注:注意 @mock.patch('mymodule.os') 便是模拟 mymodule.os 下的 os),会不会更有意义呢?

当然,当涉及到导入和管理模块,Python 的用法就像蛇一样灵活。在运行时,mymodule 模块有它自己的被导入到本模块局部作用域的 os。因此,如果我们模拟 os,我们是看不到 mock 在 mymodule 模块中的模仿作用的。

这句话需要深刻地记住:

模拟一个东西要看它用在何处,而不是来自哪里。

如果你需要为 myproject.app.MyElaborateClass 模拟 tempfile 模块,你可能需要将 mock 用于 myproject.app.tempfile,而其他模块保持自己的导入。

先将那个陷阱放一边,让我们继续模拟。

向 ‘rm’ 中加入验证

之前定义的 rm 方法相当的简单。在盲目地删除之前,我们倾向于验证一个路径是否存在,并验证其是否是一个文件。让我们重构 rm 使其变得更加智能:

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import os.path

def rm(filename):
    if os.path.isfile(filename):
        os.remove(filename)

很好。现在,让我们调整测试用例来保持测试的覆盖率。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from mymodule import rm

import mock
import unittest

class RmTestCase(unittest.TestCase):

    @mock.patch('mymodule.os.path')
    @mock.patch('mymodule.os')
    def test_rm(self, mock_os, mock_path):
        # set up the mock
        mock_path.isfile.return_value = False

        rm("any path")

        # test that the remove call was NOT called.
        self.assertFalse(mock_os.remove.called, "Failed to not remove the file if not present.")

        # make the file 'exist'
        mock_path.isfile.return_value = True

        rm("any path")

        mock_os.remove.assert_called_with("any path")

我们的测试用例完全改变了。现在我们可以在没有任何副作用的情况下核实并验证方法的内部功能。

将文件删除作为服务

到目前为止,我们只是将 mock 应用在函数上,并没应用在需要传递参数的对象和实例的方法上。我们现在开始涵盖对象的方法。

首先,我们将 rm 方法重构成一个服务类。实际上将这样一个简单的函数转换成一个对象,在本质上这不是一个合理的需求,但它能够帮助我们了解 mock 的关键概念。让我们开始重构:

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import os.path

class RemovalService(object):
    """A service for removing objects from the filesystem."""

    def rm(filename):
        if os.path.isfile(filename):
            os.remove(filename)

你会注意到我们的测试用例没有太大变化:

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from mymodule import RemovalService

import mock
import unittest

class RemovalServiceTestCase(unittest.TestCase):

    @mock.patch('mymodule.os.path')
    @mock.patch('mymodule.os')
    def test_rm(self, mock_os, mock_path):
        # instantiate our service
        reference = RemovalService()

        # set up the mock
        mock_path.isfile.return_value = False

        reference.rm("any path")

        # test that the remove call was NOT called.
        self.assertFalse(mock_os.remove.called, "Failed to not remove the file if not present.")

        # make the file 'exist'
        mock_path.isfile.return_value = True

        reference.rm("any path")

        mock_os.remove.assert_called_with("any path")

很好,我们知道 RemovalService 会如预期般的工作。接下来让我们创建另一个服务,将 RemovalService 声明为它的一个依赖:

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import os.path

class RemovalService(object):
    """A service for removing objects from the filesystem."""

    def rm(self, filename):
        if os.path.isfile(filename):
            os.remove(filename)


class UploadService(object):

    def __init__(self, removal_service):
        self.removal_service = removal_service

    def upload_complete(self, filename):
        self.removal_service.rm(filename)

因为我们的测试覆盖了 RemovalService,因此我们不会对我们测试用例中 UploadService 的内部函数 rm 进行验证。相反,我们将调用 UploadServiceRemovalService.rm 方法来进行简单测试(当然没有其他副作用),我们通过之前的测试用例便能知道它可以正确地工作。

这里有两种方法来实现测试:

  1. 模拟 RemovalService.rm 方法本身。
  2. 在 UploadService 的构造函数中提供一个模拟实例。

因为这两种方法都是单元测试中非常重要的方法,所以我们将同时对这两种方法进行回顾。

方法 1:模拟实例的方法

mock 库有一个特殊的方法装饰器,可以模拟对象实例的方法和属性,即 @mock.patch.object decorator 装饰器:

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from mymodule import RemovalService, UploadService

import mock
import unittest

class RemovalServiceTestCase(unittest.TestCase):

    @mock.patch('mymodule.os.path')
    @mock.patch('mymodule.os')
    def test_rm(self, mock_os, mock_path):
        # instantiate our service
        reference = RemovalService()

        # set up the mock
        mock_path.isfile.return_value = False

        reference.rm("any path")

        # test that the remove call was NOT called.
        self.assertFalse(mock_os.remove.called, "Failed to not remove the file if not present.")

        # make the file 'exist'
        mock_path.isfile.return_value = True

        reference.rm("any path")

        mock_os.remove.assert_called_with("any path")


class UploadServiceTestCase(unittest.TestCase):

    @mock.patch.object(RemovalService, 'rm')
    def test_upload_complete(self, mock_rm):
        # build our dependencies
        removal_service = RemovalService()
        reference = UploadService(removal_service)

        # call upload_complete, which should, in turn, call `rm`:
        reference.upload_complete("my uploaded file")

        # check that it called the rm method of any RemovalService
        mock_rm.assert_called_with("my uploaded file")

        # check that it called the rm method of _our_ removal_service
        removal_service.rm.assert_called_with("my uploaded file")

非常棒!我们验证了 UploadService 成功调用了我们实例的 rm 方法。你是否注意到一些有趣的地方?这种修补机制(patching mechanism)实际上替换了我们测试用例中的所有 RemovalService 实例的 rm 方法。这意味着我们可以检查实例本身。如果你想要了解更多,可以试着在你模拟的代码下断点,以对这种修补机制的原理获得更好的认识。

陷阱:装饰顺序

当我们在测试方法中使用多个装饰器,其顺序是很重要的,并且很容易混乱。基本上,当装饰器被映射到方法参数时,装饰器的工作顺序是反向的。思考这个例子:

    @mock.patch('mymodule.sys')
    @mock.patch('mymodule.os')
    @mock.patch('mymodule.os.path')
    def test_something(self, mock_os_path, mock_os, mock_sys):
        pass

注意到我们的参数和装饰器的顺序是反向匹配了吗?这部分是由 Python 的工作方式所导致的。这里是使用多个装饰器的情况下它们执行顺序的伪代码:

patch_sys(patch_os(patch_os_path(test_something)))

因为 sys 补丁位于最外层,所以它最晚执行,使得它成为实际测试方法参数的最后一个参数。请特别注意这一点,并且在运行你的测试用例时,使用调试器来保证正确的参数以正确的顺序注入。

方法 2:创建 Mock 实例

我们可以使用构造函数为 UploadService 提供一个 Mock 实例,而不是模拟特定的实例方法。我更推荐方法 1,因为它更加精确,但在多数情况,方法 2 或许更加有效和必要。让我们再次重构测试用例:

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from mymodule import RemovalService, UploadService

import mock
import unittest

class RemovalServiceTestCase(unittest.TestCase):

    @mock.patch('mymodule.os.path')
    @mock.patch('mymodule.os')
    def test_rm(self, mock_os, mock_path):
        # instantiate our service
        reference = RemovalService()

        # set up the mock
        mock_path.isfile.return_value = False

        reference.rm("any path")

        # test that the remove call was NOT called.
        self.assertFalse(mock_os.remove.called, "Failed to not remove the file if not present.")

        # make the file 'exist'
        mock_path.isfile.return_value = True

        reference.rm("any path")

        mock_os.remove.assert_called_with("any path")


class UploadServiceTestCase(unittest.TestCase):

    def test_upload_complete(self, mock_rm):
        # build our dependencies
        mock_removal_service = mock.create_autospec(RemovalService)
        reference = UploadService(mock_removal_service)

        # call upload_complete, which should, in turn, call `rm`:
        reference.upload_complete("my uploaded file")

        # test that it called the rm method
        mock_removal_service.rm.assert_called_with("my uploaded file")

在这个例子中,我们甚至不需要修补任何功能,只需为 RemovalService 类创建一个 auto-spec,然后将实例注入到我们的 UploadService 以验证功能。

mock.create_autospec 方法为类提供了一个同等功能实例。实际上来说,这意味着在使用返回的实例进行交互的时候,如果使用了非法的方式将会引发异常。更具体地说,如果一个方法被调用时的参数数目不正确,将引发一个异常。这对于重构来说是非常重要。当一个库发生变化的时候,中断测试正是所期望的。如果不使用 auto-spec,尽管底层的实现已经被破坏,我们的测试仍然会通过。

陷阱:mock.Mock 和 mock.MagicMock 类

mock 库包含了两个重要的类 mock.Mockmock.MagicMock,大多数内部函数都是建立在这两个类之上的。当在选择使用 mock.Mock 实例、mock.MagicMock 实例还是 auto-spec 的时候,通常倾向于选择使用 auto-spec,因为对于未来的变化,它更能保持测试的健全。这是因为 mock.Mockmock.MagicMock 会无视底层的 API,接受所有的方法调用和属性赋值。比如下面这个用例:

class Target(object):
    def apply(value):
        return value

def method(target, value):
    return target.apply(value)

我们可以像下面这样使用 mock.Mock 实例进行测试:

class MethodTestCase(unittest.TestCase):

    def test_method(self):
        target = mock.Mock()

        method(target, "value")

        target.apply.assert_called_with("value")

这个逻辑看似合理,但如果我们修改 Target.apply 方法接受更多参数:

class Target(object):
    def apply(value, are_you_sure):
        if are_you_sure:
            return value
        else:
            return None

重新运行你的测试,你会发现它仍能通过。这是因为它不是针对你的 API 创建的。这就是为什么你总是应该使用 create_autospec 方法,并且在使用 @patch@patch.object 装饰方法时使用 autospec 参数。

现实例子:模拟 Facebook API 调用

作为这篇文章的结束,我们写一个更加适用的现实例子,一个在介绍中提及的功能:发布消息到 Facebook。我将写一个不错的包装类及其对应的测试用例。

import facebook

class SimpleFacebook(object):

    def __init__(self, oauth_token):
        self.graph = facebook.GraphAPI(oauth_token)

    def post_message(self, message):
        """Posts a message to the Facebook wall."""
        self.graph.put_object("me", "feed", message=message)

这是我们的测试用例,它可以检查我们发布的消息,而不是真正地发布消息:

import facebook
import simple_facebook
import mock
import unittest

class SimpleFacebookTestCase(unittest.TestCase):

    @mock.patch.object(facebook.GraphAPI, 'put_object', autospec=True)
    def test_post_message(self, mock_put_object):
        sf = simple_facebook.SimpleFacebook("fake oauth token")
        sf.post_message("Hello World!")

        # verify
        mock_put_object.assert_called_with(message="Hello World!")

正如我们所看到的,在 Python 中,通过 mock,我们可以非常容易地动手写一个更加智能的测试用例。

Python Mock 总结

即使对它的使用还有点不太熟悉,对单元测试来说,Python 的 mock 库可以说是一个规则改变者。我们已经演示了常见的用例来了解了 mock 在单元测试中的使用,希望这篇文章能够帮助 Python 开发者克服初期的障碍,写出优秀、经受过考验的代码。


via: https://www.toptal.com/python/an-introduction-to-mocking-in-python

作者:NAFTULI TZVI KAY 译者:cposture 校对:wxy

本文由 LCTT 原创翻译,Linux中国 荣誉推出