使用Python和snap7异步读取西门子S7-1500 PLC时解决‘CLI : Job pending’错误及实现低延迟批量读取的方案

你遇到的b'CLI : Job pending'错误本质是因为snap7的Client实例不是线程安全的——你所有的异步任务都在共用同一个全局Client对象，当多个线程同时用它发起请求时，PLC或snap7库会因为当前已有未完成的任务而拒绝新请求，就抛出了这个错误。另外你在sync_read_plc里每次读完就断开连接，第一个线程断开后，后面的线程再用这个Client也会出问题。

下面给你两个针对性的解决方案，兼顾异步并发和低延迟需求：

方案一：为每个读取任务创建独立的Client连接 #

既然共享Client会冲突，那我们就让每个线程都创建自己的Client实例，用完就销毁，彻底避免线程间的资源竞争。同时注意限制线程池的大小，不要超过PLC允许的最大连接数（西门子S7-1500一般支持几十到上百个连接，根据型号而定）。

修改后的完整代码：

import snap7
from snap7.util import get_int
import asyncio
from concurrent.futures import ThreadPoolExecutor

PLC_IP = "192.168.0.162"
RACK = 0
SLOT = 0

# 限制线程池大小，避免PLC连接数过载
executor = ThreadPoolExecutor(max_workers=10)

# 同步读取函数：每个任务独立创建Client
def sync_read_plc(db_number, offset, length):
    client = None
    try:
        client = snap7.client.Client()
        client.connect(PLC_IP, RACK, SLOT)
        if not client.get_connected():
            raise ConnectionError(f"Failed to connect to PLC for tag ({db_number}, {offset})")
        data = client.db_read(db_number, offset, length)
        return get_int(data, 0)
    finally:
        # 确保资源释放
        if client:
            client.disconnect()
            client.destroy()

# 异步包装函数
async def async_read_plc(db_number, offset, length):
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(executor, sync_read_plc, db_number, offset, length)

# 批量读取多个标签
async def read_multiple_tags():
    tags = [
        (19, 278, 2),
        (19, 280, 2),
        (19, 282, 2),
        # 可以添加更多标签
    ]

    tasks = [async_read_plc(db, off, len_) for db, off, len_ in tags]
    results = await asyncio.gather(*tasks)

    for i, value in enumerate(results):
        print(f"Tag {i}: {value}")

def main():
    asyncio.run(read_multiple_tags())

if __name__ == "__main__":
    main()

方案二：批量读取连续地址（更适合1000个标签的低延迟需求） #

你的目标是读取1000个标签且延迟控制在20ms内，单标签异步读取的效率其实不如批量读取同一DB内的连续地址——因为每次PLC请求都有网络开销，批量读取能把多个标签的请求合并成一次，大幅减少网络往返时间，延迟更低，同时也从根源上避免了多线程冲突的问题。

比如你示例中的3个标签都在DB19，偏移是连续的（278、280、282，每个占2字节），我们可以一次读取6字节，再分别解析每个标签的值：

import snap7
from snap7.util import get_int
import asyncio
from concurrent.futures import ThreadPoolExecutor

PLC_IP = "192.168.0.162"
RACK = 0
SLOT = 0

executor = ThreadPoolExecutor(max_workers=1)

# 批量读取同一DB内的连续标签
def sync_read_batch_tags():
    client = None
    try:
        client = snap7.client.Client()
        client.connect(PLC_IP, RACK, SLOT)
        if not client.get_connected():
            raise ConnectionError("Failed to connect to PLC")
        
        # 读取DB19，从偏移278开始，共6字节（3个int，每个2字节）
        data = client.db_read(19, 278, 6)
        
        # 解析每个标签的值
        values = [
            get_int(data, 0),   # 对应偏移278的标签
            get_int(data, 2),   # 对应偏移280的标签
            get_int(data, 4)    # 对应偏移282的标签
        ]
        return values
    finally:
        if client:
            client.disconnect()
            client.destroy()

async def async_read_batch():
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(executor, sync_read_batch_tags)

async def main_task():
    results = await async_read_batch()
    for i, value in enumerate(results):
        print(f"Tag {i}: {value}")

def main():
    asyncio.run(main_task())

if __name__ == "__main__":
    main()

针对1000个标签的优化建议 #

• 按DB分组：把所有标签按DB号分类，同一DB内的标签再按偏移排序，合并连续的地址段（比如偏移从100到200的所有标签，一次读取101字节）。
• 分批次批量读取：如果同一DB内的标签不连续，可以分成几个连续的块分别读取，尽量减少请求次数。
• 调整线程池大小：如果需要同时读取多个DB的批量数据，可以设置线程池大小为DB分组的数量，每个线程处理一个DB的批量读取。

这样的批量读取方式，能把1000个标签的请求从1000次降到几次或几十次，完全能满足20ms的低延迟需求。

备注：内容来源于stack exchange，提问作者debabrata-aic