# Agent数据分析V3.0.1.7

Junjie

3 天以前 63b01db83d9aad8a15276b4236a9a22e4aeef065

 src/main/java/com/zy/core/network/ZyStationConnectDriver.java

@@ -3,20 +3,23 @@
import com.zy.asrs.entity.DeviceConfig;
import com.zy.common.utils.RedisUtil;
import com.zy.core.ThreadHandler;
import com.zy.core.enums.RedisKeyType;
import com.zy.core.model.CommandResponse;
import com.zy.core.model.command.StationCommand;
import com.zy.core.network.api.ZyStationConnectApi;
import com.zy.core.network.entity.ZyStationStatusEntity;
import java.util.List;
import com.zy.core.network.fake.ZyStationFakeConnect;
import com.zy.core.network.fake.ZyStationFakeSegConnect;
import com.zy.core.network.real.ZyStationRealConnect;
import com.zy.core.network.real.ZyStationV3RealConnect;
import com.zy.core.network.real.ZyStationV5RealConnect;
import lombok.extern.slf4j.Slf4j;
import java.util.Collections;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.UUID;

/**
 * 输送站连接驱动
@@ -24,15 +27,21 @@
@Slf4j
public class ZyStationConnectDriver implements ThreadHandler {

    private static final ZyStationFakeConnect zyStationFakeConnect = new ZyStationFakeConnect();
    private static final ZyStationFakeSegConnect zyStationFakeSegConnect = new ZyStationFakeSegConnect();
    private static final long SEND_LOCK_WARN_MS = 500L;
    private static final long SEND_COST_WARN_MS = 5_000L;
    private static final long SEND_LOCK_POLL_MS = 20L;
    private static final long SEND_LOCK_EXPIRE_SECONDS = 15L;

    private boolean connected = false;
    private volatile boolean connected = false;
    private volatile boolean connecting = false;
    private DeviceConfig deviceConfig;
    private RedisUtil redisUtil;
    private ZyStationConnectApi zyStationConnectApi;
    private volatile ZyStationConnectApi zyStationConnectApi;
    private volatile boolean closed = false;
    private volatile boolean fakeConfigUnsupported = false;
    private ScheduledExecutorService executor;
    private final Object connectLock = new Object();

    public ZyStationConnectDriver(DeviceConfig deviceConfig, RedisUtil redisUtil) {
        this.deviceConfig = deviceConfig;
@@ -46,25 +55,45 @@

    @Override
    public boolean connect() {
        if (deviceConfig.getFake() == 0) {
            if ("ZyStationV3Thread".equals(deviceConfig.getThreadImpl())) {
                zyStationConnectApi = new ZyStationV3RealConnect(deviceConfig, redisUtil);
            } else {
                zyStationConnectApi = new ZyStationRealConnect(deviceConfig, redisUtil);
        synchronized (connectLock) {
            if (closed) {
                return false;
            }
        } else {
            if ("ZyStationV3Thread".equals(deviceConfig.getThreadImpl())) {
                zyStationFakeSegConnect.addFakeConnect(deviceConfig, redisUtil);
                zyStationConnectApi = zyStationFakeSegConnect;
            } else {
                zyStationFakeConnect.addFakeConnect(deviceConfig, redisUtil);
                zyStationConnectApi = zyStationFakeConnect;
            if (fakeConfigUnsupported) {
                return false;
            }
            if (connected && zyStationConnectApi != null) {
                return true;
            }

            connecting = true;
            try {
                ZyStationConnectApi connectApi;
                if (deviceConfig.getFake() == 0) {
                    if ("ZyStationV3Thread".equals(deviceConfig.getThreadImpl())) {
                        connectApi = new ZyStationV3RealConnect(deviceConfig, redisUtil);
                    } else if ("ZyStationV5Thread".equals(deviceConfig.getThreadImpl())) {
                        connectApi = new ZyStationV5RealConnect(deviceConfig, redisUtil);
                    } else {
                        connectApi = new ZyStationRealConnect(deviceConfig, redisUtil);
                    }
                } else {
                    zyStationFakeSegConnect.addFakeConnect(deviceConfig, redisUtil);
                    connectApi = zyStationFakeSegConnect;
                }

                boolean connect = connectApi.connect();
                connected = connect;
                if (connect) {
                    zyStationConnectApi = connectApi;
                } else {
                    zyStationConnectApi = null;
                }
                return connect;
            } finally {
                connecting = false;
            }
        }

        boolean connect = zyStationConnectApi.connect();
        connected = connect;
        return connect;
    }

    @Override
@@ -74,11 +103,13 @@
        if (ex != null) {
            try { ex.shutdownNow(); } catch (Exception ignore) {}
        }
        if (zyStationConnectApi != null) {
            zyStationConnectApi.disconnect();
        ZyStationConnectApi connectApi = zyStationConnectApi;
        if (connectApi != null) {
            connectApi.disconnect();
            zyStationConnectApi = null;
        }
        connected = false;
        connecting = false;
    }

    public void start() {
@@ -107,21 +138,101 @@
    }

    public List<ZyStationStatusEntity> getStatus() {
        if (zyStationConnectApi == null) {
            return null;
        ZyStationConnectApi connectApi = zyStationConnectApi;
        if (!connected || connecting || connectApi == null) {
            return Collections.emptyList();
        }
        return zyStationConnectApi.getStatus(deviceConfig.getDeviceNo());
        return connectApi.getStatus(deviceConfig.getDeviceNo());
    }

    public CommandResponse sendCommand(StationCommand command) {
        return zyStationConnectApi.sendCommand(deviceConfig.getDeviceNo(), command);
        ZyStationConnectApi connectApi = zyStationConnectApi;
        if (!connected || connecting || connectApi == null) {
            return new CommandResponse(false, "设备未连接，命令下发失败");
        }
        String lockKey = buildStationExecuteLockKey();
        String lockToken = UUID.randomUUID().toString();
        long lockWaitStart = System.currentTimeMillis();
        int waitRounds = 0;
        if (redisUtil != null) {
            while (true) {
                if (redisUtil.trySetStringIfAbsent(lockKey, lockToken, SEND_LOCK_EXPIRE_SECONDS)) {
                    break;
                }
                waitRounds++;
                try {
                    Thread.sleep(SEND_LOCK_POLL_MS);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
        long lockWaitCost = System.currentTimeMillis() - lockWaitStart;
        if (lockWaitCost >= SEND_LOCK_WARN_MS) {
            log.warn("输送命令等待设备发送锁超时，deviceNo={}, taskNo={}, stationId={}, targetStaNo={}, waitMs={}, waitRounds={}, lockKey={}",
                    deviceConfig == null ? null : deviceConfig.getDeviceNo(),
                    command == null ? null : command.getTaskNo(),
                    command == null ? null : command.getStationId(),
                    command == null ? null : command.getTargetStaNo(),
                    lockWaitCost,
                    waitRounds,
                    lockKey);
        }
        long sendStart = System.currentTimeMillis();
        try {
            return connectApi.sendCommand(deviceConfig.getDeviceNo(), command);
        } finally {
            releaseDeviceSendLock(lockKey, lockToken);
            long sendCostMs = System.currentTimeMillis() - sendStart;
            if (sendCostMs >= SEND_COST_WARN_MS) {
                log.warn("输送命令底层发送耗时过长，deviceNo={}, taskNo={}, stationId={}, targetStaNo={}, sendCostMs={}",
                        deviceConfig == null ? null : deviceConfig.getDeviceNo(),
                        command == null ? null : command.getTaskNo(),
                        command == null ? null : command.getStationId(),
                        command == null ? null : command.getTargetStaNo(),
                        sendCostMs);
            }
        }
    }

    private String buildStationExecuteLockKey() {
        Integer deviceNo = deviceConfig == null ? null : deviceConfig.getDeviceNo();
        return RedisKeyType.STATION_EXECUTE_COMMAND_LOCK.key + ":" + deviceNo;
    }

    private void releaseDeviceSendLock(String lockKey, String lockToken) {
        if (redisUtil == null || lockKey == null) {
            return;
        }
        try {
            redisUtil.compareAndDelete(lockKey, lockToken);
        } catch (Exception e) {
            log.warn("释放输送设备发送锁失败，lockKey={}", lockKey, e);
        }
    }

    public CommandResponse sendOriginCommand(String address, short[] data) {
        return zyStationConnectApi.sendOriginCommand(address, data);
        ZyStationConnectApi connectApi = zyStationConnectApi;
        if (!connected || connecting || connectApi == null) {
            return new CommandResponse(false, "设备未连接，原始命令下发失败");
        }
        return connectApi.sendOriginCommand(address, data);
    }

    public boolean clearTaskBufferSlot(Integer stationId, Integer slotIdx) {
        ZyStationConnectApi connectApi = zyStationConnectApi;
        if (!connected || connecting || connectApi == null) {
            return false;
        }
        CommandResponse response = connectApi.clearTaskBufferSlot(deviceConfig.getDeviceNo(), stationId, slotIdx);
        return response != null && Boolean.TRUE.equals(response.getResult());
    }

    public byte[] readOriginCommand(String address, int length) {
        return zyStationConnectApi.readOriginCommand(address, length);
        ZyStationConnectApi connectApi = zyStationConnectApi;
        if (!connected || connecting || connectApi == null) {
            return new byte[0];
        }
        return connectApi.readOriginCommand(address, length);
    }
}