面试题：Python网络编程中自定义数据传输协议的设计与实现

协议设计整体思路

数据格式：

使用JSON格式来定义结构化数据。JSON易于阅读、编写且在不同语言间兼容性好。例如：

{
    "device_info": {
        "device_id": "123456",
        "device_type": "sensor"
    },
    "timestamp": "2024 - 01 - 01T12:00:00Z",
    "business_data": [1, 2, 3]
}

校验机制：
- 采用CRC（循环冗余校验）来验证数据的完整性。在发送端，对JSON序列化后的数据计算CRC值，并将其附加到数据末尾。在接收端，重新计算接收到数据（除CRC值部分）的CRC，并与接收到的CRC值对比，若一致则数据完整。
连接管理：
- 使用TCP协议，因为它提供可靠的连接，能保证数据按序到达。对于长连接应用场景，可通过心跳机制保持连接活跃，例如每隔一定时间（如10秒）发送一个简单的心跳包（如一个特定的JSON格式心跳消息），若接收方未在规定时间内收到心跳包，则判定连接断开并尝试重新连接。
加密方式：
- 使用AES（高级加密标准）对称加密算法。在连接建立初期，双方通过密钥交换协议（如Diffie - Hellman）协商出一个共享密钥，之后所有传输的数据都用该密钥进行AES加密和解密。

Python代码实现框架及关键代码段解释

数据处理模块：

JSON序列化与反序列化：

import json


def serialize_data(data):
    return json.dumps(data).encode('utf - 8')


def deserialize_data(data):
    return json.loads(data.decode('utf - 8'))

CRC校验：

import binascii
import struct


def calculate_crc(data):
    return binascii.crc32(data) & 0xFFFFFFFF


def verify_crc(data, received_crc):
    calculated_crc = calculate_crc(data)
    return calculated_crc == received_crc

连接管理模块：

TCP连接：

import socket


def create_tcp_socket():
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    return sock


def connect_to_server(sock, host, port):
    sock.connect((host, port))


def send_data(sock, data):
    sock.sendall(data)


def receive_data(sock, buffer_size):
    return sock.recv(buffer_size)

加密模块：

AES加密与解密：

from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad


def generate_aes_key():
    from Crypto.Random import get_random_bytes
    return get_random_bytes(16)


def encrypt_data(key, data):
    cipher = AES.new(key, AES.MODE_CBC)
    ct_bytes = cipher.encrypt(pad(data, AES.block_size))
    iv = cipher.iv
    return iv + ct_bytes


def decrypt_data(key, data):
    iv = data[:AES.block_size]
    ct = data[AES.block_size:]
    cipher = AES.new(key, AES.MODE_CBC, iv)
    pt = unpad(cipher.decrypt(ct), AES.block_size)
    return pt

整体通信示例：

def main():
    # 假设这是要发送的数据
    data = {
        "device_info": {
            "device_id": "123456",
            "device_type": "sensor"
        },
        "timestamp": "2024 - 01 - 01T12:00:00Z",
        "business_data": [1, 2, 3]
    }
    serialized_data = serialize_data(data)
    crc = calculate_crc(serialized_data)
    full_data = serialized_data + struct.pack('!I', crc)

    key = generate_aes_key()
    encrypted_data = encrypt_data(key, full_data)

    sock = create_tcp_socket()
    connect_to_server(sock, '127.0.0.1', 12345)
    send_data(sock, encrypted_data)

    received_data = receive_data(sock, 1024)
    decrypted_data = decrypt_data(key, received_data)
    received_serialized_data = decrypted_data[: - 4]
    received_crc = struct.unpack('!I', decrypted_data[-4:])[0]
    if verify_crc(received_serialized_data, received_crc):
        received_json = deserialize_data(received_serialized_data)
        print(received_json)
    else:
        print('CRC verification failed')
    sock.close()


if __name__ == "__main__":
    main()

以上代码框架展示了如何实现自定义数据传输协议，包括数据格式处理、校验机制、连接管理和加密方式的实现。具体应用中，可能需要根据实际需求进行调整和优化，如心跳机制的完善、异常处理等。

星途面试题库

面试题：Python网络编程中自定义数据传输协议的设计与实现

知识考点

面试题答案

协议设计整体思路

Python代码实现框架及关键代码段解释