处理以换行符结尾的套接字数据

Question

在我需要 var data 以换行符 \n 结束的情况下，处理套接字连接的最佳方法是什么？ 我正在使用下面的代码，但有时tcp 数据包会分块，并且需要很长时间才能匹配data.endswith("\n")。 我还尝试了其他方法，例如如果最后一行不以\n 结尾，则保存它，并将其附加到下一个循环中的data。但这也不起作用，因为多个数据包被分块并且第一和第二部分不匹配。 我无法控制另一端，它基本上发送多行以\r\n 结尾。

欢迎提出任何建议，因为我对套接字连接了解不多。

def receive_bar_updates(s):
    global all_bars
    data = ''
    buffer_size = 4096
    while True:
        data += s.recv(buffer_size)
        if not data.endswith("\n"):
            continue
        lines = data.split("\n")
        lines = filter(None, lines)
        for line in lines:
            if line.startswith("BH") or line.startswith("BC"):
                symbol = str(line.split(",")[1])
                all_bars[symbol].append(line)
                y = Thread(target=proccess_bars, kwargs={'symbol': symbol})
                y.start()
        data = ""

“正常”示例data:

line1\r\n
line2\r\n
line3\r\n

分块示例data:

line1\r\n
line2\r\n
lin

Answer 1

如果您想将原始输入作为行处理，则 io 模块是您的朋友，因为它会在行中对数据包进行低级组装。

你可以使用：

class SocketIO(io.RawIOBase):
    def __init__(self, sock):
        self.sock = sock
    def read(self, sz=-1):
        if (sz == -1): sz=0x7FFFFFFF
        return self.sock.recv(sz)
    def seekable(self):
        return False

它比endswith('\n') 更健壮，因为如果一个数据包包含嵌入的换行符（'ab\ncd'），io 模块将正确处理它。你的代码可能变成：

def receive_bar_updates(s):
    global all_bars
    data = ''
    buffer_size = 4096
    fd = SocketIO(s)  # fd can be used as an input file object

    for line in fd:
        if should_be_rejected_by_filter(line): continue # do not know what filter does...
        if line.startswith("BH") or line.startswith("BC"):
            symbol = str(line.split(",")[1])
            all_bars[symbol].append(line)
            y = Thread(target=proccess_bars, kwargs={'symbol': symbol})
            y.start()

Answer 2

使用socket.socket.makefile() 将套接字包装在实现Text I/O 的类中。它处理缓冲、字节和字符串之间的转换，并允许您遍历行。记得刷新所有写入。

例子：

#!/usr/bin/env python3
import socket, threading, time


def client(addr):
    with socket.create_connection(addr) as conn:
        conn.sendall(b'aaa')
        time.sleep(1)
        conn.sendall(b'bbb\n')
        time.sleep(1)
        conn.sendall(b'cccddd\n')
        time.sleep(1)
        conn.sendall(b'eeefff')
        time.sleep(1)
        conn.sendall(b'\n')
        conn.shutdown(socket.SHUT_WR)
        response = conn.recv(1024)
        print('client got %r' % (response,))


def main():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) as listen_socket:
        listen_socket.bind(('localhost', 0))
        listen_socket.listen(1)
        addr = listen_socket.getsockname()
        threading.Thread(target=client, args=(addr,)).start()
        conn, _addr = listen_socket.accept()
        conn_file = conn.makefile(mode='rw', encoding='utf-8')
        for request in conn_file:
            print('server got %r' % (request,))
        conn_file.write('response1\n')
        conn_file.flush()


if __name__ == '__main__':
    main()

$ ./example.py
server got 'aaabbb\n'
server got 'cccddd\n'
server got 'eeefff\n'
client got b'response1\n'
$

Answer 3

您是否接受不同的连接？或者它是一个数据流，由\r\n 分割？

当接受多个连接时，您将等待与s.accept() 的连接，然后处理其所有数据。当您拥有所有数据包时，处理其数据并等待下一个连接。 然后你做什么取决于每个数据包的结构。 （例如：https://wiki.python.org/moin/TcpCommunication）

如果您使用的是数据流，您可能应该处理在单独线程中找到的每一“行”，同时继续使用另一个线程。

编辑： 所以，如果我的情况正确；一个连接，数据是由\r\n 分解的字符串，以\n 结尾。然而，数据并不符合您的预期，而是无限循环等待\n。

据我了解，套接字接口以空数据结果结束。所以最后一个缓冲区可能以\n 结束，但随后继续获取None 对象，试图找到另一个\n。

请尝试添加以下内容：

if not data:
    break

完整代码：

def receive_bar_updates(s):
    global all_bars
    data = ''
    buffer_size = 4096
    while True:
        data += s.recv(buffer_size)
        if not data:
            break
        if not data.endswith("\n"):
            continue
        lines = data.split("\n")
        lines = filter(None, lines)
        for line in lines:
            if line.startswith("BH") or line.startswith("BC"):
                symbol = str(line.split(",")[1])
                all_bars[symbol].append(line)
                y = Thread(target=proccess_bars, kwargs={'symbol': symbol})
                y.start()
        data = ""

Edit2：糟糕，代码错误

Answer 4

我没有测试过这段代码，但它应该可以工作：

def receive_bar_updates(s):
    global all_bars
    data = ''
    buf = ''
    buffer_size = 4096
    while True:
        if not "\r\n" in data:  # skip recv if we already have another line buffered.
            data += s.recv(buffer_size)
        if not "\r\n" in data:
            continue
        i = data.rfind("\r\n")
        data, buf = data[:i+2], data[i+2:]
        lines = data.split("\r\n")
        lines = filter(None, lines)
        for line in lines:
            if line.startswith("BH") or line.startswith("BC"):
                symbol = str(line.split(",")[1])
                all_bars[symbol].append(line)
                y = Thread(target=proccess_bars, kwargs={'symbol': symbol})
                y.start()
        data = buf

编辑：忘了说，我只修改了接收数据的代码，我不知道函数的其余部分（以lines = data.split("\n")开头）是做什么的。

编辑 2： 现在使用“\r\n”代替“\n”进行换行。

编辑 3： 修复了一个问题。

Answer 5

您基本上似乎想从套接字读取行。也许您最好不要使用低级别的recv 调用，而只需使用sock.makefile() 并将结果视为常规文件，您可以从中读取行：from line in sfile: ...

这就留下了延迟/块问题。这很可能是由发送方的Nagle's algorithm 引起的。尝试禁用它：

sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)