这是一个高效的Python客户端TCP/IP套接字循环吗？

Question

我正在连接一个服务器，这个服务器会给我发送流式数据，我需要逐行处理这些数据。所以我得把每一行分开，然后处理每一行。以下的代码看起来运行得很好，但我在想，处理这种情况有没有什么标准的设计模式？或者说这样做是不是可以？

使用队列会不会带来什么严重的性能问题？我希望处理速度尽可能快和高效，这也是我不想使用像twisted这样的库的原因。

import socket, multiprocessing

def receive_proc(s, q):
    data = ''
    while True:
        data += s.recv(4096)
        if '\n' in data:
            lines = data.split('\n')[:-1]
            for line in lines:
                if len(line) > 0:
                    q.put(line)
                    data = data.replace(line+'\n', '', 1)

q = multiprocessing.Queue()
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(('127.0.0.1', 1234))

p = multiprocessing.Process(target=receive_proc, args=(s,q))
p.start()

while True:
    line = q.get()

    # do your processing here

Answer 1

确实有一些合理的理由让人想要远离像 twisted 这样的东西，但我觉得效率并不是其中之一——我怀疑它们更可能在正确的方向上进行了优化。性能是个复杂的问题，很多时候瓶颈并不在你想的地方，这就是为什么在进行真正的优化之前，你需要先进行性能分析。例如，一些框架可能已经努力将更多的代码推到 C 扩展中，这肯定会帮助提升性能。如果性能是你最主要的考虑，使用第三方的东西可能是更安全的选择。此外，使用别人测试过并为各种不同使用场景和环境调整过的代码也是一个很大的优势——如果你重新发明了太多轮子，总是有可能会漏掉一些重要的部分。

不过，你需要做的事情看起来相对简单，所以安装和学习一个框架的开销，以及给你的代码增加另一个运行时依赖，可能就不太值得了。而且，如果你的程序主要是输入输出（IO）密集型，那么多花一点 CPU 来处理其实也没什么大不了的。我过去确实有时候避免使用像 twisted 这样的东西，因为我知道自己写会更快（在时间上），而且性能也“足够好”。我总觉得 twisted 的回调系统让调试变得有点棘手——比如获取错误信息就有点麻烦。当然，这并不是说不可能，很多人都用得很好，但就我个人而言，我觉得它对于简单任务来说太“麻烦”了。

我觉得把接收和处理分开到不同进程的想法在这种情况下可能是个错误的节省——从套接字接收数据是非常快的，如果你在纯 Python 中进行大量处理，这可能是主要的性能因素。不过，我不能确定，因为我不知道你具体在做什么处理。如果处理过程耗时较长和/或 CPU 密集型，并且你可以独立处理每一行，那么这样做可能是合理的，但你可能想把处理任务分配给一组工作进程。根据你现有的代码，这个过程相对简单——只需让主进程负责接收，而不是“从属”进程，然后创建一个共享 Queue 的工作进程池。每个工作进程循环取下一个任务并生成结果。每个任务耗时多少并不重要，它们只需在任务可用时获取下一个任务（而 Queue 会为你处理这个过程）。

然而，如果你的处理循环也是主要依赖 IO（例如写入文件），那么你可能会发现单个进程实际上比把所有东西通过管道传输的开销要好。这取决于很多因素，包括你的 CPU 架构（某些系统在 CPU 核心之间的传输成本更高），但最终，除非你非常确定使用多个进程会带来性能提升，否则不建议使用多个进程。

无论如何，如果这个循环确实是 IO 密集型的，你可能会发现使用单个非阻塞 IO 的进程是更好的选择。你可以使用 Python 的 select 模块自己实现，或者你可能会觉得使用像 eventlet 或 gevent 这样的库会更简洁。

顺便提一下，你处理缓冲区开头的方式效率不高——你不需要使用 replace()，可以直接用你现有的 split()，像这样：

while True:
    data += s.recv(4096)
    if '\n' in data:
        lines = data.split('\n')
        for line in lines[:-1]:
            if len(line) > 0:
                q.put(line)
        data = lines[-1]