Python Socket因主线程time.sleep接收过期数据的问题解决问询

问题场景 #

我通过Socket接收数据，在while循环中使用socket.recv(size)按特定大小接收数据。高频率接收时一切正常，但在循环中加入time.sleep(sec)后，每次都会接收到过期数据。看起来Socket缓冲区被旧数据填满，例如当主机每秒发送0.002次数据时，我以1秒的频率接收只能获取到过期数据。

相关代码 #

import ctypes
import datetime
import logging
import socket
import time

from app.service.c_structures import RTDStructure
logging.basicConfig(level=logging.DEBUG)


class RTDSerializer:
    def __init__(self, ip: str, port: int = 29000, frequency: float = 0.002):
        self.data: dict = {}
        self.ip = ip
        self.port = port
        self.frequency = frequency
        self.sock = None
        self.struct_size = ctypes.sizeof(RTDStructure)
        self.logger = logging
        print(ctypes.sizeof(RTDStructure))

    def connect(self):
        try:
            self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.sock.connect((self.ip, self.port))
            self.sock.settimeout(None)
            logging.debug(f"Socket connect [{self.ip}:{self.port}] --> Ok")
            while True:
                c_structure = RTDStructure.from_buffer_copy(self.receive_raw_data() * ctypes.sizeof(RTDStructure))
                self.data = self.ctypes_to_dict(c_structure)
                print(datetime.datetime.now(), self.data['move_des_q'])
                time.sleep(self.frequency)
        except Exception as error:
            logging.error(f"Socket connect [{self.ip}:{self.port}] --> False\n{error}")
            return 0


    def receive_raw_data(self) -> bytes or connect:
        raw_data = self.sock.recv(self.struct_size)
        if raw_data == b'':
            logging.error('Connection lost')
            return self.connect()
        return raw_data

    def ctypes_to_dict(self, ctypes_obj) -> dict or list:
        if isinstance(ctypes_obj, ctypes.Structure):
            data_dict = {}
            for field_name, field_type in ctypes_obj.get_fields():
                field_value = getattr(ctypes_obj, field_name)
                if isinstance(field_value, (ctypes.Structure, ctypes.Array)):
                    data_dict[field_name] = self.ctypes_to_dict(field_value)
                else:
                    data_dict[field_name] = field_value
            return data_dict
        elif isinstance(ctypes_obj, ctypes.Array):
            data_list = []
            for element in ctypes_obj:
                if isinstance(element, (ctypes.Structure, ctypes.Array)):
                    data_list.append(self.ctypes_to_dict(element))
                else:
                    data_list.append(element)
            return data_list


if __name__ == '__main__':
    rtd = RTDSerializer(ip='192.168.0.226', port=29000, frequency=0.05)
    rtd.connect()

注：接收的是包含C类型值的字节串，通过ctypes结构序列化后转换为字典，序列化逻辑并非问题重点。此外Socket有时会返回0字节，因此每次接收时需要检查该情况。

尝试过的解决方案 #

曾在while循环中强制重连，一定程度解决了过期数据问题，但高频率接收时会严重影响运行速度：

def connect(self):
        try:
            logging.debug(f"Socket connect [{self.ip}:{self.port}] --> Ok")
            while True:
                self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                self.sock.connect((self.ip, self.port))
                self.sock.settimeout(None)
                c_structure = RTDStructure.from_buffer_copy(self.receive_raw_data() * ctypes.sizeof(RTDStructure))
                self.data = self.ctypes_to_dict(c_structure)
                print(datetime.datetime.now(), self.data['move_des_q'])
                time.sleep(self.frequency)
                self.sock.close()
        except Exception as error:
            logging.error(f"Socket connect [{self.ip}:{self.port}] --> False\n{error}")
            return 0

疑问 #

• 如何解决过期数据问题？
• Socket断开重连能解决问题的根源是什么？
• 为何Socket缓冲区会被填满且不自动丢弃过期数据？

解决方案思考 #

曾考虑编写clear_buffer函数在线程休眠时丢弃所有未使用的数据，但实现起来过于复杂。另外想到双线程方案：

• 一个线程以高频率接收数据并更新data_buffer
• 另一个线程按设定频率唤醒，读取当前data_buffer的值并打印，之后再次休眠

更新思考：是否应该采用以下双线程模式？

• 打印线程：按设定频率唤醒，读取data_buffer的当前值并打印，之后休眠
• Socket线程：以高频率接收数据并覆盖data_buffer的值

简化示例代码 #

import ctypes
import datetime
import logging
import socket
import time

logging.basicConfig(level=logging.DEBUG)


class RTDReceiver:
    def __init__(self, ip: str, port: int, frequency: float = 0.002):
        self.data: dict = {}
        self.ip = ip
        self.port = port
        self.frequency = frequency
        self.sock = None
        self.struct_size = 1064  # 1064 bytes in my case. 

    def connect(self):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.connect((self.ip, self.port))
        self.sock.settimeout(None)

    def receive_raw_data(self) -> bytes:
        raw_data = self.sock.recv(self.struct_size)
        if raw_data == b'':
            logging.error('Connection lost')
            self.connect()
        time.sleep(self.frequency)
        return raw_data


if __name__ == '__main__':
    rec = RTDReceiver(ip='here your ip', port='here is your port',  frequency=0.02)
    rec.connect()
    while True:
        print(rec.receive_raw_data())

内容的提问来源于stack exchange，提问作者Mika