package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpclient;
|
|
import cn.hutool.core.lang.Assert;
|
import cn.hutool.extra.spring.SpringUtil;
|
import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi;
|
import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusDeviceConfigRespDTO;
|
import cn.iocoder.yudao.module.iot.core.enums.IotProtocolTypeEnum;
|
import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
|
import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
|
import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties.ProtocolProperties;
|
import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocol;
|
import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpclient.handler.downstream.IotModbusTcpClientDownstreamHandler;
|
import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpclient.handler.downstream.IotModbusTcpClientDownstreamSubscriber;
|
import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpclient.handler.upstream.IotModbusTcpClientUpstreamHandler;
|
import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpclient.manager.IotModbusTcpClientConfigCacheService;
|
import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpclient.manager.IotModbusTcpClientConnectionManager;
|
import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpclient.manager.IotModbusTcpClientPollScheduler;
|
import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
|
import io.vertx.core.Vertx;
|
import lombok.Getter;
|
import lombok.extern.slf4j.Slf4j;
|
import org.redisson.api.RedissonClient;
|
|
import java.util.List;
|
import java.util.Set;
|
import java.util.concurrent.TimeUnit;
|
|
/**
|
* IoT 网关 Modbus TCP Client 协议：主动轮询 Modbus 从站设备数据
|
*
|
* @author 芋道源码
|
*/
|
@Slf4j
|
public class IotModbusTcpClientProtocol implements IotProtocol {
|
|
/**
|
* 协议配置
|
*/
|
private final ProtocolProperties properties;
|
/**
|
* 服务器 ID（用于消息追踪，全局唯一）
|
*/
|
@Getter
|
private final String serverId;
|
|
/**
|
* 运行状态
|
*/
|
@Getter
|
private volatile boolean running = false;
|
|
/**
|
* Vert.x 实例
|
*/
|
private final Vertx vertx;
|
/**
|
* 配置刷新定时器 ID
|
*/
|
private Long configRefreshTimerId;
|
|
/**
|
* 连接管理器
|
*/
|
private final IotModbusTcpClientConnectionManager connectionManager;
|
/**
|
* 下行消息订阅者
|
*/
|
private IotModbusTcpClientDownstreamSubscriber downstreamSubscriber;
|
|
private final IotModbusTcpClientConfigCacheService configCacheService;
|
private final IotModbusTcpClientPollScheduler pollScheduler;
|
|
public IotModbusTcpClientProtocol(ProtocolProperties properties) {
|
IotModbusTcpClientConfig modbusTcpClientConfig = properties.getModbusTcpClient();
|
Assert.notNull(modbusTcpClientConfig, "Modbus TCP Client 协议配置（modbusTcpClient）不能为空");
|
this.properties = properties;
|
this.serverId = IotDeviceMessageUtils.generateServerId(properties.getPort());
|
|
// 初始化 Vertx
|
this.vertx = Vertx.vertx();
|
|
// 初始化 Manager
|
RedissonClient redissonClient = SpringUtil.getBean(RedissonClient.class);
|
IotDeviceCommonApi deviceApi = SpringUtil.getBean(IotDeviceCommonApi.class);
|
IotDeviceMessageService messageService = SpringUtil.getBean(IotDeviceMessageService.class);
|
this.configCacheService = new IotModbusTcpClientConfigCacheService(deviceApi);
|
this.connectionManager = new IotModbusTcpClientConnectionManager(redissonClient, vertx,
|
messageService, configCacheService, serverId);
|
|
// 初始化 Handler
|
IotModbusTcpClientUpstreamHandler upstreamHandler = new IotModbusTcpClientUpstreamHandler(messageService, serverId);
|
|
// 初始化轮询调度器
|
this.pollScheduler = new IotModbusTcpClientPollScheduler(vertx, connectionManager, upstreamHandler, configCacheService);
|
}
|
|
@Override
|
public String getId() {
|
return properties.getId();
|
}
|
|
@Override
|
public IotProtocolTypeEnum getType() {
|
return IotProtocolTypeEnum.MODBUS_TCP_CLIENT;
|
}
|
|
@Override
|
public void start() {
|
if (running) {
|
log.warn("[start][IoT Modbus TCP Client 协议 {} 已经在运行中]", getId());
|
return;
|
}
|
|
try {
|
// 1.1 首次加载配置
|
refreshConfig();
|
// 1.2 启动配置刷新定时器
|
int refreshInterval = properties.getModbusTcpClient().getConfigRefreshInterval();
|
configRefreshTimerId = vertx.setPeriodic(
|
TimeUnit.SECONDS.toMillis(refreshInterval),
|
id -> refreshConfig()
|
);
|
running = true;
|
log.info("[start][IoT Modbus TCP Client 协议 {} 启动成功，serverId={}]", getId(), serverId);
|
|
// 2. 启动下行消息订阅者
|
IotMessageBus messageBus = SpringUtil.getBean(IotMessageBus.class);
|
IotModbusTcpClientDownstreamHandler downstreamHandler = new IotModbusTcpClientDownstreamHandler(connectionManager,
|
configCacheService);
|
this.downstreamSubscriber = new IotModbusTcpClientDownstreamSubscriber(this, downstreamHandler, messageBus);
|
this.downstreamSubscriber.start();
|
} catch (Exception e) {
|
log.error("[start][IoT Modbus TCP Client 协议 {} 启动失败]", getId(), e);
|
stop0();
|
throw e;
|
}
|
}
|
|
@Override
|
public void stop() {
|
if (!running) {
|
return;
|
}
|
stop0();
|
}
|
|
private void stop0() {
|
// 1. 停止下行消息订阅者
|
if (downstreamSubscriber != null) {
|
try {
|
downstreamSubscriber.stop();
|
log.info("[stop][IoT Modbus TCP Client 协议 {} 下行消息订阅者已停止]", getId());
|
} catch (Exception e) {
|
log.error("[stop][IoT Modbus TCP Client 协议 {} 下行消息订阅者停止失败]", getId(), e);
|
}
|
downstreamSubscriber = null;
|
}
|
|
// 2.1 取消配置刷新定时器
|
if (configRefreshTimerId != null) {
|
vertx.cancelTimer(configRefreshTimerId);
|
configRefreshTimerId = null;
|
}
|
// 2.2 停止轮询调度器
|
pollScheduler.stopAll();
|
// 2.3 关闭所有连接
|
connectionManager.closeAll();
|
|
// 3. 关闭 Vert.x 实例
|
if (vertx != null) {
|
try {
|
vertx.close().result();
|
log.info("[stop][IoT Modbus TCP Client 协议 {} Vertx 已关闭]", getId());
|
} catch (Exception e) {
|
log.error("[stop][IoT Modbus TCP Client 协议 {} Vertx 关闭失败]", getId(), e);
|
}
|
}
|
running = false;
|
log.info("[stop][IoT Modbus TCP Client 协议 {} 已停止]", getId());
|
}
|
|
/**
|
* 刷新配置
|
*/
|
private synchronized void refreshConfig() {
|
try {
|
// 1. 从 biz 拉取最新配置（API 失败时返回 null）
|
List<IotModbusDeviceConfigRespDTO> configs = configCacheService.refreshConfig();
|
if (configs == null) {
|
log.warn("[refreshConfig][API 失败，跳过本轮刷新]");
|
return;
|
}
|
log.debug("[refreshConfig][获取到 {} 个 Modbus 设备配置]", configs.size());
|
|
// 2. 更新连接和轮询任务
|
for (IotModbusDeviceConfigRespDTO config : configs) {
|
try {
|
// 2.1 确保连接存在
|
connectionManager.ensureConnection(config);
|
// 2.2 更新轮询任务
|
pollScheduler.updatePolling(config);
|
} catch (Exception e) {
|
log.error("[refreshConfig][处理设备配置失败, deviceId={}]", config.getDeviceId(), e);
|
}
|
}
|
|
// 3. 清理已删除设备的资源
|
Set<Long> removedDeviceIds = configCacheService.cleanupRemovedDevices(configs);
|
for (Long deviceId : removedDeviceIds) {
|
pollScheduler.stopPolling(deviceId);
|
connectionManager.removeDevice(deviceId);
|
}
|
} catch (Exception e) {
|
log.error("[refreshConfig][刷新配置失败]", e);
|
}
|
}
|
|
}
|
