对于 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

想要远离扭曲之类的东西当然有充分的理由，但我不认为效率是其中之一——我怀疑它们更有可能以正确的方式进行优化。性能是一个棘手的野兽，通常瓶颈并不在您想象的地方，这就是为什么您需要先进行分析才能正确优化。例如，框架可能努力将更多代码推送到 C 扩展中，这肯定会提高性能。如果性能是您的主要动力，那么第三方产品可能是更安全的选择。此外，对于使用其他人针对各种不同用例和环境进行测试和调整的代码也存在很大的争议——如果您最终重新发明了太多的轮子，那么总是存在丢失一些辐条的风险。

但是，您需要做的事情似乎非常简单，因此安装和学习框架以及向代码添加另一个运行时依赖项的开销可能不合理。另外，如果您主要受 IO 限制，那么在处理过程中消耗一些额外的 CPU 也不会产生太大影响。当然，我过去有时会避免使用扭曲之类的东西，只是因为我知道自己编写它会更快（就我的时间而言）并且性能会“足够好”。我一直发现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]