使用 Python 和 ZeroMQ 构建代码插桩库

本文作者为 Rob Martin，最初发表于 2013 年 8 月的《软件开发者杂志》（August 2013 issue）。你可以在 SDJ 网站找到更多文章。 

问题所在  

和很多人一样，我曾将海森堡不确定性原理与观察者效应混淆。海森堡不确定性原理断言，我们无法精确测量粒子的成对物理性质。也就是说，如果我们知道一个值，另一个值就无法得知。最能说明这一点的是海森堡被警察拦下的故事。警察问海森堡是否知道他开多快。海森堡说：“不知道，但我知道我在哪里。” 警察说：“先生，你超速了，时速 76 英里。” 海森堡回答：“太好了。现在我迷路了。”

另一方面，观察者效应描述了测量系统中事物对系统产生的影响。一个常见的例子是测量轮胎压力。如果不漏掉一点空气就很难做到，从而影响轮胎的压力。我们测量的越多，轮胎就越瘪。 

测量代码性能的仪器也受观察者效应的影响。插入代码（或反思现有代码）以测量应用程序性能的过程本身也会影响应用程序的性能。如果我们天真地针对 Librato 或 New Relic 等云端指标工具编写代码，我们可能会因为等待阻塞 I/O 来将测量结果记录到远程位置而严重影响性能。

我需要一个测量代码性能的库，并且希望将观察者效应的影响降到最低。这个问题暗示了一种异步编程模式，该模式可以实时记录，但将结果存储在另一个线程或进程中。

ZeroMQ

介绍 ZeroMQ。

表面上看，ZeroMQ 似乎是一个非常快速（“零”时间）的消息队列，但我认为这个说法有点误导。它不像 RabbitMQ 那样是消息代理。它不支持高级消息队列协议 (AMPQ)。它没有管理界面。它不会将消息持久化到磁盘，如果没有订阅者，默认情况下发布者的所有消息都会被丢弃。你无法检查消息或获取队列的统计信息——至少在不编写自己的管理层的情况下不行。

ZeroMQ 更像是一个出色且快速的套接字库，内置支持多种异步模式。当你不需要复杂的代理功能时，这使得 ZeroMQ 成为理想的消息调度程序。对于我的指标库，我不需要这些功能，但我需要速度。

构建代码插桩库

我想要的功能很简单

• 方便进行计时和计数插桩。

• 在插桩代码中高效运行。

• 能够在一个通用系统中对多个程序、进程和线程进行插桩。这些进程是指标的发布者。

• 支持多个后端系统消费指标。这些进程是指标的订阅者。

由于 ZeroMQ 是我的消息调度程序，而 Librato.com 是我记录和聚合这些指标的主要目标，因此我选择了 Zibrato 这个组合词作为我的项目名称。

Zibrato 代码插桩库

在我们深入了解库的架构之前，先快速看一下 API。Zibrato 库提供了三种方式来插桩你的 Python 代码：

• 计时器：Zibrato 提供了一个装饰器，可以计时任何定义的函数或方法，以及一个与任何代码块配合使用的上下文管理器。

  from zibrato import Zibrato

  z = Zibrato()

  # 装饰过的函数

  @z.time_me(level = 'debug', name = 'myfunct_timer', source = 'myprog')

  def myfunct()

  time_consuming_operations()

  # 上下文管理器

  with z.Time_me(level = 'debug', name = 'timer_name')

      slow_function_to_time() 

• 计数器：Zibrato 以装饰器或上下文管理器的形式提供计数器方法。  

   from zibrato import Zibrato

  z = Zibrato()

  # 装饰过的函数

  @z.count_me(level = 'info', name = 'myfunct_counter', value = 5) # 增加 5

  def myfunctc()

    pass

  # 上下文管理器

  with z.Count_me(level = 'info', name = 'counter_name', source = 'deathstar')

    pass 

• Gauge：最后，Zibrato 可用于在代码中的任何点将任意值插入后端。

  from zibrato import Zibrato

  z = Zibrato()

  # Zibrato gauge

  z.gauge(level = 'crit', name = 'gauge_name', value=123) 

这只是 Zibrato 如何用于代码插桩的快速概览。有关更多信息，请访问 Pypi (https://pypi.python.org/pypi/Zibrato) 上的库或查看我的 Github.com 存储库 (https://github.com/version2beta/zibrato)。

架构

为了实现 API 背后的目标，Zibrato 分为三个部分：

• Zibrato 库，实现了上述 API 并将指标发布到消息队列。作为用户，我可以拥有零个或多个插桩进程，它们都与我的消息队列通信。

• 消息代理，它订阅零个或多个指标发布者，然后又将指标重新发布给零个或多个后端订阅者。

• 零个或多个后端提供程序，它们订阅消息代理以接收指标，然后执行相应的操作。在我的应用程序中，后端提供程序将消息发送到我的 Librato 账户。

这被称为“扩展的发布-订阅模式”。消息代理（有时称为消息总线）是此拓扑的核心。它支持多个发布者，并过滤后端提供程序将接收哪些消息。

为了在服务器上实现 Zibrato，我通常使用 supervisord 来启动代理和我的 Librato 后端。然后，任何插桩代码都可以连接到代理，后端将接收满足其过滤条件的任何消息，并将其转发到 Librato。后端聚合消息并执行工作中阻塞的 I/O 部分，通过网络与 Librato.com 通信，这完全是一个独立于插桩代码的进程。

使用 ZeroMQ

显然，ZeroMQ 是 Zibrato 架构中的“秘密武器”，它完成了从插桩代码到后端提供程序的消息重担。它提供了异步性。

如果你还没有安装 ZeroMQ，可以使用 pip 轻松安装：

pip install --upgrade python-dev pyzmq  

如果你正在运行 Continuum Analytics 的 Anaconda Python，pyzmq 已经安装好了。

创建消息代理

我们的代理订阅发布者，然后发布给订阅者。ZeroMQ 将这种类型的设备称为“转发器”。用 Python 实现这一点非常简单。

• 首先，我们创建一个 ZeroMQ 上下文，它基本上是一个线程安全的套接字容器，允许我们在完成工作后干净地关闭所有内容。

• 然后，我们创建一个 TCP 套接字作为插桩代码的订阅者。这也可以通过 Unix 域套接字（类似于命名管道）或 PGM 组播 IP 套接字来实现，但我希望能够连接到特定 IP 地址和端口上的代理，即使代理运行在与插桩代码不同的服务器上。潜在的陷阱：默认情况下，订阅者会过滤掉所有消息，所以我们必须告诉它接收哪些消息。空字符串表示接收所有消息。

• 接下来，我们创建一个 TCP 套接字作为后端发布者。同样，我做了一个设计决策，允许后端运行在不同的服务器上。

• 最后，我们将订阅者套接字和发布者套接字组合成一个 ZeroMQ 转发器设备。

代码如下。

import zmq 

    # 获取 ZeroMQ Context

    context = zmq.Context()

    # 创建一个订阅者

    subscriber = context.socket(zmq.SUB)

    subscriber.bind('tcp://127.0.0.1:5550')

    # 订阅所有消息

    subscriber.setsockopt(zmq.SUBSCRIBE, '')

    # 创建一个发布者

    publisher = context.socket(zmq.PUB)

    publisher.bind('tcp://127.0.0.1:5551')

    # 将订阅者和发布者组合成一个转发器

    zmq.device(zmq.FORWARDER, subscriber, publisher)   

这段代码让我们入门。现在，任何运行在本地的用任何编程语言编写的 ZeroMQ 发布者都可以向我们的代理发送消息。同样，任何本地订阅者都可以接收这些消息。如果我们选择在网络可访问的 IP 地址上创建套接字，而不是 127.0.0.1，那么代理也可以连接到其他机器上的发布者和订阅者。

构建订阅者

没有监听器的代理几乎没有意义。让我们创建一个简单的订阅者，它接收消息并将其打印到标准输出。我们将在一个单独的 Python 脚本中创建它——它与代理分开运行。

创建订阅者的方法如下：

• 首先，我们连接到上面描述的 ZeroMQ 上下文。

• 然后，我们创建监听器的套接字。

• 接下来，我们设置一个过滤器，套接字将订阅该过滤器。我们调用以接收来自消息代理的消息将仅返回以该过滤器开头的值。空字符串订阅所有消息，但默认是不订阅任何消息，所以我们必须将其设置为某个值，即使是空字符串。

• 最后，我们设置一个循环，持续监听，直到有内容进来。

我们的代码可能看起来像这样：

import zmq

    # 获取 ZeroMQ context

    context = zmq.Context()

    # 创建一个套接字

    socket = context.socket(zmq.SUB)

    socket.connect('tcp://127.0.0.1:5551')

    # 订阅所有消息

    socket.setsockopt(zmq.SUBSCRIBE, '')

    # 持续监听

    while True

        print socket.recv()  

这段代码提供了一个监听器，它将接收代理转发的任何内容并将其打印到 STDOUT。现在我们只需要有人向代理发送需要转发的消息。

构建发布者

如果森林里的一棵树倒下了，没有人听到，它会发出声音吗？我不知道这个问题的答案，但我知道一个没有发布者的消息代理的生活很安静。

将我们的代理连接到发布者很容易。在第三个 Python 脚本中，我们将创建一个发布者，它向代理发送消息。如果我们做得对，我们在上一节创建的订阅者将收到这些消息并将其打印到 STDOUT。

构建发布者的方法如下：

• 首先，我们像上面一样连接到 ZeroMQ 上下文。

• 接下来，我们创建发布者套接字。

• 最后，我们从发布者向消息代理发送一条消息。

这是代码

import zmq

    # 获取 ZeroMQ context

    context = zmq.Context()

    # 创建一个套接字

    socket = context.socket(zmq.SUB)

    socket.connect('tcp://127.0.0.1:5551')

    # 订阅所有消息

    socket.setsockopt(zmq.SUBSCRIBE, '')

    # 持续监听

    while True

        print socket.recv() 

有了这三个组件，我们就拥有了一个发送消息的发布者，一个接收消息的订阅者，以及一个将消息从任何连接的发布者转发到任何连接的订阅者的代理。

整合

Zibrato 基于上面描述的扩展 PubSub 模式，并使用相同的三个基本组件。

前端是 Zibrato 类，它提供了插桩方法。这是我们的 ZeroMQ 发布者。代码本身非常轻量，它只需要将消息发送到我们的代理，因此对性能的影响非常小。

中间是一个与上面示例非常相似的代理。它是一个简单的转发器，可以接受来自多个发布者的连接，并将消息转发给多个订阅者。

后端是一个使用 Python Requests 库实现 Librato HTTP API 的库，用于存储我们记录的指标。它按设定的时间表将数据刷新到 Librato，包括汇总计数器。Librato 后端继承自标准的后端类，因此很容易实现其他类型的后端——例如简单地输出到日志文件或发送到中央 Statsd 服务器。

鸣谢

艾萨克·牛顿曾说过：“如果我看得更远，那是因为我站在巨人的肩膀上。” 充其量我只是蹲着，并且有掉下巨人肩膀的风险，所以我更喜欢以赛亚·特兰尼（Isaiah di Trani）的早期引言：“谁看得更远，是矮子还是巨人？当然是巨人，因为他的眼睛比矮子高。但是，如果矮子被放在巨人的肩膀上，谁看得更远？…我们也像矮子一样站在巨人的肩膀上。我们掌握他们的智慧，并超越他们。”

ZeroMQ 是 Pieter Hintjens 和 iMatix Corporation 的杰出作品。它是一个强大而灵活的消息平台，我强烈推荐用于异步应用程序。我还建议阅读 Pieter 的 ZeroMQ 指南。它冗长且全面，但相当易懂，甚至读起来也很有趣。

• https://zeromq.cn/

• https://zguide.zeromq.cn/

我最初是通过 Ruby Rogues 播客 #62（由 Librato 的 CTO 和联合创始人 Joseph Ruscio 主持）了解到 Librato 的。他们出色地简化了指标工作。Librato 提供免费的开发账户和第一个月的免费生产使用，之后定价也非常合理。

• https://metrics.librato.com/

• http://rubyrogues.com/062-rr-monitoring-with-joseph-ruscio/

Zibrato 最初受到 Etsy 的 Statsd 包的启发，这是一个 Node.js 服务，与 Graphite（用 Python 和 Django 编写）结合使用，提供了一个完整的异步插桩堆栈。顺便说一句，看看 Steve Ivy。他不仅为与 Statsd 交互编写了一个 Python 库，还用 Python 重写了它。

• https://github.com/etsy/statsd/

• http://graphite.wikidot.com/

• https://github.com/sivy

我使用了 Kenneth Reitz 的 Request 库：HTTP for Humans。这个库使 Web 交互变得轻而易举。

• http://docs.python-requests.org/

我的测试设置在 Gary Bernhardt 的 Expecter 包的帮助下得到了极大的改进，未来我可能会重构我的测试以也使用他的 Dingus 库。自从我在 Ruby 中学习 BDD 以来，Gary 在弥合我从 Ruby 到 Python 的知识鸿沟方面非常有帮助。

• https://github.com/garybernhardt/expecter

• https://github.com/garybernhardt/dingus

我越来越欣赏 Continuum Analytics 的 Anaconda Python 的基础版本。它现在是我所有开发机器和虚拟开发环境都安装的软件。

• https://store.continuum.io/cshop/anaconda/

特别感谢 Tracy Harms (https://twitter.com/kaleidic)，他花了几天时间与我一起进行 Zibrato 项目的结对编程。他的反馈和见解非常宝贵。

~~~

简介：Rob Martin 是位于美国犹他州普罗沃的 i.TV 的一名开发人员。他工作中最酷的部分之一是他需要学习他们技术栈中使用的所有语言。在加入 i.TV 之前，他曾在一家 Python 公司从事 Ruby 开发，在一个 PHP 公司从事 Python 开发，并在工厂车间从事 Perl 开发。

Rob Martin 在网上大部分地方的用户名是 version2beta。在 Twitter (@version2beta)、Github.com (https://github.com/Version2beta) 上关注他，或访问他的博客 (http://version2beta.com/)。  i.TV 正在招聘。发送电子邮件至 rob@version2beta.com 获取更多信息。 

即将发布的内容 

如果您对即将发布的内容感兴趣，请查看我们的网站。例如，您可以查看我们两合一的全新 Python 包的目录。《Python 几行代码》和《Python 入门套件》。 

历史

在此处保持您所做的任何更改或改进的实时更新。