面试题：Python多线程网络编程的复杂场景与设计模式

设计思路

职责分离：
- 数据包捕获：创建一个PacketCapture类，负责捕获TCP和UDP数据包。使用pcap库（如pyshark）来实现。
- 数据包分析：设计PacketAnalyzer类，用于分析捕获到的数据包，提取关键信息。
- 数据交互：构建DataInteraction类，处理与远程服务器的数据交互，包括发送监控数据和接收指令。
线程管理：
- 使用ThreadPoolExecutor来管理线程，提高线程的复用性和管理效率。这样可以避免频繁创建和销毁线程带来的开销。
- 为不同的任务（数据包捕获、分析、数据交互）分配不同的线程池，以实现任务隔离，避免相互干扰。
设计模式运用：
- 单例模式：对于一些全局共享的资源，如配置文件读取类、日志记录类等，采用单例模式，确保整个应用程序中只有一个实例，避免资源重复占用和不一致问题。
- 观察者模式：当数据包分析完成后，需要通知数据交互模块发送数据。可以使用观察者模式，让DataInteraction类作为观察者，PacketAnalyzer类作为被观察对象。当分析结果准备好时，通知观察者进行数据发送。

关键代码框架

import concurrent.futures
import pyshark
import requests


class Config:
    _instance = None

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            cls._instance = super().__new__(cls, *args, **kwargs)
        return cls._instance


class Logger:
    _instance = None

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            cls._instance = super().__new__(cls, *args, **kwargs)
        return cls._instance


class PacketCapture:
    def __init__(self):
        pass

    def capture_packets(self):
        capture = pyshark.LiveCapture(interface='eth0')
        for packet in capture.sniff_continuously():
            yield packet


class PacketAnalyzer:
    def __init__(self):
        self.observers = []

    def attach(self, observer):
        self.observers.append(observer)

    def analyze_packet(self, packet):
        # 这里进行数据包分析
        analysis_result = {}
        if 'TCP' in packet:
            analysis_result['protocol'] = 'TCP'
            analysis_result['src_port'] = packet.tcp.srcport
            analysis_result['dst_port'] = packet.tcp.dstport
        elif 'UDP' in packet:
            analysis_result['protocol'] = 'UDP'
            analysis_result['src_port'] = packet.udp.srcport
            analysis_result['dst_port'] = packet.udp.dstport
        for observer in self.observers:
            observer.update(analysis_result)


class DataInteraction:
    def __init__(self, server_url):
        self.server_url = server_url

    def update(self, data):
        try:
            response = requests.post(self.server_url, json=data)
            if response.status_code == 200:
                print("Data sent successfully")
            else:
                print(f"Failed to send data, status code: {response.status_code}")
        except Exception as e:
            print(f"Error sending data: {e}")


def main():
    config = Config()
    logger = Logger()
    packet_capture = PacketCapture()
    packet_analyzer = PacketAnalyzer()
    data_interaction = DataInteraction('http://remote_server/api/monitor')
    packet_analyzer.attach(data_interaction)

    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        future_capture = executor.submit(packet_capture.capture_packets)
        future_analyze = executor.submit(lambda: [packet_analyzer.analyze_packet(packet) for packet in
                                                 future_capture.result()])


if __name__ == "__main__":
    main()

星途面试题库

面试题：Python多线程网络编程的复杂场景与设计模式

知识考点

面试题答案

设计思路

关键代码框架