使用 IPC、Twisted 或 ZeroMQ 的架构方法？答案

【问题标题】：Architecture approach with IPC, Twisted or ZeroMQ?使用 IPC、Twisted 或 ZeroMQ 的架构方法？
【发布时间】：2013-01-23 04:37:42
【问题描述】：

我正在使用twisted从连接互联网的传感器获取消息，以便将其存储到数据库中。
我想在不干扰这些过程的情况下检查这些消息，因为我需要将每条消息与数据库中的一些基本值进行比较，如果有些匹配，我需要为此触发警报，并且这个想法不是阻止任何进程...

我的想法是创建一个新流程来检查和提醒，但我需要在第一个流程存储消息后，它将消息发送到新流程以便在需要时进行检查和提醒。

为此我需要IPC，并且我正在考虑使用ZeroMQ，但也扭曲了使用IPC的方法，我认为如果我使用ZeroMQ，但它可能会弄巧成拙......

您对我的方法有何看法？也许我完全错了？

欢迎任何建议.. 谢谢

PD：此进程将在专用服务器上运行，预期负载为 6000 msg/小时，每个 1Kb

【问题讨论】：

我认为您的问题不适合 StackOverflow。此类决策过于依赖您当前的基础架构、预期负载等，无法以简洁正确的方式回答。
@pcalcao 谢谢，我正在使用架构标签来回答这个问题，并且我添加了一些关于..的有用信息。
你检查火焰兵吗？ pypi.python.org/pypi/Pyro4
您是否考虑过使用或创建一个简单的负载均衡器，将您的传感器消息分发到一组机器进行处理？它可能会延迟对具有消息总线的复杂架构的需求。
您的负载真的只有 6000 msg/小时，即低于 2 msg/秒吗？如果是这样，这是一个非常低的负载，你甚至不需要非阻塞来处理它，更不用说使用复杂的消息总线了！

标签： python architecture ipc twisted zeromq

【解决方案1】：

收到消息后，做两件事：

检查它是否应该触发警报（并在必要时发送警报，大概）
将其插入数据库

您不需要消息队列、多个进程、IPC 或任何这些东西。例如：

def messageReceived(self, message):
    self.checkForAlerts(message).addCallbacks(self.maybeAlert, log.err)
    self.saveMessageToDatabase(message).addErrback(log.err)

【讨论】：

如果他的负载超过了一台机器的处理能力，他需要多个进程。在这种情况下，他需要将接收消息与处理消息分开。
问题中没有任何内容表明这一点。如果有的话，这个问题表明消息速率相当低（每秒少于两条消息）。
你说得对，不知何故我读到 6000msg/sec 而不是每小时。是的，即使使用简单的阻塞代码也可以很容易地处理这种负载。

【解决方案2】：

所有这些方法都是可能的。我只能抽象地说，因为我不知道你的应用程序的精确轮廓。

如果您已经有一个正在运行的应用程序，但它的速度不足以处理您向其发送的消息数量，那么请确定瓶颈。造成卡顿的两个可能原因是数据库访问或警报触发，因为其中任何一个都可能是同步 IO 操作。

您如何处理这取决于您的工作量：

如果您的消息速率很高且恒定，那么您需要确保您的数据库能够处理此速率。如果您的数据库无法处理它，那么再多的非阻塞消息传递将帮助您！按此顺序：
1. 尝试调整您的数据库。
2. 尝试将您的数据库放在更大内存的计算机上。
3. 尝试在多台机器上分片您的数据库以分配工作负载。一旦您知道您的数据库可以处理消息速率，您就可以使用其他形式的并行处理其他瓶颈。
如果您的消息速率是突发的，那么您可以使用队列来处理突发。按此顺序：
1. 在消息处理器集群前面放置一个负载平衡器。这个平衡器应该做的就是将传感器消息重新分发到不同的机器以进行检查和警报处理。这种方法的优点是您可能不需要更改现有应用程序，只需在更多机器上运行它即可。如果您的负载均衡器不需要等待响应，只需转发消息，这种方式效果最佳。
2. 如果您的通信需求更复杂或者是双向的，您可以使用消息总线（例如 ZeroMQ）作为消息处理器、警报发送器和数据库检查器之间的通信层。这个想法是通过总线进行非阻塞通信并让总线上的每个节点只做一件事来增加并行性。然后，您可以根据每个消息处理阶段花费的时间来更改节点类型的比率。（即在整个消息处理过程中使队列深度相等。）

【讨论】：