modbus数据抄读解析转换组包上传

package com.modbus; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; import com.hdx.dataExchange.common.DateUtils; import com.hdx.dataExchange.common.DhGasInfoAdd; import com.hdx.dataExchange.common.DhGasRecordAdd; import com.hdx.dataExchange.config.StaticVariables; import com.hdx.dataExchange.entity.DataUpload; import com.hdx.dataExchange.service.DataUploadService; import com.modbus.config.ModbusMasterConfig; import com.modbus.entity.ModbusFrame; import com.modbus.entity.ModbusFunction; import com.modbus.entity.exception.ConnectionException; import com.modbus.entity.func.request.ReadGasRequest; import com.modbus.eum.ConnectionState; import com.modbus.handler.ModbusChannelInitializer; import com.modbus.handler.ModbusRequestHandler; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.group.ChannelGroup; import io.netty.channel.group.DefaultChannelGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.util.concurrent.DefaultEventExecutorGroup; import io.netty.util.concurrent.EventExecutorGroup; import io.netty.util.concurrent.GenericFutureListener; import io.netty.util.concurrent.GlobalEventExecutor; import lombok.extern.slf4j.Slf4j; import org.quartz.CronExpression; import org.springframework.http.*; import org.springframework.web.client.RestTemplate; import java.beans.PropertyChangeListener; import java.beans.PropertyChangeSupport; import java.text.ParseException; import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @Slf4j public class ModbusServer{ private final ModbusMasterConfig config; private final DataUploadService dataUploadService; private ServerBootstrap bootstrap; private Channel parentChannel; private ConnectionState connectionState = ConnectionState.DOWN; private transient final PropertyChangeSupport propertyChangeSupport = new PropertyChangeSupport(this); public static final String PROP_CONNECTION_STATE = "connectionState"; private final ChannelGroup clientChannels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); public ModbusServer(ModbusMasterConfig config,DataUploadService dataUploadService){ this.config = config; this.dataUploadService = dataUploadService; } public void setup(ModbusRequestHandler handler) throws ConnectionException{ handler.setServer(this); try{ // 1、创建线两个事件循环组 final EventLoopGroup bossGroup = new NioEventLoopGroup(); //处理服务器端接收客户端连接的 final EventLoopGroup workerGroup = new NioEventLoopGroup(); //进行网络通信的（网络读写的） final EventExecutorGroup businessGroup = new DefaultEventExecutorGroup(10); //并行执行线程数 // 2、创建辅助工具类ServerBootstrap，用于服务器通道的一系列配置 bootstrap = new ServerBootstrap(); bootstrap.group(bossGroup,workerGroup) // bossGroup通道 .channel(NioServerSocketChannel.class) // workerGroup通道 3、在这里配置 通信数据的处理逻辑, 可以addLast多个 .childHandler(new ModbusChannelInitializer(handler)) // 客户端连接请求放在队列中等待处理队列大小，注意大于此队列，将拒绝客户端请求 .option(ChannelOption.SO_BACKLOG,128) // 心跳保持，默认是false .childOption(ChannelOption.SO_KEEPALIVE,true); // 4、绑定端口, bind返回future(异步), 加上sync阻塞在获取连接处 parentChannel = bootstrap.bind(config.getPort()).sync().channel(); // 设置监听 setConnectionState(ConnectionState.LISTENING); // 监听关闭事件 parentChannel.closeFuture().addListener(new GenericFutureListener<ChannelFuture>(){ @Override public void operationComplete(ChannelFuture channelFuture) throws Exception{ workerGroup.shutdownGracefully(); bossGroup.shutdownGracefully(); setConnectionState(ConnectionState.DOWN); //设置关闭 } }); }catch(Exception e){ setConnectionState(ConnectionState.DOWN); log.error(e.getMessage(),e); throw new ConnectionException(e.getLocalizedMessage()); } } public void setConnectionState(ConnectionState connectionState){ ConnectionState oldConnectionState = this.connectionState; this.connectionState = connectionState; propertyChangeSupport.firePropertyChange(PROP_CONNECTION_STATE,oldConnectionState,connectionState); } public ConnectionState getConnectionState(){ return connectionState; } public void addPropertyChangeListener(PropertyChangeListener listener){ propertyChangeSupport.addPropertyChangeListener(listener); } public void removePropertyChangeListener(PropertyChangeListener listener){ propertyChangeSupport.removePropertyChangeListener(listener); } public void close(){ if(parentChannel!=null){ parentChannel.close().awaitUninterruptibly(); } clientChannels.close().awaitUninterruptibly(); shutdown(); } public void addClient(Channel channel){ clientChannels.add(channel); setConnectionState(ConnectionState.CLIENTS_CONNECTED); } public void removeClient(Channel channel){ clientChannels.remove(channel); setConnectionState(ConnectionState.CLIENTS_CONNECTED); } public ChannelGroup getClientChannels(){ return clientChannels; } private static final long RESPONSE_TIMEOUT = 10000; //一个命令的响应在10秒外超时 private final ConcurrentHashMap<String,PendingRequest> pendingRequests = new ConcurrentHashMap<>(); //存储待处理的请求 private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); //定时任务调度器 private long firstRequestTimestamp = 0; private boolean allRequestsMatched = false; private Map<String,List<Map<ModbusFrame,ModbusFrame>>> matchedFrames = new HashMap<>(); private final ConcurrentHashMap<String,Long> restCallCounterMap = new ConcurrentHashMap<>(); // 发送请求帧的方法 public void sendFrame(Channel channel,ModbusFrame frame,String remoteIp){ if(firstRequestTimestamp==0){ firstRequestTimestamp = System.currentTimeMillis(); // 记录第一个请求的发送时间戳 } long timestamp = System.currentTimeMillis(); String key = generateRequestKey(frame,remoteIp); //生成请求的唯一键 PendingRequest pendingRequest = new PendingRequest(frame,timestamp); pendingRequests.put(key,pendingRequest); scheduleTimeoutRemoval(key,timestamp); //启动一个定时任务来检查并移除超时的请求 channel.writeAndFlush(frame); //发送请求帧（不附加任何标识符） log.info("发送："+frame); } private String generateRequestKey(ModbusFrame frame,String ipAddr){ //生成请求的唯一键，基于地址和功能码 return frame.getHeader().getDeviceAddr()+"_"+frame.getHeader().getFunctionCode()+"_"+ipAddr; //根据实际的Modbus帧格式生成键 } // 安排超时移除请求 private void scheduleTimeoutRemoval(String key,long timestamp){ scheduler.schedule(()->{ pendingRequests.remove(key); //移除超时的请求 },RESPONSE_TIMEOUT,TimeUnit.MILLISECONDS); } // 处理响应帧的方法 public void handleResponse(ModbusFrame responseFrame,String deviceIp){ String key = generateRequestKey(responseFrame,deviceIp); //生成响应的唯一键 PendingRequest pendingRequest = pendingRequests.get(key); if(pendingRequest!=null){ long sendTimestamp = pendingRequest.getTimestamp(); long receiveTimestamp = System.currentTimeMillis(); if(receiveTimestamp-sendTimestamp<=RESPONSE_TIMEOUT){ processResponse(deviceIp,pendingRequest.getFrame(),responseFrame); //响应在超