python - 这是 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 的回调系统使调试有点棘手——例如，访问错误消息可能有点令人担忧。这绝不是不可能的，许多人都非常成功地使用它，但我个人觉得它太“笨拙”，不适合简单的任务。

我认为在这种情况下，您将接收和处理拆分到它们自己的进程中的想法可能是错误的经济 - 从套接字接收数据非常快，如果您在纯 Python 中进行大量处理，则可能成为主导性能因素。但是，如果不知道你在做什么处理，我不能肯定地说。如果这将是耗时和/或 CPU 密集型的，并且您可以独立于前几行处理每一行，那么这可能是合理的，但您可能希望将处理工作分配给一整套工作进程。根据您现有的代码，这非常简单 - 只需让主进程成为接收方而不是“从属”进程，并创建一个共享一个 Queue 的工作池。每个 worker 都经历一个循环，选择下一个项目并产生结果。每项花费多长时间并不重要，他们只会在下一项可用时获取下一项(Queue 将为您处理)。

但是，如果您的处理循环也主要受 IO 限制(例如写入文件)，那么您可能会发现单个进程实际上比将所有内容都插入管道的开销要好。这取决于许多因素，包括您的 CPU 架构(某些系统使 CPU 内核之间的传输比其他系统更昂贵)，但最终您不想使用多个进程，除非您非常有信心它会给您带来性能上的胜利。

无论如何，如果循环是 IO 绑定(bind)，您可能会发现具有非阻塞 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]