企业官网建设流程全解析

八、性能分析与优化

8.1 性能基准测试

8.1.1 理论最大吞吐量

RTU @9600 bps：

• 每字节时间：10位/9600 = 1.0417 ms

• 请求帧（读10个寄存器）：地址1 + 功能码3 + 起始地址2 + 数量2 + CRC2 = 8字节 → 8.33 ms

• 3.5字符间隔：3.65 ms

• 响应帧（20字节数据）：地址1 + 功能码3 + 字节数1 + 数据20 + CRC2 = 25字节 → 26.04 ms

• 单次总耗时：8.33 + 3.65 + 26.04 =~38 ms

• 每秒次数：1000/38 ≈26 次/秒

• 寄存器吞吐量：26 × 10 =260 寄存器/秒

RTU @115200 bps（线长<50米）：

• 每字节时间：10/115200 = 0.0868 ms

• 单次总耗时：~4 ms

• 寄存器吞吐量：~2500 寄存器/秒

TCP @100 Mbps（局域网）：

• 网络延迟 <1 ms，协议栈开销 <0.5 ms

• 寄存器吞吐量：>50000 寄存器/秒（受限于应用层处理）

8.1.2 影响性能的因素

8.2 RTU优化技巧

8.2.1 提高波特率

# 前提：所有设备支持且线缆长度足够短 ser = serial.Serial('/dev/ttyUSB0', baudrate=115200, timeout=0.5)

8.2.2 批量读取

# 不好：每次读1个寄存器，来回20次 for addr in addresses: val = read_holding_register(addr) # 好：一次读10个连续寄存器 vals = read_holding_registers(start_addr=0x0000, count=10)

8.2.3 减少轮询频率

# 静态数据（设备型号、版本）只读一次 device_info = read_device_info() # 动态数据（温度、压力）根据需要轮询 temperature = read_temperature() # 每2秒

8.2.4 使用功能码23（原子读写）

# 传统方式：读-处理-写（3次往返） old_val = read_register(addr) new_val = calculate(old_val) write_register(addr, new_val) # 读写方式：1次往返（且无竞争） result = read_write_registers(read_addr, read_count, write_addr, write_data)

8.3 TCP优化技巧

8.3.1 禁用Nagle算法

# Nagle会合并小报文，延迟最多40ms sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)

8.3.2 使用连接池

from queue import Queue class ModbusTCPPool: def __init__(self, host, port, pool_size=10): self.pool = Queue(maxsize=pool_size) for _ in range(pool_size): client = ModbusTCPClient(host, port) client.connect() self.pool.put(client) def acquire(self): return self.pool.get() def release(self, client): self.pool.put(client)

8.3.3 调整TCP缓冲区

# 增大接收缓冲区（减少丢包重传） sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 65536) sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 65536)

8.3.4 广域网优化

# 对于高延迟链路，增加超时时间 client = ModbusTCPClient(host, port, timeout_ms=5000) # 5秒 # 使用管道化请求（需服务器支持） # 发送多个请求而不等待响应（事务ID区分）

8.4 性能测试脚本

import time def benchmark_rtu(port, count=100): ser = serial.Serial(port, 9600, timeout=1) start = time.time() for i in range(count): # 发送读10个寄存器请求 frame = bytes([0x01, 0x03, 0x00, 0x00, 0x00, 0x0A]) crc = modbus_crc(frame) ser.write(frame + crc.to_bytes(2, 'little')) response = ser.read(100) elapsed = time.time() - start print(f"RTU: {count} requests in {elapsed:.2f}s -> {count/elapsed:.1f} req/s") def benchmark_tcp(host, count=1000): client = ModbusTCPClient(host) start = time.time() for i in range(count): client.read_holding_registers(0, 10) elapsed = time.time() - start print(f"TCP: {count} requests in {elapsed:.2f}s -> {count/elapsed:.1f} req/s") client.close()