Wireshark 抓包实战：以太网温湿度传感器 Modbus TCP 异常重传与连接复位排查

# 宿主侧，按传感器 IP 过滤存盘，循环分片
sudo tcpdump -i eth0 -nn -s0 -G 3600 -W 4 \
  -w /data/pcap/modbus_%Y%m%d_%H%M%S.pcap \
  "host 192.168.10.21 and tcp port 502"

host 192.168.10.21 and tcp port 502

# 基本聚焦
ip.addr == 192.168.10.21 and tcp.port == 502

# 重传 / 乱序 / 零窗口
tcp.analysis.retransmission
tcp.analysis.duplicate_ack
tcp.analysis.zero_window_probe || tcp.zero_window || tcp.window_size == 0

# 连接复位与异常关闭
tcp.flags.reset == 1
tcp.flags.fin == 1
tcp.analysis.connection_reset  # 专家信息辅助

# Modbus 层
modbus
mbtcp.trans_id == 5
modbus.func_code == 3
modbus.exception_code != 0

# 按流跟踪
tcp.stream eq 12

C->S  [PSH,ACK] MBAP/TCP Read Holding Registers 40001 len=6
S->C  [ACK]
... 无响应数据
C->S  [Retransmission] 同上段
C->S  [Retransmission]
C->S  [Retransmission]
C  RTO 退避后放弃，应用层超时，关闭 socket
C->S  [FIN,ACK] 或 close() 后无 FIN 直接新连接

C->S  [ACK] 空闲 300s
中间设备（防火墙/NAT/交换机状态检测）老化会话
C 下次轮询发 [PSH,ACK]
S 或中间设备回 [RST,ACK]
C 应用层报 ConnectionResetError / WinError 10054

C 持长连接轮询
S 固件会话超时（如 60–180s 无活动）或最大连接数限制
S->C [RST,ACK] 或 [FIN,ACK]
C pymodbus 复用旧 socket 继续写 -> EPIPE/Broken pipe/10054

S->C [FIN,ACK] 固件重启或任务复位
C TCP 栈回 [ACK]，进入 CLOSE_WAIT，应用未调用 close()
后续轮询继续在旧 fd 上 send()，返回错误或未报错但丢弃
新轮询逻辑新建连接，出现 TIME_WAIT 堆积

import asyncio, logging, time
from pymodbus.client import AsyncModbusTcpClient
from pymodbus.exceptions import ModbusIOException, ConnectionException

log = logging.getLogger("modbus_poller")

class Poller:
    def __init__(self, host, port=502, slave=1, poll=5.0, timeout=2.0):
        self.host, self.port, self.slave = host, port, slave
        self.poll, self.timeout = poll, timeout
        self.client = None
        self._conn_lock = asyncio.Lock()

    async def _ensure_conn(self):
        # 惰性建连 + 健康探测
        if self.client is None or not self.client.connected:
            async with self._conn_lock:
                if self.client is not None:
                    try: self.client.close()
                    except Exception: pass
                self.client = AsyncModbusTcpClient(
                    self.host, port=self.port, timeout=self.timeout,
                    retry_on_empty=True, retries=2,
                )
                # pymodbus 3.x 走 asyncio 事件循环
                ok = await self.client.connect()
                if not ok:
                    raise ConnectionException("connect failed")
                # 可选：设置 sockopt keepalive（见下）

    async def read(self):
        for attempt in range(3):
            try:
                await self._ensure_conn()
                resp = await self.client.read_holding_registers(0, count=2, slave=self.slave)
                # pymodbus: resp None / isError / 异常响应都要处理
                if resp is None or resp.isError():
                    raise ModbusIOException(f"bad resp: {resp}")
                temp = resp.registers[0] * 0.1
                humid = resp.registers[1] * 0.1
                return {"temp": temp, "humid": humid, "quality": 0}
            except (ConnectionException, ModbusIOException, OSError) as e:
                log.warning("poll fail attempt=%d: %s", attempt, e)
                # 强制销毁，下一轮重建
                try: self.client.close()
                except Exception: pass
                self.client = None
                await asyncio.sleep(min(2 ** attempt, 8))   # 退避，别自旋
        return {"quality": 2}

    async def run(self):
        while True:
            t0 = time.monotonic()
            res = await self.read()
            # 上报/缓存/补传逻辑略
            dt = time.monotonic() - t0
            await asyncio.sleep(max(0.0, self.poll - dt))

import socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
# Linux
s.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 30)
s.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 10)
s.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 3)