+ * 0:位；1:字；2:双字（32位整数，占用连续2字）；3:浮点（IEEE754，占用连续2字）。 + * DWORD/FLOAT 底层报文与 WORD 相同（按字读取），仅解析方式不同，故复用字策略。 + *
*/ @Getter @AllArgsConstructor public enum AreaTypeEnum { /** - * 0:位 - * 1:字 + * 位（0=位号读取） */ BIT(0, "位", "areaTypeBitStrategy"), - WORD(1, "字", "areaTypeWordStrategy"); + /** + * 字 + */ + WORD(1, "字", "areaTypeWordStrategy"), + /** + * 双字（32位整数，占用连续2字，低字在前） + */ + DWORD(2, "双字", "areaTypeWordStrategy"), + /** + * 浮点（IEEE754，占用连续2字，低字在前） + */ + FLOAT(3, "浮点", "areaTypeWordStrategy"); private final int code; private final String desc; diff --git a/src/main/java/com/fjg/omorntcpfins/common/enums/FinsResponseFlagEnum.java b/src/main/java/com/fjg/omorntcpfins/common/enums/FinsResponseFlagEnum.java deleted file mode 100644 index 8935de4..0000000 --- a/src/main/java/com/fjg/omorntcpfins/common/enums/FinsResponseFlagEnum.java +++ /dev/null @@ -1,25 +0,0 @@ -package com.fjg.omorntcpfins.common.enums; - -import lombok.AllArgsConstructor; -import lombok.Getter; - -@Getter -@AllArgsConstructor -public enum FinsResponseFlagEnum { - /** - * 0000: 正常 - * 1103：命令错误 - */ - RESPONSE_FLAG_0000("0000", "正常"), - RESPONSE_FLAG_0103("0103", "命令错误") - ; - private final String code; - private final String msg; - - /** - * 校验响应是否为正常 - */ - public static Boolean isNormal(String responseFlag) { - return RESPONSE_FLAG_0000.getCode().equals(responseFlag); - } -} diff --git a/src/main/java/com/fjg/omorntcpfins/common/enums/MemoryAreaEnum.java b/src/main/java/com/fjg/omorntcpfins/common/enums/MemoryAreaEnum.java new file mode 100644 index 0000000..abe5868 --- /dev/null +++ b/src/main/java/com/fjg/omorntcpfins/common/enums/MemoryAreaEnum.java @@ -0,0 +1,106 @@ +package com.fjg.omorntcpfins.common.enums; + +import lombok.AllArgsConstructor; +import lombok.Getter; + +/** + * 欧姆龙 FINS 内存区（Memory Area）枚举 + *+ * 覆盖常见 PLC（CP/CJ/CS 系列）的读写区域，每个区域包含字（Word）与位（Bit） + * 两种 FINS 内存区代码。地址格式为「区域前缀 + 编号」，如 D6004、C100、W0、H50、A1、E100、T10。 + *
+ * + * @author fengjianguo + * @date 2026/09/28 + */ +@Getter +@AllArgsConstructor +public enum MemoryAreaEnum { + + /** + * D：数据存储器 DM（Data Memory），字 0x82 / 位 0x02 + */ + D("D", "数据存储器DM", "82", "02"), + + /** + * C：CIO 区（Channel I/O），字 0xB0 / 位 0x30 + */ + C("C", "CIO区", "B0", "30"), + + /** + * W：工作继电器 WR（Work Relay），字 0xB1 / 位 0x31 + */ + W("W", "工作继电器WR", "B1", "31"), + + /** + * H：保持继电器 HR（Holding Relay），字 0xB2 / 位 0x32 + */ + H("H", "保持继电器HR", "B2", "32"), + + /** + * A：辅助继电器 AR（Auxiliary Relay），字 0xB3 / 位 0x33 + */ + A("A", "辅助继电器AR", "B3", "33"), + + /** + * E：扩展数据寄存器 EM（Extended Data Memory，当前 Bank），字 0xA0 / 位 0x20 + */ + E("E", "扩展数据寄存器EM", "A0", "20"), + + /** + * T：定时器/计数器 TC（Timer/Counter），当前值 PV 0x89 / 完成标志位 0x09 + *字读写访问 PV 当前值；位读写访问完成标志（置位表示定时/计数完成）。
+ */ + T("T", "定时器/计数器TC", "89", "09"); + + /** + * 地址前缀（不区分大小写） + */ + private final String prefix; + + /** + * 区域名称 + */ + private final String name; + + /** + * 字读写内存区代码 + */ + private final String wordCode; + + /** + * 位读写内存区代码 + */ + private final String bitCode; + + /** + * 按地址前缀解析内存区（不区分大小写） + * + * @param prefix 区域前缀，如 D、d、W、T + * @return 内存区枚举 + * @throws IllegalArgumentException 前缀不支持时抛出 + */ + public static MemoryAreaEnum of(String prefix) { + if (prefix == null || prefix.isEmpty()) { + throw new IllegalArgumentException("区域前缀不能为空"); + } + for (MemoryAreaEnum area : values()) { + if (area.prefix.equalsIgnoreCase(prefix)) { + return area; + } + } + throw new IllegalArgumentException("不支持的FINS内存区前缀: " + prefix + + "（支持: D/C/W/H/A/E/T）"); + } + + /** + * 获取指定读写类型对应的内存区代码 + * + * @param areaType BIT=位代码，WORD=字代码 + * @return FINS 内存区代码 + */ + public String codeOf(AreaTypeEnum areaType) { + return areaType == AreaTypeEnum.BIT ? bitCode : wordCode; + } + +} diff --git a/src/main/java/com/fjg/omorntcpfins/common/enums/MsgEnum.java b/src/main/java/com/fjg/omorntcpfins/common/enums/MsgEnum.java index ff9e021..fae7735 100644 --- a/src/main/java/com/fjg/omorntcpfins/common/enums/MsgEnum.java +++ b/src/main/java/com/fjg/omorntcpfins/common/enums/MsgEnum.java @@ -4,7 +4,8 @@ import lombok.AllArgsConstructor; import lombok.Getter; /** - * @author Zhang GuiHong + * 统一返回结果枚举 + * @author fengjianguo * @date 2021/12/27 10:07 **/ @AllArgsConstructor diff --git a/src/main/java/com/fjg/omorntcpfins/common/queue/HexQueue.java b/src/main/java/com/fjg/omorntcpfins/common/queue/HexQueue.java index c923126..9228d24 100644 --- a/src/main/java/com/fjg/omorntcpfins/common/queue/HexQueue.java +++ b/src/main/java/com/fjg/omorntcpfins/common/queue/HexQueue.java @@ -1,19 +1,34 @@ package com.fjg.omorntcpfins.common.queue; -/** - * @author fengjianguo - * @date 2024/7/30 16:12 - */ import lombok.extern.slf4j.Slf4j; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.stream.IntStream; +/** + * FINS 服务ID(SID)队列 + *+ * FINS 协议中 SID 取值范围 00-FF 共 256 个，用于唯一标识一次请求。 + * 每次发送请求时从队列中取出一个 SID，响应返回或超时后归还， + * 从而支持在同一连接上并发进行多笔读写。 + *
+ * + * @author fengjianguo + * @date 2024/7/30 16:12 + */ @Slf4j public class HexQueue { + private static final ConcurrentLinkedQueue+ * 交互报文、订阅值变化通过 MDC 标记后路由到各自独立的日志文件。 + * 以下名称均可在 application.yml 中手动配置，未配置时使用默认值。 + *
+ * + * @author fengjianguo + * @date 2026/09/28 + */ +@Data +@Configuration +@ConfigurationProperties("omron.log") +public class OmronLogProps { + + /** + * 用于路由专用日志的 MDC 键名（默认值 finsLogCategory） + */ + private String mdcKey = "finsLogCategory"; + + /** + * 交互报文对应的 MDC 取值（默认值 interact） + */ + private String interactValue = "interact"; + + /** + * 订阅值变化对应的 MDC 取值（默认值 change） + */ + private String changeValue = "change"; + + /** + * 交互报文日志文件名前缀（默认值 omron-tcp-fins_interact） + */ + private String interactLogName = "omron-tcp-fins_interact"; + + /** + * 订阅值变化日志文件名前缀（默认值 omron-tcp-fins_change） + */ + private String changeLogName = "omron-tcp-fins_change"; + +} diff --git a/src/main/java/com/fjg/omorntcpfins/config/OmronSubscribeProps.java b/src/main/java/com/fjg/omorntcpfins/config/OmronSubscribeProps.java new file mode 100644 index 0000000..4076836 --- /dev/null +++ b/src/main/java/com/fjg/omorntcpfins/config/OmronSubscribeProps.java @@ -0,0 +1,43 @@ +package com.fjg.omorntcpfins.config; + +import com.fjg.omorntcpfins.model.FinsSubscribePoint; +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.context.annotation.Configuration; + +import java.util.ArrayList; +import java.util.List; + +/** + * 订阅监控配置 + * + * @author fengjianguo + * @date 2026/09/28 + */ +@Data +@Configuration +@ConfigurationProperties("omron.subscribe") +public class OmronSubscribeProps { + + /** + * 订阅监控总开关 + */ + private boolean enabled = true; + + /** + * 默认轮询间隔（毫秒），点位未单独配置时使用 + */ + private long defaultIntervalMs = 1000; + + /** + * 全局默认：首次读取是否触发值变化（默认 true=首读记为一次变化；false=首读仅建立基线）。 + * 优先级低于点位级 firstReadTrigger：点位单独配置时以点位为准，未配置时使用本全局值 + */ + private Boolean firstReadTrigger = true; + + /** + * 配置的订阅点位 + */ + private List+ * 提供三组可配置的线程池： + *
任务积压时采用 AbortPolicy 拒绝并抛异常，便于及时暴露配置过小的问题。
+ */ + @Bean(value = "finsSubscribeScheduler", destroyMethod = "shutdown") + public ScheduledThreadPoolExecutor finsSubscribeScheduler(OmronThreadPoolProps props) { + OmronThreadPoolProps.Subscribe subscribe = props.getSubscribe(); + ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor( + subscribe.getCorePoolSize(), + new ThreadFactoryBuilder().setNamePrefix(subscribe.getThreadNamePrefix()).build(), + new ThreadPoolExecutor.AbortPolicy()); + executor.setKeepAliveTime(subscribe.getKeepAliveSeconds(), TimeUnit.SECONDS); + executor.setRemoveOnCancelPolicy(true); + executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); + executor.setContinueExistingPeriodicTasksAfterShutdownPolicy(false); + log.info("初始化订阅调度线程池：core={}", subscribe.getCorePoolSize()); + return executor; + } + + /** + * 订阅值变化通知分发线程池 + *队列满时采用 CallerRunsPolicy，由调度线程代为执行，避免丢失通知。
+ */ + @Bean(value = "finsNotifyExecutor", destroyMethod = "shutdown") + public ThreadPoolExecutor finsNotifyExecutor(OmronThreadPoolProps props) { + OmronThreadPoolProps.Notify notify = props.getNotify(); + ThreadPoolExecutor executor = new ThreadPoolExecutor( + notify.getCorePoolSize(), + notify.getMaxPoolSize(), + notify.getKeepAliveSeconds(), + TimeUnit.SECONDS, + new ArrayBlockingQueue<>(notify.getQueueCapacity()), + new ThreadFactoryBuilder().setNamePrefix(notify.getThreadNamePrefix()).build(), + new ThreadPoolExecutor.CallerRunsPolicy()); + log.info("初始化通知分发线程池：core={}, max={}, queue={}", + notify.getCorePoolSize(), notify.getMaxPoolSize(), notify.getQueueCapacity()); + return executor; + } + + /** + * 异步读写执行线程池 + *队列满时采用 CallerRunsPolicy，避免任务丢失。
+ */ + @Bean(value = "finsReadExecutor", destroyMethod = "shutdown") + public ThreadPoolExecutor finsReadExecutor(OmronThreadPoolProps props) { + OmronThreadPoolProps.Read read = props.getRead(); + ThreadPoolExecutor executor = new ThreadPoolExecutor( + read.getCorePoolSize(), + read.getMaxPoolSize(), + read.getKeepAliveSeconds(), + TimeUnit.SECONDS, + new ArrayBlockingQueue<>(read.getQueueCapacity()), + new ThreadFactoryBuilder().setNamePrefix(read.getThreadNamePrefix()).build(), + new ThreadPoolExecutor.CallerRunsPolicy()); + log.info("初始化异步读写线程池：core={}, max={}, queue={}", + read.getCorePoolSize(), read.getMaxPoolSize(), read.getQueueCapacity()); + return executor; + } + +} diff --git a/src/main/java/com/fjg/omorntcpfins/config/OmronThreadPoolProps.java b/src/main/java/com/fjg/omorntcpfins/config/OmronThreadPoolProps.java new file mode 100644 index 0000000..e437196 --- /dev/null +++ b/src/main/java/com/fjg/omorntcpfins/config/OmronThreadPoolProps.java @@ -0,0 +1,102 @@ +package com.fjg.omorntcpfins.config; + +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.context.annotation.Configuration; + +/** + * 线程池配置 + *+ * PLC 节点号在 TCP 握手成功后，从握手响应报文解析获取，无需再手动配置。 + * 本机（PC）节点号同时保留配置值，用于组包。 + *
+ * + * @author fengjianguo + * @date 2026/09/28 + */ +@Slf4j +@Component +public class FinsNodeContext { + + /** + * PLC 节点号（握手响应解析所得，未完成握手时为 null） + */ + private volatile Integer plcNode; + + /** + * 本机节点号（握手响应回显，校验用） + */ + private volatile Integer pcNode; + + /** + * 是否已完成握手并成功解析 + */ + private volatile boolean handshakeDone = false; + + /** + * 根据握手响应更新节点信息 + * + * @param pcNodeHex 本机节点（十六进制字符串，如 "18"） + * @param plcNodeHex PLC节点（十六进制字符串，如 "17"） + */ + public void updateFromHandshake(String pcNodeHex, String plcNodeHex) { + try { + this.pcNode = Integer.parseInt(pcNodeHex, 16); + this.plcNode = Integer.parseInt(plcNodeHex, 16); + this.handshakeDone = true; + log.info("握手解析节点成功：本机节点=0x{}（{}），PLC节点=0x{}（{}）", + pcNodeHex, pcNode, plcNodeHex, plcNode); + } catch (NumberFormatException e) { + log.warn("握手节点解析失败：本机节点hex={}，PLC节点hex={}", pcNodeHex, plcNodeHex); + } + } + + /** + * PLC 节点号（握手解析所得，未握手时为 null） + */ + public Integer getPlcNode() { + return plcNode; + } + + /** + * 本机节点号（握手回显，未握手时为 null） + */ + public Integer getPcNode() { + return pcNode; + } + + /** + * 是否已完成握手并成功解析节点 + */ + public boolean isHandshakeDone() { + return handshakeDone; + } + +} diff --git a/src/main/java/com/fjg/omorntcpfins/config/fins/client/FinsClientConfig.java b/src/main/java/com/fjg/omorntcpfins/config/fins/client/FinsClientConfig.java index 992156d..ab841ca 100644 --- a/src/main/java/com/fjg/omorntcpfins/config/fins/client/FinsClientConfig.java +++ b/src/main/java/com/fjg/omorntcpfins/config/fins/client/FinsClientConfig.java @@ -1,94 +1,252 @@ package com.fjg.omorntcpfins.config.fins.client; +import com.fjg.omorntcpfins.config.OmronConfigProps; +import com.fjg.omorntcpfins.common.queue.HexQueue; +import com.fjg.omorntcpfins.config.log.FinsLogContext; +import com.fjg.omorntcpfins.handler.fins.FinsMessageHandler; +import com.fjg.omorntcpfins.model.FinsSubscribePoint; import com.fjg.omorntcpfins.util.HexStringUtil; import io.netty.bootstrap.Bootstrap; -import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.*; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.util.concurrent.DefaultThreadFactory; import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import javax.annotation.PreDestroy; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Consumer; /** - * fins客户端配置 + * FINS TCP 客户端 + *+ * 基于 Netty 实现，提供： + *
与配置文件（yml）注册的订阅点位共存，点位名重复时跳过；断线重连不会重复注册。
+ * + * @param subscribePoints 本次连接指定的订阅点位 + */ + public synchronized void connect(List+ * 用于将带指定 MDC 键值的日志事件路由到独立日志文件： + *
+ * 在打印交互报文、订阅值变化日志前，向 MDC 写入对应的键值， + * logback 中配置的 {@code FinsCategoryFilter} 据此将日志路由到独立文件。 + *
+ * + * @author fengjianguo + * @date 2026/09/28 + */ +@Component +public class FinsLogContext { + + private final OmronLogProps props; + + public FinsLogContext(OmronLogProps props) { + this.props = props; + } + + /** + * 标记当前线程日志为交互报文 + */ + public void putInteract() { + MDC.put(props.getMdcKey(), props.getInteractValue()); + } + + /** + * 标记当前线程日志为订阅值变化 + */ + public void putChange() { + MDC.put(props.getMdcKey(), props.getChangeValue()); + } + + /** + * 清除 MDC 标记（务必在 finally 中调用） + */ + public void clear() { + MDC.remove(props.getMdcKey()); + } + + /** + * 当前使用的 MDC 键名 + */ + public String getMdcKey() { + return props.getMdcKey(); + } + +} diff --git a/src/main/java/com/fjg/omorntcpfins/controller/FinsSubscribeController.java b/src/main/java/com/fjg/omorntcpfins/controller/FinsSubscribeController.java new file mode 100644 index 0000000..9437248 --- /dev/null +++ b/src/main/java/com/fjg/omorntcpfins/controller/FinsSubscribeController.java @@ -0,0 +1,95 @@ +package com.fjg.omorntcpfins.controller; + +import com.fjg.omorntcpfins.model.FinsSubscribePoint; +import com.fjg.omorntcpfins.subscribe.FinsSubscribeManager; +import lombok.extern.slf4j.Slf4j; +import org.springframework.web.bind.annotation.DeleteMapping; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +import java.util.List; +import java.util.Map; + +/** + * 订阅监控管理接口 + * + * @author fengjianguo + * @date 2026/09/28 + */ +@Slf4j +@RestController +@RequestMapping("/subscribe") +public class FinsSubscribeController { + + private final FinsSubscribeManager subscribeManager; + + public FinsSubscribeController(FinsSubscribeManager subscribeManager) { + this.subscribeManager = subscribeManager; + } + + /** + * 查询全部订阅点位（含当前值与运行统计） + */ + @GetMapping("/list") + public List
+ * {
+ * "name": "D6004-温度",
+ * "address": "D6004",
+ * "offset": 1,
+ * "dataType": "WORD",
+ * "intervalMs": 1000,
+ * "enabled": true,
+ * "description": "温度"
+ * }
+ *
+ */
+ @PostMapping
+ public FinsSubscribePoint add(@RequestBody FinsSubscribePoint point) {
+ log.info("新增订阅点位：{}", point);
+ return subscribeManager.addPoint(point);
+ }
+
+ /**
+ * 启动指定点位轮询
+ */
+ @PostMapping("/{name}/start")
+ public FinsSubscribePoint start(@PathVariable String name) {
+ return subscribeManager.startPoint(name);
+ }
+
+ /**
+ * 停止指定点位轮询
+ */
+ @PostMapping("/{name}/stop")
+ public FinsSubscribePoint stop(@PathVariable String name) {
+ return subscribeManager.stopPoint(name);
+ }
+
+ /**
+ * 删除指定点位
+ */
+ @DeleteMapping("/{name}")
+ public String remove(@PathVariable String name) {
+ subscribeManager.removePoint(name);
+ return "success";
+ }
+
+ /**
+ * 订阅监控聚合统计
+ */
+ @GetMapping("/stats")
+ public Map支持全部内存区类型（type: D/C/W/H/A/E/T）与全部读写数据类型。
+ *+ * 所有业务校验、协议解析、读写失败等可控错误统一抛出本异常，携带错误码与提示信息； + * 由调用方（REST 层/上层服务）决定如何呈现。非受检异常，无需强制捕获。 + *
+ * + * @author fengjianguo + * @date 2023/09/05 + */ +@Getter +public class OmronException extends RuntimeException { + + private static final long serialVersionUID = 1L; + + /** + * 错误码：0=成功；-1=通用错误；业务可自定义 + */ + private final Integer code; + + public OmronException(String msg) { + super(msg); + this.code = MsgEnum.ERROR.code(); + } + + public OmronException(String msg, Throwable cause) { + super(msg, cause); + this.code = MsgEnum.ERROR.code(); + } + + public OmronException(Integer code, String msg) { + super(msg); + this.code = code; + } + + public OmronException(Integer code, String msg, Throwable cause) { + super(msg, cause); + this.code = code; + } + + public OmronException(MsgEnum msgEnum) { + super(msgEnum.msg()); + this.code = msgEnum.code(); + } + + public OmronException(MsgEnum msgEnum, Throwable cause) { + super(msgEnum.msg(), cause); + this.code = msgEnum.code(); + } + + /** + * 快速构造：指定错误码与信息 + */ + public static OmronException of(Integer code, String msg) { + return new OmronException(code, msg); + } + + /** + * 快速构造：通用错误码（-1） + */ + public static OmronException of(String msg) { + return new OmronException(msg); + } + + /** + * 快速构造：携带底层原因 + */ + public static OmronException of(Integer code, String msg, Throwable cause) { + return new OmronException(code, msg, cause); + } +} diff --git a/src/main/java/com/fjg/omorntcpfins/handler/fins/FinsMessageHandler.java b/src/main/java/com/fjg/omorntcpfins/handler/fins/FinsMessageHandler.java index 900c9fe..ef23ed0 100644 --- a/src/main/java/com/fjg/omorntcpfins/handler/fins/FinsMessageHandler.java +++ b/src/main/java/com/fjg/omorntcpfins/handler/fins/FinsMessageHandler.java @@ -4,95 +4,143 @@ import com.fjg.omorntcpfins.common.enums.FinsErrorCodeEnum; import com.fjg.omorntcpfins.common.enums.FinsFixedValueEnum; import com.fjg.omorntcpfins.common.enums.FinsReadWriteEnum; import com.fjg.omorntcpfins.common.queue.HexQueue; -import com.fjg.omorntcpfins.config.fins.client.FinsClientConfig; +import com.fjg.omorntcpfins.config.fins.FinsNodeContext; +import com.fjg.omorntcpfins.exception.OmronException; import com.fjg.omorntcpfins.util.FinsMsgAnalysisUtil; -import com.fjg.omorntcpfins.util.HexStringUtil; -import io.netty.buffer.ByteBuf; -import io.netty.buffer.Unpooled; -import io.netty.channel.ChannelHandlerContext; import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; -import java.util.Arrays; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; /** + * FINS 响应消息处理器 + *+ * 按 SID 匹配请求与响应：收到 PLC 响应后，从在途请求表中取出对应的 + * {@link CompletableFuture} 并写入结果（读返回数据、写返回响应标识）。 + * 握手响应到达时，解析其中的节点信息（PLC 节点不再手动配置）。 + *
+ * * @author fengjianguo * @date 2024/7/29 10:04 */ @Slf4j +@Component public class FinsMessageHandler { - private final static Map注意：此处不归还SID，SID队列会在重连建立时整体重置。
+ * + * @param reason 失败原因 + */ + public void failAllPending(String reason) { + if (requestFutures.isEmpty()) { + return; + } + log.warn("FINS连接断开，清理在途请求 {} 个，原因：{}", requestFutures.size(), reason); + requestFutures.forEach((key, future) -> + future.completeExceptionally(new OmronException(reason))); + requestFutures.clear(); } - public static void removeRequestFutures(String key) { - REQUEST_FUTURES.remove(key); - log.info("删除请求：{}" , key); - log.info("归还sid：{}", key); - HexQueue.returnHex(key); + /** + * 当前在途请求数量 + */ + public int pendingCount() { + return requestFutures.size(); } + } diff --git a/src/main/java/com/fjg/omorntcpfins/handler/strategy/area/AreaTypeBitStrategy.java b/src/main/java/com/fjg/omorntcpfins/handler/strategy/area/AreaTypeBitStrategy.java index 5c108a3..9e96c92 100644 --- a/src/main/java/com/fjg/omorntcpfins/handler/strategy/area/AreaTypeBitStrategy.java +++ b/src/main/java/com/fjg/omorntcpfins/handler/strategy/area/AreaTypeBitStrategy.java @@ -1,34 +1,43 @@ package com.fjg.omorntcpfins.handler.strategy.area; +import com.fjg.omorntcpfins.common.enums.MemoryAreaEnum; +import com.fjg.omorntcpfins.exception.OmronException; import com.fjg.omorntcpfins.model.DbInfoPO; import org.springframework.stereotype.Component; /** + * 位读写地址解析策略 + *+ * 支持 D/C/W/H/A/E/T 全部内存区。FINS 位读写地址为 3 字节：高 2 字节为字号、低字节为位号（0~15）， + * 因此 offset 语义为位号（如 D6000 offset=1 表示 D6000.1），读取/写入个数固定为 1。 + *
+ * * @author fengjianguo * @date 2024/7/29 11:24 */ @Component("areaTypeBitStrategy") public class AreaTypeBitStrategy extends BaseAreaTypeStrategy { + + /** + * 位号上限（0~15） + */ + private static final int MAX_BIT_NUMBER = 15; + @Override public DbInfoPO parseDbInfo(String address, int offset) { - DbInfoPO dbInfo = buildDbInfo(address,offset); - String substring = address.substring(0, 1); - String dbArea; - // D位:02,D字:82,W位:31,C位:30,W字:B1,C字:B0 - switch (substring.toUpperCase()) { - case "D": - dbArea = "02"; - break; - case "W": - dbArea = "31"; - break; - case "C": - dbArea = "30"; - break; - default: - dbArea = "00"; + DbInfoPO dbInfo = buildDbInfo(address, offset); + String prefix = address.substring(0, 1); + // 按区域前缀取位内存区代码（D:02, C:30, W:31, H:32, A:33, E:20, T:09） + dbInfo.setDbArea(MemoryAreaEnum.of(prefix).getBitCode()); + if (offset < 0 || offset > MAX_BIT_NUMBER) { + throw new OmronException("位号超出范围(0~15): " + offset); } - dbInfo.setDbArea(dbArea); + // 位地址 = 字号×256 + 位号，并入低字节 + int word = Integer.parseInt(address.substring(1)); + String bitAddress = Integer.toHexString(word * 256 + offset); + dbInfo.setDbBlock(String.format("%6s", bitAddress).replace(" ", "0")); + // 位读写个数固定为 1 + dbInfo.setDbOffset("0001"); return dbInfo; } } diff --git a/src/main/java/com/fjg/omorntcpfins/handler/strategy/area/AreaTypeWordStrategy.java b/src/main/java/com/fjg/omorntcpfins/handler/strategy/area/AreaTypeWordStrategy.java index 0fbf198..6a87395 100644 --- a/src/main/java/com/fjg/omorntcpfins/handler/strategy/area/AreaTypeWordStrategy.java +++ b/src/main/java/com/fjg/omorntcpfins/handler/strategy/area/AreaTypeWordStrategy.java @@ -1,10 +1,13 @@ package com.fjg.omorntcpfins.handler.strategy.area; -import com.fjg.omorntcpfins.common.enums.AreaTypeEnum; +import com.fjg.omorntcpfins.common.enums.MemoryAreaEnum; import com.fjg.omorntcpfins.model.DbInfoPO; import org.springframework.stereotype.Component; /** + * 字读写地址解析策略 + *支持 D/C/W/H/A/E/T 全部内存区，offset 为读取/写入的个数（一次最多 1000 个字）。
+ * * @author fengjianguo * @date 2024/7/29 11:24 */ @@ -12,24 +15,10 @@ import org.springframework.stereotype.Component; public class AreaTypeWordStrategy extends BaseAreaTypeStrategy { @Override public DbInfoPO parseDbInfo(String address, int offset) { - DbInfoPO dbInfo = buildDbInfo(address,offset); - String substring = address.substring(0, 1); - String dbArea; - // D位:02,D字:82,W位:31,C位:30,W字:B1,C字:B0 - switch (substring.toUpperCase()) { - case "D": - dbArea = "82"; - break; - case "W": - dbArea = "B1"; - break; - case "C": - dbArea = "B0"; - break; - default: - dbArea = "00"; - } - dbInfo.setDbArea(dbArea); + DbInfoPO dbInfo = buildDbInfo(address, offset); + String prefix = address.substring(0, 1); + // 按区域前缀取字内存区代码（D:82, C: B0, W:B1, H:B2, A:B3, E:A0, T:89） + dbInfo.setDbArea(MemoryAreaEnum.of(prefix).getWordCode()); return dbInfo; } } diff --git a/src/main/java/com/fjg/omorntcpfins/handler/strategy/area/BaseAreaTypeStrategy.java b/src/main/java/com/fjg/omorntcpfins/handler/strategy/area/BaseAreaTypeStrategy.java index f5280e2..29c2e77 100644 --- a/src/main/java/com/fjg/omorntcpfins/handler/strategy/area/BaseAreaTypeStrategy.java +++ b/src/main/java/com/fjg/omorntcpfins/handler/strategy/area/BaseAreaTypeStrategy.java @@ -1,7 +1,7 @@ package com.fjg.omorntcpfins.handler.strategy.area; +import com.fjg.omorntcpfins.exception.OmronException; import com.fjg.omorntcpfins.model.DbInfoPO; -import com.sun.org.apache.bcel.internal.generic.PUSH; import lombok.extern.slf4j.Slf4j; /** @@ -11,6 +11,16 @@ import lombok.extern.slf4j.Slf4j; @Slf4j public abstract class BaseAreaTypeStrategy implements AreaTypeStrategy { + /** + * 字地址在 FINS 指令中占 3 字节（6 位十六进制），换算规则：字号 × 256 + */ + private static final int WORD_ADDRESS_LENGTH = 6; + + /** + * 字号上限：3 字节地址最大 0xFFFFFF，对应字号最大 0xFFFF + */ + private static final int MAX_WORD_NUMBER = 0xFFFF; + @Override public DbInfoPO parseDbInfo(String address, int offset) { log.info("address:{}, offset:{}", address, offset); @@ -18,12 +28,34 @@ public abstract class BaseAreaTypeStrategy implements AreaTypeStrategy { return null; } + /** + * 构建 FINS 字地址信息 + *起始地址 = 字号 × 256（低字节恒为 0，供位读写时填入位号）， + * 统一补零为 6 位十六进制，避免小地址导致报文长度错误。
+ * + * @param address 地址，如 D6004、W100、T10 + * @param offset 字偏移/读取个数或位号 + * @return 地址信息 + */ public DbInfoPO buildDbInfo(String address, int offset) { + if (address == null || address.length() < 2) { + throw new OmronException("地址格式错误: " + address + "（应为 区域前缀+编号，如 D6004）"); + } + int word; + try { + word = Integer.parseInt(address.substring(1)); + } catch (NumberFormatException e) { + throw new OmronException("地址编号解析失败: " + address); + } + if (word < 0 || word > MAX_WORD_NUMBER) { + throw new OmronException("地址编号超出范围(0~65535): " + address); + } DbInfoPO dbInfo = new DbInfoPO(); - String dbBlock = Integer.toHexString(Integer.parseInt(address.substring(1)) * 256); + String dbBlock = Integer.toHexString(word * 256); + dbInfo.setDbBlock(String.format("%" + WORD_ADDRESS_LENGTH + "s", dbBlock).replace(" ", "0")); String dbOffset = Integer.toHexString(offset); - dbInfo.setDbBlock(dbBlock); - dbInfo.setDbOffset(String.format("%4s",dbOffset).replace(" ", "0")); + dbInfo.setDbOffset(String.format("%4s", dbOffset).replace(" ", "0")); return dbInfo; } + } diff --git a/src/main/java/com/fjg/omorntcpfins/model/FinsSubscribePoint.java b/src/main/java/com/fjg/omorntcpfins/model/FinsSubscribePoint.java new file mode 100644 index 0000000..8c2f0b5 --- /dev/null +++ b/src/main/java/com/fjg/omorntcpfins/model/FinsSubscribePoint.java @@ -0,0 +1,136 @@ +package com.fjg.omorntcpfins.model; + +import com.fjg.omorntcpfins.common.enums.AreaTypeEnum; +import lombok.Data; + +import java.util.concurrent.atomic.AtomicLong; + +/** + * 订阅点位 + *+ * 配置字段（yml 或 REST API 传入）：name / address / offset / dataType / intervalMs / enabled / description； + * 运行字段（由 {@code FinsSubscribeManager} 维护）：当前值、累计读取/成功/变化次数、最近读取时间、错误信息、运行状态。 + *
+ * + * @author fengjianguo + * @date 2026/09/28 + */ +@Data +public class FinsSubscribePoint { + + /** + * 最小轮询间隔（毫秒），防止配置过小打爆PLC + */ + public static final long MIN_INTERVAL_MS = 100L; + + /** + * 点位运行状态 + */ + public enum SubscribeStatus { + /** + * 已注册未启动 + */ + INIT, + /** + * 轮询中 + */ + RUNNING, + /** + * 已停止 + */ + STOPPED + } + + // ==================== 配置字段 ==================== + + /** + * 点位名称（唯一标识） + */ + private String name; + + /** + * PLC 地址，如 D6004、W100、C50 + */ + private String address; + + /** + * 偏移量/一次读取个数，通常为 1 + */ + private Integer offset = 1; + + /** + * 数据类型：BIT / WORD + */ + private AreaTypeEnum dataType = AreaTypeEnum.WORD; + + /** + * 首次读取是否触发值变化：true=首次读到值记为一次变化 null→值；false=首次读取仅建立基线，不触发监听/变化日志/变化计数。 + * 为空时由全局配置 {@code omron.subscribe.first-read-trigger} 兜底（优先级：点位配置 > 全局配置） + */ + private Boolean firstReadTrigger; + + /** + * 轮询间隔（毫秒），为空时使用全局默认值 + */ + private Long intervalMs; + + /** + * 是否启用 + */ + private Boolean enabled = true; + + /** + * 描述 + */ + private String description; + + // ==================== 运行状态字段 ==================== + + /** + * 当前值 + */ + private volatile String currentValue; + + /** + * 最近错误信息 + */ + private volatile String lastError; + + /** + * 累计读取次数 + */ + private AtomicLong readCount = new AtomicLong(0); + + /** + * 累计成功次数 + */ + private AtomicLong successCount = new AtomicLong(0); + + /** + * 累计值变化次数 + */ + private AtomicLong changeCount = new AtomicLong(0); + + /** + * 最近读取时间戳 + */ + private volatile Long lastReadTime; + + /** + * 是否正在读取中（防止同一点位轮询重叠） + */ + private volatile Boolean reading = false; + + /** + * 运行状态 + */ + private volatile SubscribeStatus status = SubscribeStatus.INIT; + + /** + * 生效的轮询间隔（毫秒） + */ + public long effectiveIntervalMs() { + return intervalMs == null || intervalMs < MIN_INTERVAL_MS ? MIN_INTERVAL_MS : intervalMs; + } + +} diff --git a/src/main/java/com/fjg/omorntcpfins/run/AppRunning.java b/src/main/java/com/fjg/omorntcpfins/run/AppRunning.java index 9e5a56b..49a6ae5 100644 --- a/src/main/java/com/fjg/omorntcpfins/run/AppRunning.java +++ b/src/main/java/com/fjg/omorntcpfins/run/AppRunning.java @@ -1,15 +1,14 @@ package com.fjg.omorntcpfins.run; +import com.fjg.omorntcpfins.config.OmronConfigProps; import com.fjg.omorntcpfins.config.fins.client.FinsClientConfig; import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.CommandLineRunner; import org.springframework.stereotype.Component; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; - /** + * 启动后自动连接 PLC + * * @author fengjianguo * @date 2024/7/22 9:28 */ @@ -17,24 +16,18 @@ import java.util.concurrent.Executors; @Component public class AppRunning implements CommandLineRunner { - @Value("${omron.fins-tcp-ip:(127.0.0.1)}") - private String url; + private final FinsClientConfig finsClientConfig; + private final OmronConfigProps omronConfigProps; - @Value("${omron.fins-tcp-port:(9600)}") - private Integer port; + public AppRunning(FinsClientConfig finsClientConfig, OmronConfigProps omronConfigProps) { + this.finsClientConfig = finsClientConfig; + this.omronConfigProps = omronConfigProps; + } @Override - public void run(String... args) throws Exception { - log.info("PLC连接中······"); - FinsClientConfig finsClientConfig = new FinsClientConfig(url, port); - ExecutorService executorService = Executors.newFixedThreadPool(1); - //启动Fins通讯 - executorService.submit(() -> { - try { - finsClientConfig.connect(); - } catch (Exception e) { - throw new RuntimeException(e); - } - }); + public void run(String... args) { + log.info("启动FINS客户端，连接 PLC {}:{} ...", omronConfigProps.getFinsTcpIp(), omronConfigProps.getFinsTcpPort()); + // 非阻塞连接，断线自动重连由客户端内部处理 + finsClientConfig.connect(); } } diff --git a/src/main/java/com/fjg/omorntcpfins/subscribe/FinsSubscribeLogListener.java b/src/main/java/com/fjg/omorntcpfins/subscribe/FinsSubscribeLogListener.java new file mode 100644 index 0000000..9182949 --- /dev/null +++ b/src/main/java/com/fjg/omorntcpfins/subscribe/FinsSubscribeLogListener.java @@ -0,0 +1,30 @@ +package com.fjg.omorntcpfins.subscribe; + +import com.fjg.omorntcpfins.model.FinsSubscribePoint; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; + +/** + * 订阅值变化默认日志监听器 + *值变化已由 {@link FinsSubscribeManager} 统一输出到专用 change 日志文件， + * 本监听器以 DEBUG 级别记录，便于排查时在调试日志中跟踪，避免与专用日志重复刷屏。
+ * + * @author fengjianguo + * @date 2026/09/28 + */ +@Slf4j +@Component +public class FinsSubscribeLogListener implements FinsValueChangeListener { + + @Override + public void onChange(FinsSubscribePoint point, String oldValue, String newValue) { + log.debug("【订阅变化】点位[{}] 地址[{}-{}] {} → {}（累计变化{}次）", + point.getName(), + point.getAddress(), + point.getOffset(), + oldValue, + newValue, + point.getChangeCount()); + } + +} diff --git a/src/main/java/com/fjg/omorntcpfins/subscribe/FinsSubscribeManager.java b/src/main/java/com/fjg/omorntcpfins/subscribe/FinsSubscribeManager.java new file mode 100644 index 0000000..36c1747 --- /dev/null +++ b/src/main/java/com/fjg/omorntcpfins/subscribe/FinsSubscribeManager.java @@ -0,0 +1,462 @@ +package com.fjg.omorntcpfins.subscribe; + +import com.fjg.omorntcpfins.common.enums.AreaTypeEnum; +import com.fjg.omorntcpfins.common.queue.HexQueue; +import com.fjg.omorntcpfins.config.OmronSubscribeProps; +import com.fjg.omorntcpfins.config.fins.client.FinsClientConfig; +import com.fjg.omorntcpfins.config.log.FinsLogContext; +import com.fjg.omorntcpfins.exception.OmronException; +import com.fjg.omorntcpfins.model.FinsSubscribePoint; +import com.fjg.omorntcpfins.util.FinsUtil; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.stereotype.Component; + +import javax.annotation.PostConstruct; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +/** + * 订阅监控管理器 + *+ * 核心能力： + *
连接成功后由客户端回调触发：逐点注册并启动；点位名与已有点位（yml/REST）重复时跳过并告警。
+ * + * @param points 连接时指定的订阅点位 + */ + private void registerPointsFromConnect(List+ * 实现该接口并注册为 Spring Bean（{@code @Component}），即可自动接收所有订阅点位的值变化通知； + * 也可通过 {@code FinsSubscribeManager#addListener} 动态注册。 + *
+ * + * @author fengjianguo + * @date 2026/09/28 + */ +@FunctionalInterface +public interface FinsValueChangeListener { + + /** + * 点位值变化回调 + * + * @param point 点位（含最新值及运行统计） + * @param oldValue 变化前值 + * @param newValue 变化后值 + */ + void onChange(FinsSubscribePoint point, String oldValue, String newValue); + +} diff --git a/src/main/java/com/fjg/omorntcpfins/util/FinsMsgAnalysisUtil.java b/src/main/java/com/fjg/omorntcpfins/util/FinsMsgAnalysisUtil.java index c7db9f6..ea8d592 100644 --- a/src/main/java/com/fjg/omorntcpfins/util/FinsMsgAnalysisUtil.java +++ b/src/main/java/com/fjg/omorntcpfins/util/FinsMsgAnalysisUtil.java @@ -81,6 +81,21 @@ public class FinsMsgAnalysisUtil { */ private static final int DATA_START = 60; + /** + * 握手响应中本机节点位置（如 00000018 中的 18） + */ + private static final int HANDSHAKE_PC_NODE_START = 38; + + /** + * 握手响应中PLC节点位置（如 00000017 中的 17） + */ + private static final int HANDSHAKE_PLC_NODE_START = 46; + + /** + * 握手响应报文最小长度 + */ + private static final int HANDSHAKE_TOTAL_LENGTH = 48; + /** * 步长2 */ @@ -283,4 +298,20 @@ public class FinsMsgAnalysisUtil { return "00000001".equals(finsCommand(msg)); } + /** + * 获取握手响应中的本机节点（十六进制字符串，如 "18"） + */ + public static String finsHandshakePcNode(String msg) { + validateMsgLength(msg, HANDSHAKE_TOTAL_LENGTH); + return extractSubString(msg, HANDSHAKE_PC_NODE_START, STEP_TWO); + } + + /** + * 获取握手响应中的PLC节点（十六进制字符串，如 "17"） + */ + public static String finsHandshakePlcNode(String msg) { + validateMsgLength(msg, HANDSHAKE_TOTAL_LENGTH); + return extractSubString(msg, HANDSHAKE_PLC_NODE_START, STEP_TWO); + } + } diff --git a/src/main/java/com/fjg/omorntcpfins/util/FinsUtil.java b/src/main/java/com/fjg/omorntcpfins/util/FinsUtil.java index af873e4..ac3ed65 100644 --- a/src/main/java/com/fjg/omorntcpfins/util/FinsUtil.java +++ b/src/main/java/com/fjg/omorntcpfins/util/FinsUtil.java @@ -1,24 +1,29 @@ package com.fjg.omorntcpfins.util; import com.fjg.omorntcpfins.common.enums.AreaTypeEnum; +import com.fjg.omorntcpfins.common.enums.MemoryAreaEnum; import com.fjg.omorntcpfins.common.queue.HexQueue; import com.fjg.omorntcpfins.config.OmronConfigProps; +import com.fjg.omorntcpfins.config.fins.FinsNodeContext; import com.fjg.omorntcpfins.config.fins.client.FinsClientConfig; -import com.fjg.omorntcpfins.exception.GlobalException; +import com.fjg.omorntcpfins.exception.OmronException; import com.fjg.omorntcpfins.handler.fins.FinsMessageHandler; import com.fjg.omorntcpfins.handler.strategy.area.AreaTypeStrategy; import com.fjg.omorntcpfins.model.DbInfoPO; import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Component; import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; /** + * FINS 读写工具 + * * @author fengjianguo * @date 2024/7/27 15:57 */ @@ -26,11 +31,12 @@ import java.util.concurrent.TimeoutException; @Component public class FinsUtil { - @Autowired - private OmronConfigProps omronConfigProps; - - @Autowired - private Map如地址 D100 存储 DINT 时：D100=低16位、D101=高16位。
+ * + * @param address 起始地址，如 D100 + * @param offset 起始字号，通常为起始地址（保留参数，读取固定 2 个字） + * @return 32 位无符号整数（-2147483648~4294967295 用 long 表示） + */ + public Long readDword(String address, int offset) { + java.util.ListFINS 位读响应中每个位占 1 字节（2 位十六进制），数据长度 = 个数 × 2。
+ * + * @param address 地址，如 D6000 + * @param bit 起始位号（0~15） + * @param count 读取个数（1~1000） + * @return 位值列表 + */ + public java.util.List值变化已由 {@link FinsSubscribeManager} 统一输出到专用 change 日志文件， - * 本监听器以 DEBUG 级别记录，便于排查时在调试日志中跟踪，避免与专用日志重复刷屏。
+ *打印前打上 MDC 值变化标记，日志路由到控制台与专用 change 日志文件； + * 与 {@link FinsSubscribeManager} 内建的值变化日志保持一致。
* * @author fengjianguo * @date 2026/09/28 @@ -16,15 +17,27 @@ import org.springframework.stereotype.Component; @Component public class FinsSubscribeLogListener implements FinsValueChangeListener { + private final FinsLogContext finsLogContext; + + public FinsSubscribeLogListener(FinsLogContext finsLogContext) { + this.finsLogContext = finsLogContext; + } + @Override public void onChange(FinsSubscribePoint point, String oldValue, String newValue) { - log.debug("【订阅变化】点位[{}] 地址[{}-{}] {} → {}（累计变化{}次）", - point.getName(), - point.getAddress(), - point.getOffset(), - oldValue, - newValue, - point.getChangeCount()); + // 值变化日志打上 MDC 标记，路由到控制台与独立 change 日志文件 + finsLogContext.putChange(); + try { + log.info("【订阅变化】点位[{}] 地址[{}-{}] {} → {}（累计变化{}次）", + point.getName(), + point.getAddress(), + point.getOffset(), + oldValue, + newValue, + point.getChangeCount()); + } finally { + finsLogContext.clear(); + } } } -- Gitee