基于 ZeroMQ 优化处理云计算中的并发问题

多核时代的挑战

传统并发策略的弊端

由于在提高 CPU 频率上遇到的瓶颈，当前多核架构已经成为提高 CPU 计算能力的主要方式，并且在可以预见的时间内，一台计算机尤其是大型机可以拥有的核数乃至 CPU 数仍将不断提升。

然而传统的软件技术在处理并发问题能力上的滞后性却使得多核硬件技术无法充分发挥其优势。传统的软件开发语言通常使用线程来处理并发问题，这种做法存在一些固有的弊端，比如：

• 同步缺失：线程之间共享内存数据，当多个线程访问同一数据而其中一个线程存在修改数据的行为时，就会出现 race-condition 而导致程序出错。开发人员必须对各线程中访问同一数据的代码段使用各种同步机制进行同步，比如设立 critical section，使用 Semaphore 控制访问。
• 同步粒度设置不当：为了避免同步缺失问题，必须使得不同线程中的某些代码段顺序执行而非并发执行，这样的区域成为关键区域。显而易见，关键区域越大，整个程序的并发性就越差；但若将关键区域设置的太小，又容易导致同步缺失的问题。
• 可分解的读写操作：即使是最简单的 32 位或是 64 位的数据类型读写操作，也未必是原子操作，这需要程序员对系统的底层实现有深入了解。而一旦它们不是原子操作，就需要进行同步处理。
• 潜在的代码优化：当前的 CPU 和代码编译器通常具有足够的智能来对接收的代码进行优化，这通常会导致代码的实际执行顺序与开发人员所给出的顺序在局部有所差别。这增加了处理同步缺失问题的难度。
• 死锁：一旦我们开始使用锁来处理同步问题，就引入了潜在的死锁危险。当多个线程已经持有的资源和它们试图访问的资源形成一个环路，就会导致所有的线程都因为无法继续执行而处于停滞的状态，进而导致程序失去响应。
• 循环让步：为了避免死锁的状况，通常会采取让无法继续执行的线程让步给其他线程的情况，但若第二个线程也无法继续执行又让步给前一个线程，那么这两个线程就会不断的在挂起和唤醒两个状态之间进行切换。而线程切换是一种耗费 CPU 的工作。
• 优先级倒置：通过在运行过程中动态调整线程优先级来让低优先级的线程获得执行机会的策略，可能使得原本高优先级的任务被原本低优先级的任务所阻塞。

即便我们的开发人员有足够的能力来处理好上述问题，我们也要为此付出额外的代价：

• 编写和维护多线程的应用是非常耗费时间和精力的工作。
• 因为处理多线程问题的难度，多线程应用的线程数往往有限。当计算机的 CPU 核数达到 16 个以上时 CPU 核数和应用线程数的差异往往导致 CPU 无法被充分利用。
• 线程同步机制存在的必要性使得在多线程应用中某些代码实际上仍然是顺序执行，这 CPU 的多核设计背道而驰。

本文介绍了如何运用开源的 ZeroMQ 来解决上述问题，并且详述了 ZeroMQ 在主流云技术 Openstack 中的运用，给云计算开发人员解决此类问题提供思路和参考。

从 Erlang 到 ZeroMQ

理想化的并发应用必须能够适应任何数量的 CPU 核数，杜绝锁的使用，在多数情况下，其编程实践应该与单线程开发一致而无需额外的繁琐同步处理。Erlang 语言为我们提供一个范例，它具备如下优点：

（注：这里的进程与操作系统进程并无直接对应关系）

• 快速的创建和销毁进程
• 能从容应对 10000 个以上的并发进程。
• 进程间基于消息传递进行快速的异步通信。
• 基于消息传递，进程间实现数据零共享。
• 对进程状态进行实时监控。
• 选择性的消息接收。

这是一种与多线程机制截然不同的并发策略，其关键在于通过消息传递 (Messaging) 来实现进程间的通信而非共享内存数据。进程首先将消息发送给消息队列，后者再将消息传递给目标进程，这一过程无需使用锁来进行同步。

然而 Erlang 的受众仍十分有限，需要有一种受众更广，更易于使用的技术，来最大限度的发挥多核硬件的能力。这正是创立 ZeroMQ 的初衷，它在继承了 Erlang 优秀的并发特性的基础上，更具备如下优势：

• 开放源码，拥有一个活跃的开源社区。
• 基于广为人知的 BSD socket 接口，易于使用。
• 实现了多种常见的消息通信模型，易于和实际问题相结合。
• 兼容多数编程语言、操作系统和硬件，具有良好的可移植性。
• 没有消息中间件，这避免了 Single Point of Failure，减少了维护成本。

ZeroMQ 针对不同应用场合的消息传递模型

ZeroMQ 简介

ZeroMQ（也拼写作ØMQ，0MQ 或 ZMQ) 是一个为可伸缩的 分布式 或并发应用程序设计的高性能异步消息库。它提供一个 消息队列 , 但是与 面向消息的中间件 不同，ZeroMQ 的运行不需要专门的 消息代理 （ message broker ）。该库设计成常见的 套接字 风格的 API 。ZeroMQ 是由 iMatix 公司和大量贡献者组成的社群共同开发的。ZeroQ 通过许多第三方软件支持大部分流行的编程语言，从 Java 和 Python 到 Erlang 和 Haskell 。

上一节中我们提到，走出传统并发策略困境的关键在于，以消息传递取代内存数据共享来进行进程（线程）间的通信。而 ZeroMQ 正是通过为传统的 Socket 接口赋予消息队列的能力来实现这一目标。

Request-reply 模式

Request-reply 是 ZeroMQ 提供的最常用的消息传递模式之一。在这种模式下，客户端进程发起请求，服务器端进程接受请求并返回响应给客户端。客户端和服务器端进程都可以有多个。

清单 1 和清单 2 实现了一个简单的“请求-应答”应用的服务器端和客户端。在 HelloWorldServer.py 中，我们首先创建了一个 socket 对象，将它绑定到一个特定的地址。一旦接受到客户端的请求，就发送内容为”World”的回复。

清单 1. HelloＷorldServer.py

import zmq import time  context = zmq.Context() socket = context.socket(zmq.REP) socket.bind("tcp://*:5555")  while True:  # 等待下一个来自客户端的请求  message = socket.recv()  print "Received request: ", message   # 休眠时间  time.sleep (1) # Do some 'work'   # 发送回复给客户端  socket.send("World")

清单 2. HelloWorldClient.py

import zmq  context = zmq.Context()  # 发送给服务器端的 Socket print "Connecting to hello world server..." socket = context.socket(zmq.REQ) socket.connect ("tcp://localhost:5555")  # 发送 10 个请求，每次都等待相应 for request in range (10):  print "Sending request ", request,"..."  socket.send ("Hello")    # 收到回复  message = socket.recv()  print "Received reply ", request, "[", message, "]"

从表面上看这种风格与传统的 socket 十分相似，但实际上它们有重大的差别。首先，ZeroMQ 的 socket 是面向消息的，我们从 socket 里直接获得消息字符串，而非字节流，发送亦然。其次，开发者无需关心负责底层通讯的连接的管理，这种连接可能是传统的 socket 连接，也可能基于其他协议。这些底层连接的创建，销毁，重连以及它如何确保消息被有效的发送，都由 ZeroMQ 负责管理。最后，ZeroMQ 的 socket 之间的连接不受任何限制，而传统的 socket 之间往往无法建立多对多的连接。因此，ZeroMQ 的 socket 可以被看作一个功能完善的消息队列。

REQ 类型的 socket 通常被用来发送请求，并且只有在收到第一个请求的回复之后，才能发送第二个请求。在 HelloWorldClient.py 中该 REQ 类型的 socket 只连接到了一个地址，但它也可以连接多个地址。在这种情况下，ZeroMQ 将确保消息被均匀的发送给每个地址，但每次只有一个地址会受到请求。REP 类型的 socket 用于接受请求。它必须在发送第一个请求的回复之后才能接受第二个请求。尚未来得及处理的请求按顺序被置于队列中。

REQ 和 REP 类型的 socket 在消息发送和接受的操作序列上存在严格限制，为了应对更复杂的情况，ZeroMQ 也提供了更为灵活的 socket 类型，这就是 DEALER 和 ROUTER。

DEALER 和 REQ 的区别在于，它可以按照任意的次序执行发送消息和接受消息的操作，而不必等待上一个请求的回复。同样，ROUTER 也不必等待发送上一次请求的响应完成就能接受第二个请求。此外，ROUTER 会为请求加上标识以记录最初请求者的身份。这样一来它可以将该请求发送给其他进程处理，得到返回结果后，仍可以根据消息中的身份标识将该请求准确的返回给最初请求者。因此 ROUTER 和 DEALER 可以被用来实现类似于传统消息队列架构中的消息服务器的进程。

Publish-subscribe 模式

Publish-subscribe 是用于广播消息的模式，在这种模式下发布的消息将同时发送给多个节点。它包含 PUB 和 SUB 两种 socket 类型。与 Request-reply 不同，PUB 和 SUB 都只能进行单向的消息传递。PUB 只能发送消息，而 SUB 只能接受消息。

清单 3，清单 4 是一个简单的 Publish-subscribe 模式的实现。从中我们可以看到，作为消息订阅者的 syncsub.py，将一个 SUB 类型 socket 绑定到‘tcp://localhost:5561’，这代表了一个单一的地址。而作为消息发布者的 syncpub.py，将一个 PUB 类型的 socket 绑定到‘tcp://*:5561’，这实际上匹配了多个地址。也就是说，凡是绑定到符合格式‘tcp://*:5561’地址的任何 SUB 类型的 socket 都可以接收到该 PUB 进程发布的消息。

清单 3. syncsub.py

def main():  context = zmq.Context()    # 首先，连接我们的订阅 socket  subscriber = context.socket(zmq.SUB)  subscriber.connect('tcp://localhost:5561')  subscriber.setsockopt(zmq.SUBSCRIBE, "")   # 其次，跟 publisher 同步  syncclient = context.socket(zmq.REQ)  syncclient.connect('tcp://localhost:5562')    # 发送一个同步请求  syncclient.send('')    # 等待同步的回复  syncclient.recv()   # 第三，收到更新并且报告我们收到的数目  nbr = 0  while True:  msg = subscriber.recv()  if msg == 'END':  break  nbr += 1    print 'Received %d updates' % nbr  if __name__ == '__main__':  main()

清单 4. syncpub.py

import zmq  # 我们等待 10 个 subscribers SUBSCRIBERS_EXPECTED = 10  def main():  context = zmq.Context()    # 发送给客户端的 socket  publisher = context.socket(zmq.PUB)  publisher.bind('tcp://*:5561')   # 收到信号的 socket  syncservice = context.socket(zmq.REP)  syncservice.bind('tcp://*:5562')   # 从订阅者收到同步  subscribers = 0  while subscribers < SUBSCRIBERS_EXPECTED:  # 等待同步请求  msg = syncservice.recv()  # send synchronization reply  syncservice.send('')  subscribers += 1  print "+1 subscriber"    # 广播 1M 的更新后结束  for i in range(1000000):  publisher.send('Rhubarb');   publisher.send('END')  if __name__ == '__main__':  main()

在这一实现中，我们还使用 REQ 和 REP 类型的 socket 对 SUB 和 PUB 进程进行了同步，仅当出现 10 个 SUB 进程时，PUB 进程才会开始发送消息。

Pipeline 模式

Pipeline 模式通常用于实现工作流的概念，每个进程负责整个处理流程中的一个步骤。每个步骤接受上一步的处理结果，并将自己的处理结果传递给下一步。每一步可以有多个备选进程。它包含 PUSH 和 PULL 两种 socket 类型。与 PUB 和 SUB 类型的 socket 类似，这两种 socket 都只能做单向消息传递，PUSH 只能发送消息，PULL 只能接受消息。因此通常一个进程需要同时包含这两种类型的 socket。此外，与 PUB 不同的是，PUSH 只会将消息发送给单个 PULL 节点。

Exclusive pair 模式

这种模式仅适用于两个特定节点之间互相传递消息的场合。

ZeroMQ 核心机制分析

在上一节中我们简要介绍了 ZeroMQ 针对各种不同应用场合提供的各种消息传递模型。然而消息队列技术发展至今，有许多成功的产品，它们同样提供了这些模型。那么 ZeroMQ 与这些传统的消息队列技术相比有什么独特的优势呢？可以说，其中最大的区别就是 ZeroMQ 弱化了消息中间件的概念。

在传统的消息队列系统中，消息中间件扮演着核心的角色。进程间并不是直接交互，而是连接到消息中间件，由消息中间件确保消息的有效传递。这种模式的优点是：

• 进程本身不需要维护其他进程的地址信息，而只需知道消息中间件的地址。无论是增加、删除或者修改消息接收进程的地址都不会对消息发送进程产生影响，反之亦然。
• 进程的生命周期不会存在依赖。消息发送进程在发送消息时不必担心消息接收进程尚未启动而导致消息丢失，因为消息中间件会负责保存消息并在消息接收进程启动时将消息传递给它。
• 即使消息接收进程和消息发送进程都产生故障，消息仍然可以保留在消息中间件上。

尽管如此，消息中间件的缺点也同样明显：

• 极大的增加了网络开销。本可以由进程 A 发送给进程 B 的消息，现在必须先由进程 A 发送至消息中间件，再由消息中间件发送给进程 B。如图 1，图 2 所示。
• 消息中间件可能成为整个系统的瓶颈。由于所有的消息都必须经由消息中间件，容易导致消息中间件负载过高，而此时其他进程却因为无法接受到消息而处于空闲的状态。系统的整体效率因此而大幅降低。

图 1. 存在消息中间件时的消息传递

图 2. 没有消息中间件时的消息传递

不难看出，不管是使用单一的消息中间件或者是完全不使用消息中间件，都不是完美的解决方案。在实际应用，往往需要采用一些折中的解决方式，比如：

• 用目录服务取代消息中间件。消息中间件的主要作用之一体现在它管理着所有参与消息传递的进程地址。当整个系统中存在上百个乃至更多这样的进程时，这种统一管理的服务是不可或缺的。但是在这种庞大的系统中，如果同时由消息中间件负责消息传递，容易导致其成为系统的瓶颈。因此将目录服务和消息传递服务进行分离，消息中间件仅提供目录服务而将消息传递功能交由各个进程自身去实现，可以在保证进程有效管理的同时，有效分散系统的负载到各个进程上，提高系统的整体效率。
• 当系统的总进程数较多时，可以建立多个消息中间件，每个中间件管理部分进程。
• 同理，可以建立多个目录服务来避免 SPOF(Single Point of Failure)，提高系统的稳定性。

由于 ZeroMQ 本身并不依赖于消息中间件，因此开发者可以根据实际情况来选择合适的消息传递模型。而传统的消息队列技术因为过度依赖现有的消息中间件产品，难以提供这种灵活性。

ZeroMQ 在 Openstack 中的应用

Openstack 作为一个开源的 IaaS 平台为人们所熟知，并在近年来伴随着云计算的兴起而成为热点。Nova 是 Openstack 中最为核心的组件，负责完成与虚机相关的各种操作。Nova 采用了多进程的架构，通过消息传递来完成各进程间的相互协作，并且提供了一个基于 ZeroMQ 的实现。

Nova 中的消息传递

在 Nova 的诸多进程中，nova-api 负责提供统一的对外接口，nova-scheduler 负责为虚机选择合适的物理机宿主，nova-compute 负责虚机的创建和启停等操作，nova-network 和 nova-volume 分别负责虚机的网卡和存储的相关操作。这些进程之间完全依赖消息传递来进行通信。

图 3. 创建新虚机操作的消息传递流程

以图 3 中的创建新虚机操作为例，首先用户将新虚机的规格参数提供给 nova-api。nova-api 将用户提供的参数组装成一个创建虚机的请求，以消息的形式发送给 nova-scheduler。nova-scheduler 根据相关策略为新的虚机选择一个合适的物理机宿主，然后发送消息给 nova-compute，要求在该物理机上按要求创建一个新的虚机。当所有操作都完成后，nova-scheduler 再将所创建虚机的信息封装成消息发送给 nova-api，最终返回给用户。

ZeroMQ 在 Nova 中的应用

为了实现上一节中所描述的架构，Nova 利用 ZeroMQ 建立了一套消息传递机制，提供以下操作：

• CALL 和 MULTICALL。这类操作发送请求给其他进程，然后等待回应。尽管看起来 REQ 和 REP 类型的 socket 非常适合这种场景，但是它们对消息传递操作的顺序有严格限制，无法并发处理请求，这显然会降低系统的工作效率。Nova 的开发者结合 PUB，SUB，PUSH 和 PULL 四类型的 socket 来实现这类操作。完成一个 call 操作，通常涉及三个进程，如图 4 所示。与普通的 Request-reply 模式相比，这里引入了响应中转进程来转发响应给请求发起进程。这样做的好处是，请求响应进程无需知晓请求发起进程的存在，减轻了进程间的依赖程度。MUTICALL 是可以同时发送请求给多个响应进程的操作。

图 4. call 操作的实现流程

• CAST 和 NOTIFY。这类操作发送指令给其他进程，但并不需要获得回应，可以看做是 CALL 的简化版。在 Nova 里，这类操作也是由 PULL 和 PUSH 类型的 socket 实现的。
• FANOUT-CAST。这类操作用于发送广播指令，由 PUB 和 SUB 类型的 socket 实现。

图 5. 创建新虚机操作的实现

结合这些操作，图 3 可以重新描述如下，如图 5 所示。值得一提的是，nova-scheduler 和 nova-compute 之间的 CAST 和 NOTIFY 是两个异步的操作，即 nova-scheduler 通过 cast 操作发出部署请求后，就已经发送响应给 nova-api，此时虚机处于部署中的状态。当部署完成时，nova-compute 另行通过 NOTIFY 发送通知给 nova-scheduler，将虚机状态设置为部署完毕。此外，在创建虚机的过程中，nova-scheduler 也会发送相应的事件给其他进程，以便进行协调。

结束语

通过以上各章节，我们了解了 ZeroMQ 的来历、基本功能以及在开源 IaaS 框架 Openstack 中的应用。总而言之，ZeroMQ 为开发者提供了简单易用、功能完善的消息通讯机制，能够很好的解决多进程并发应用中各进程间通信协作的问题。同时，ZeroMQ 不依赖于消息中间件，这为开发者提供了足够的灵活性，根据实际问题的需要来构建适应性更强的应用。