Python：xdrlib 库高级用法举例和应用详解

import xdrlib  # 直接导入即可使用

import xdrlib

# 创建一个Packer对象用于数据打包
packer = xdrlib.Packer()

# 场景：将一个整数和一个字符串打包发送
integer_value = 42
string_value = "Hello, XDR!"

# 将整数和字符串打包
packer.pack_int(integer_value)  # 打包一个整数
packer.pack_string(string_value.encode('utf-8'))  # 打包一个字符串

# 获取打包后的二进制数据
packed_data = packer.get_buffer()

# 创建一个Unpacker对象用于数据解包
unpacker = xdrlib.Unpacker(packed_data)

# 解包数据
unpacked_int = unpacker.unpack_int()  # 解包整数
unpacked_string = unpacker.unpack_string().decode('utf-8')  # 解包字符串

# 验证解包后的数据是否与原始数据一致
assert integer_value == unpacked_int
assert string_value == unpacked_string

print("数据打包和解包成功")

import xdrlib

# 创建一个Packer对象用于数据打包
packer = xdrlib.Packer()

# 场景：将一个复合结构打包发送(一个整数，一个浮点数，一个字符串列表)
integer_value = 123
float_value = 456.78
string_list = ["Alice", "Bob", "Charlie"]

# 将复合数据结构打包
packer.pack_int(integer_value)
packer.pack_float(float_value)

# 打包字符串列表
packer.pack_int(len(string_list)) # 先打包列表长度
for item in string_list:
    packer.pack_string(item.encode('utf-8'))

# 获取打包后的二进制数据
packed_data = packer.get_buffer()

# 创建一个Unpacker对象用于数据解包
unpacker = xdrlib.Unpacker(packed_data)

# 解包数据
unpacked_int = unpacker.unpack_int()
unpacked_float = unpacker.unpack_float()

# 解包字符串列表
string_list_size = unpacker.unpack_int()
unpacked_string_list = [unpacker.unpack_string().decode('utf-8') for _ in range(string_list_size)]

# 验证解包后的数据是否与原始数据一致
assert integer_value == unpacked_int
assert float_value == unpacked_float
assert string_list == unpacked_string_list

print("复合结构数据打包和解包成功")

import xdrlib
import socket

# 假设我们有一个简单的客户端-服务器模型
# 服务器端
def server():
    # 创建一个TCP/IP 套接字对象
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.bind(('localhost', 9999))
    server_socket.listen(1)

    print("等待客户端连接...")
    connection, client_address = server_socket.accept()

    try:
        # 接收客户端的数据
        data = connection.recv(1024)
        
        # 解包接收的数据
        unpacker = xdrlib.Unpacker(data)
        received_int = unpacker.unpack_int()
        received_string = unpacker.unpack_string().decode('utf-8')

        print(f"接收到的数据: 整数={received_int}, 字符串='{received_string}'")

        # 处理后返回响应
        packer = xdrlib.Packer()
        packer.pack_string(b"数据接收成功")
        connection.sendall(packer.get_buffer())

    finally:
        connection.close()


# 客户端
def client():
    # 创建一个TCP/IP 套接字对象
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client_socket.connect(('localhost', 9999))

    # 打包待发送的数据
    packer = xdrlib.Packer()
    packer.pack_int(999)
    packer.pack_string(b"你好，服务器！")

    # 发送打包的数据
    client_socket.sendall(packer.get_buffer())

    # 接收服务器响应
    response_data = client_socket.recv(1024)

    # 解包响应数据
    unpacker = xdrlib.Unpacker(response_data)
    response_string = unpacker.unpack_string().decode('utf-8')

    print(f"服务器响应: {response_string}")

    client_socket.close()

# 运行服务器和客户端
import threading
server_thread = threading.Thread(target=server)
server_thread.start()

client_thread = threading.Thread(target=client)
client_thread.start()

server_thread.join()
client_thread.join()