Java多线程写入PLC出现数据丢失问题求助

问题背景 #

通过8个并发线程调用writeASICRatio函数执行PLC写入操作：

• 单线程执行该函数约需2分钟，串行执行8次总耗时约16分钟，且串行执行时一切正常。
• 多线程执行时，仅能成功完成1个线程的写入，有时甚至全部失败，出现数据丢失情况。

writeASICRatio函数内部逻辑：向PLC发送128条指令（含验证步骤），PLC在10毫秒后返回响应，其中验证步骤示例代码如下：

//Note: Start command mode
//Note: ON - WRITE - SIZE = 3 - Value[0] = 0x72, Value[1] = 0xF5, Value[2] = 0xA2
writeCmdASIC(sID, (short) 1, (short) 0, 3, "72", "F5", "A2", "0", "0", "0", "0", "0", "0", "0", "0", "0", 
                        mClient, 0);

//Note: Write read start command mode.
//Note: ON - READ - SIZE = 1 - Value = 0        
writeCmdASIC(sID, (short)1, (short)1, 1, "0", "0", "0", "0", "0", "0", "0", "0", "0", "0", "0", "0", mClient, 0);

//Note: Read return of start command mode.
responseData = Integer.parseInt(main.getValDataA().getValueAt(sID, 0).toString());

if (responseData == 72)
{
    //Note: Do nothing
}
else
{
    message += "M" + manID + "S" + sID + ".Start command mode not received correctly. Aborting calibration mode... <br>";
    return "";
}

其余127条为纯写入指令。多线程创建及启动代码如下：

Thread thread0 = new Thread(() -> {
                        try {
                            writeASICRatio(modbusClientTmp, manID, 0);
                        } catch (Exception ex) {
                            Logger.getLogger(MainCalibration.class.getName()).log(Level.SEVERE, null, ex);
                        }
                    });
                    
                    Thread thread1 = new Thread(() -> {
                        try {
                            writeASICRatio(modbusClientTmp, manID, 1);
                        } catch (Exception ex) {
                            Logger.getLogger(MainCalibration.class.getName()).log(Level.SEVERE, null, ex);
                        }
                    });
                    
                    Thread thread2 = new Thread(() -> {
                        try {
                            writeASICRatio(modbusClientTmp, manID, 2);
                        } catch (Exception ex) {
                            Logger.getLogger(MainCalibration.class.getName()).log(Level.SEVERE, null, ex);
                        }
                    });
                    
                    Thread thread3 = new Thread(() -> {
                        try {
                            writeASICRatio(modbusClientTmp, manID, 3);
                        } catch (Exception ex) {
                            Logger.getLogger(MainCalibration.class.getName()).log(Level.SEVERE, null, ex);
                        }
                    });
                    
                    Thread thread4 = new Thread(() -> {
                        try {
                            writeASICRatio(modbusClientTmp, manID, 4);
                        } catch (Exception ex) {
                            Logger.getLogger(MainCalibration.class.getName()).log(Level.SEVERE, null, ex);
                        }
                    });
                    
                    Thread thread5 = new Thread(() -> {
                        try {
                            writeASICRatio(modbusClientTmp, manID, 5);
                        } catch (Exception ex) {
                            Logger.getLogger(MainCalibration.class.getName()).log(Level.SEVERE, null, ex);
                        }
                    });
                    
                    Thread thread6 = new Thread(() -> {
                        try {
                            writeASICRatio(modbusClientTmp, manID, 6);
                        } catch (Exception ex) {
                            Logger.getLogger(MainCalibration.class.getName()).log(Level.SEVERE, null, ex);
                        }
                    });
                    
                    Thread thread7 = new Thread(() -> {
                        try {
                            writeASICRatio(modbusClientTmp, manID, 7);
                        } catch (Exception ex) {
                            Logger.getLogger(MainCalibration.class.getName()).log(Level.SEVERE, null, ex);
                        }
                    });

                    
                    // Start both threads
                    thread0.start();
                    thread1.start();
                    thread2.start();
                    thread3.start();
                    thread4.start();
                    thread5.start();
                    thread6.start();
                    thread7.start();
                    
                    //Note: Wait for both threads to complete
                    try
                    {
                        thread0.join();
                        thread1.join();
                        thread2.join();
                        thread3.join();
                        thread4.join();
                        thread5.join();
                        thread6.join();
                        thread7.join();
                    }
                    catch(InterruptedException ex)
                    {
                        ex.printStackTrace();
                    } 

问题原因分析 #

• Modbus客户端非线程安全：所有线程共用同一个modbusClientTmp实例，Modbus协议客户端通常不支持并发写入，交叉发送的指令会导致PLC接收的指令乱序，响应数据被多个线程争抢读取，验证逻辑无法匹配正确响应。
• 共享数据结构无同步控制：main.getValDataA()是多个线程共享的数据表格，读取操作未加同步锁，可能出现线程A写入的响应被线程B读取，或数据未更新就读取的情况，导致校验失败。
• PLC并发处理能力限制：多数PLC对同时接收的指令数量有上限，8个线程同时发送大量指令会导致PLC指令队列溢出，部分请求被丢弃，响应超时或数据覆盖。

解决建议 #

1. 为每个线程分配独立Modbus连接 #

每个线程创建专属的Modbus客户端实例，避免共享连接的线程安全问题：

// 定义创建新连接的方法
private ModbusClient createNewModbusClient() throws Exception {
    // 初始化并返回新的ModbusClient实例
    ModbusClient client = new ModbusClient(...); // 替换为实际初始化逻辑
    client.connect();
    return client;
}

// 线程创建逻辑修改为：
Thread thread0 = new Thread(() -> {
    try (ModbusClient client = createNewModbusClient()) { // 自动关闭连接
        writeASICRatio(client, manID, 0);
    } catch (Exception ex) {
        Logger.getLogger(MainCalibration.class.getName()).log(Level.SEVERE, null, ex);
    }
});

注意：需确认PLC支持同时建立8个连接，若连接数有限，可调整线程数或使用连接池管理。

2. 对共享资源添加同步锁 #

通过锁机制确保同一时间只有一个线程执行PLC交互操作：

• 全局锁方案：对writeASICRatio函数添加synchronized修饰符：

public synchronized void writeASICRatio(ModbusClient client, int manID, int sID) {
    // 原函数逻辑
}

• 分片锁方案：按sID分配独立锁，减少锁竞争：

private final Lock[] sidLocks = new ReentrantLock[8];

// 初始化锁
public void initSidLocks() {
    for (int i = 0; i < sidLocks.length; i++) {
        sidLocks[i] = new ReentrantLock();
    }
}

// 函数内使用锁
public void writeASICRatio(ModbusClient client, int manID, int sID) {
    sidLocks[sID].lock();
    try {
        // 原函数逻辑：写入指令、读取响应、验证
    } finally {
        sidLocks[sID].unlock();
    }
}

3. 优化响应读取逻辑 #

增加等待或重试机制，确保PLC响应已同步到共享表格：

// 替换原响应读取代码
int retryCount = 0;
int responseData = -1;
final int MAX_RETRIES = 5;
final long WAIT_MS = 20;

while (retryCount < MAX_RETRIES && responseData != 72) {
    Thread.sleep(WAIT_MS); // 等待响应同步
    responseData = Integer.parseInt(main.getValDataA().getValueAt(sID, 0).toString());
    retryCount++;
}

if (responseData != 72) {
    message += "M" + manID + "S" + sID + ".Start command mode not received correctly. Aborting calibration mode... <br>";
    return "";
}

4. 使用线程池管理并发 #

替换手动创建线程的方式，用ExecutorService统一管理线程，方便监控执行结果：

ExecutorService executor = Executors.newFixedThreadPool(8);
List<Future<?>> taskFutures = new ArrayList<>();

for (int sID = 0; sID < 8; sID++) {
    final int targetSID = sID;
    taskFutures.add(executor.submit(() -> {
        try (ModbusClient client = createNewModbusClient()) {
            writeASICRatio(client, manID, targetSID);
        } catch (Exception ex) {
            Logger.getLogger(MainCalibration.class.getName()).log(Level.SEVERE, null, ex);
        }
    }));
}

// 等待所有任务完成
for (Future<?> future : taskFutures) {
    try {
        future.get();
    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }
}

executor.shutdown();

内容的提问来源于stack exchange，提问作者R.Lerma