1、接入永阳4G电表

user-service/src/main/java/com/mh/user/config/ParamConfig.java

@@ -0,0 +1,48 @@
+package com.mh.user.config;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author LJF
+ * @version 1.0
+ * @project TAD_Server
+ * @description 参数配置管理
+ * @date 2023/7/4 11:46:48
+ */
+@Configuration
+public class ParamConfig {
+
+    private String meterPass;
+
+    private String meterOpcode;
+
+    private int port;
+    public String getMeterPass() {
+        return meterPass;
+    }
+
+    @Value("${device.meter.password}")
+    public void setMeterPass(String meterPass) {
+        this.meterPass = meterPass;
+    }
+
+    public String getMeterOpcode() {
+        return meterOpcode;
+    }
+
+    @Value("${device.meter.opcode}")
+    public void setMeterOpcode(String meterOpcode) {
+        this.meterOpcode = meterOpcode;
+    }
+
+    public int getPort() {
+        return port;
+    }
+
+    @Value("${device.port}")
+    public void setPort(int port) {
+        this.port = port;
+    }
+
+}

user-service/src/main/java/com/mh/user/constants/FourthGEnum.java

@@ -0,0 +1,111 @@
+package com.mh.user.constants;
+
+/**
+ * @author LJF
+ * @version 1.0
+ * @project TAD_Server
+ * @description 4G设备常量值
+ * @date 2023/7/3 18:14:12
+ */
+public enum FourthGEnum {
+
+    LOGIN_HEART_RESPONSE("0D", "登录和心跳确认控制码"),
+
+    LOGIN_HEART_REQUEST("8D", "登录和心跳请求控制码"),
+
+    LOGIN_HEART_EXCEPTION("CD", "登录和心跳异常应答控制码"),
+
+    LOGIN_TYPE("02", "登录类型"),
+
+    HEART_TYPE("01", "心跳类型"),
+
+    CLOSE_CONTROL_TYPE("1C", "合闸指令"),
+
+    OPEN_CONTROL_TYPE("1A", "跳闸指令"),
+
+    DATA_ACTIVE_UPLOAD("8E", "数据主动上报控制码"),
+
+    PULL_AND_CLOSE_DEVICE_REQUEST("1C", "跳合闸请求控制码"),
+
+    PULL_AND_CLOSE_DEVICE_RESPONSE("9C", "跳合闸应答控制码"),
+
+    PULL_AND_CLOSE_DEVICE_EXCEPTION("DC", "跳合闸应答异常控制码"),
+
+    CLEAR_DEVICE_REQUEST("1A", "电表清零请求控制码"),
+
+    CLEAR_DEVICE_RESPONSE("9A", "电表清零应答控制码"),
+
+    CLEAR_DEVICE_EXCEPTION("DA", "电表清零应答异常控制码"),
+
+    READ_DEVICE_REQUEST("11", "读电表请求控制码"),
+
+    READ_DEVICE_RESPONSE("91", "读电表应答控制码"),
+
+    READ_DEVICE_EXCEPTION("D1", "读电表应答异常控制码"),
+
+    WRITE_DEVICE_REQUEST("14", "写电表请求控制码"),
+
+    WRITE_DEVICE_RESPONSE("94", "读电表应答控制码"),
+    WRITE_DEVICE_IP("94", "写电表IP应答控制码"),
+
+    WRITE_DEVICE_EXCEPTION("D4", "读电表应答异常控制码"),
+
+    CHARGE_DEVICE_REQUEST("0F", "充值指令"),
+
+    CHARGE_DEVICE_RESPONSE("8F", "充值正确应答控制码"),
+
+    CHARGE_DEVICE_EXCEPTION("CF", "充值异常应答控制码"),
+
+    READ_TOTAL_CHARGE_IDENTIFY_CODE("00000000", "当前组合有功总电能"),
+
+    READ_STATUS_IDENTIFY_CODE("03050004", "读电表运行状态"),
+    READ_UPLOAD_CODE("02008104", "读取上报报文"),
+
+    TY_UPLOAD_CODE("55C2", "腾越水表上报标识"),
+
+    WRITE_CHARGE_TIME_IDENTIFY_CODE("07008104", "写入充值次数"),
+
+    WRITE_LIMIT_POWER("10008104","写入限容功率"),
+    READ_LIMIT_POWER("10008104","读取限容功率"),
+    READ_SOFT_VERSION("01008004","读软件版本"),
+    READ_HARD_VERSION("02008004","读硬件版本"),
+    READ_ICCID("05008104","读取ICCID"),
+    READ_IMEI("04008104","读取IMEI"),
+    WRITE_IP("0B008104","设置IP端口号"),
+    READ_IP("0B008104","读取IP端口号"),
+    UPGRADES("13008104","触发升级，写1触发升级")
+    ;
+
+    private String code;
+
+    private String desc;
+
+    FourthGEnum(String code, String desc) {
+        this.code = code;
+        this.desc = desc;
+    }
+
+    public String getCode() {
+        return code;
+    }
+
+    public void setCode(String code) {
+        this.code = code;
+    }
+
+    public String getDesc() {
+        return desc;
+    }
+
+    public void setDesc(String desc) {
+        this.desc = desc;
+    }
+
+    @Override
+    public String toString() {
+        return "FourthGEnum{" +
+                "code='" + code + '\'' +
+                ", desc='" + desc + '\'' +
+                '}';
+    }
+}

user-service/src/main/java/com/mh/user/constants/ProtocolEnum.java

@@ -1,9 +1,6 @@
 package com.mh.user.constants;
 
-import com.mh.user.factory.CJ188Protocol;
-import com.mh.user.factory.EleProtocol;
-import com.mh.user.factory.ModbusProtocol;
-import com.mh.user.factory.Protocol;
+import com.mh.user.factory.*;
 
 /**
  * @author LJF
@@ -22,6 +19,8 @@ public enum ProtocolEnum {
 
     ELE_07_PROTOCOL("3", EleProtocol.getInstance()),
 
+    ELE_YY_PROTOCOL("4", EleYYProtocol.getInstance()),
+
     ;
 
     private String protocolType;

user-service/src/main/java/com/mh/user/constants/ProtocolStrategyEnum.java

@@ -1,9 +1,6 @@
 package com.mh.user.constants;
 
-import com.mh.user.strategy.CJ188ProtocolStrategy;
-import com.mh.user.strategy.EleProtocolStrategy;
-import com.mh.user.strategy.ModbusProtocolStrategy;
-import com.mh.user.strategy.ProtocolStrategy;
+import com.mh.user.strategy.*;
 
 /**
  * @author LJF
@@ -22,6 +19,8 @@ public enum ProtocolStrategyEnum {
 
     ELE_07_PROTOCOL("3", EleProtocolStrategy.getInstance()),
 
+    ELE_YY_FOUR_G_PROTOCOL("4", EleYYProtocolStrategy.getInstance()),
+
     ;
 
     private String protocolType;

user-service/src/main/java/com/mh/user/factory/EleYYProtocol.java

@@ -0,0 +1,49 @@
+package com.mh.user.factory;
+
+import com.mh.user.entity.DeviceCodeParamEntity;
+import com.mh.user.entity.MeterManageEntity;
+import com.mh.user.strategy.ProtocolStrategy;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * @author LJF
+ * @version 1.0
+ * @project mh_esi
+ * @description 水表标准协议
+ * @date 2024-09-12 14:04:01
+ */
+@Slf4j
+public class EleYYProtocol implements Protocol {
+
+    private ProtocolStrategy eleProtocolStrategy;
+
+    private static class SingletonHolder{
+        private static final EleYYProtocol INSTANCE = new EleYYProtocol();
+    }
+
+    private EleYYProtocol(){
+        // 防止外部直接实例化
+    }
+
+    public static EleYYProtocol getInstance(){
+        return SingletonHolder.INSTANCE;
+    }
+
+    @Override
+    public void setStrategy(ProtocolStrategy protocolStrategy) {
+        this.eleProtocolStrategy = protocolStrategy;
+    }
+
+    @Override
+    public String createOrder(MeterManageEntity meterManageEntity) {
+        log.info("永阳电表97/07规约协议：工厂创建报文");
+        return eleProtocolStrategy.createOrder(meterManageEntity);
+    }
+
+    @Override
+    public String analysisReceiveData(DeviceCodeParamEntity deviceCodeParamEntity, String receiveData) {
+        log.info("永阳电表97/07规约协议：工厂解析报文");
+        return eleProtocolStrategy.analysisReceiveData(deviceCodeParamEntity, receiveData);
+    }
+
+}

user-service/src/main/java/com/mh/user/job/CollectionLoopRunner.java

@@ -1,6 +1,7 @@
 package com.mh.user.job;
 
 import com.mh.user.netty.EchoServer;
+import com.mh.user.netty.fourg.FourGNettyEchoServer;
 import com.mh.user.service.*;
 import gnu.io.SerialPort;
 import lombok.extern.slf4j.Slf4j;
@@ -36,6 +37,7 @@ public class CollectionLoopRunner implements ApplicationRunner {
             // 生成字典缓存
             createDictDataCache();
             new EchoServer(6001).start();                // 调用服务器的start方法,网关作为客户端,6006新珠江,6001广合科技,6077广大科技 6028(6006)珠江国际,5000保利山庄,6004东莞迎宾馆
+            new FourGNettyEchoServer().bind(6060);       // 4G电表数据上线
         } catch (Exception e) {
             log.info("6001端口已占用！");
         }

user-service/src/main/java/com/mh/user/mapper/DeviceCodeParamMapper.java

@@ -121,4 +121,8 @@ public interface DeviceCodeParamMapper extends BaseMapper<DeviceCodeParamEntity>
 
     @Delete("delete from device_code_param where project_id=#{projectId} ")
     void deleteDeviceCodeParam(@Param("projectId") long projectId);
+
+    @ResultMap("rs")
+    @Select("select top 1 * from  device_code_param  where device_addr=#{deviceCode} ")
+    DeviceCodeParamEntity queryDeviceCodeParamByDeviceCode(String deviceCode);
 }

user-service/src/main/java/com/mh/user/netty/fourg/FourGNettyEchoServer.java

@@ -0,0 +1,55 @@
+package com.mh.user.netty.fourg;
+
+import com.mh.user.netty.fourg.handle.FourthChannelInitializer;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.logging.LogLevel;
+import io.netty.handler.logging.LoggingHandler;
+import lombok.extern.slf4j.Slf4j;
+
+
+/**
+ * @author LJF
+ * @title ：Netty
+ * @description ：netty 使用
+ * @updateTime 2020-04-21
+ * @throws ：
+ */
+@Slf4j
+public class FourGNettyEchoServer {
+
+    public void bind(int port) throws Exception {
+        // accept线程组，用来接收连接
+        EventLoopGroup bossGroup = new NioEventLoopGroup(2);
+        // IO 线程组，用来处理业务逻辑
+        EventLoopGroup workerGroup = new NioEventLoopGroup(4);
+
+        try {
+            // 服务端启动引导
+            ServerBootstrap serverBootstrap = new ServerBootstrap();
+            serverBootstrap.group(bossGroup,workerGroup)                // 绑定两个线程
+                    .channel(NioServerSocketChannel.class)              // 指定通道类型
+                    .option(ChannelOption.SO_BACKLOG, 1024)       // 设置TCP连接的缓冲区
+                    .option(ChannelOption.SO_KEEPALIVE, true)
+                    .handler(new LoggingHandler(LogLevel.INFO))         // 设置日志级别
+                    .childHandler(new FourthChannelInitializer());      // 初始化handle
+            // 通过bind启动服务
+            ChannelFuture f = serverBootstrap.bind(port).sync();
+            // 阻塞主线程，知道网络服务被关闭
+            //f.channel().closeFuture().sync();
+            f.channel().closeFuture().addListener(future -> {
+                log.info("Netty服务器已关闭");
+                workerGroup.shutdownGracefully();
+                bossGroup.shutdownGracefully();
+            });
+        } catch (Exception e){
+            log.error("netty服务器异常", e);
+            workerGroup.shutdownGracefully();
+            bossGroup.shutdownGracefully();
+        }
+    }
+}

user-service/src/main/java/com/mh/user/netty/fourg/decoder/CustomFrameEncoder.java

@@ -0,0 +1,63 @@
+package com.mh.user.netty.fourg.decoder;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToByteEncoder;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class CustomFrameEncoder extends MessageToByteEncoder<String> {
+    private static final byte FRAME_START = 0x68;
+    private static final byte FRAME_END = 0x16;
+
+    @Override
+    protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) throws Exception {
+        log.info("数据到达:"+msg);
+        // Write first frame start
+        out.writeByte(FRAME_START);
+
+        // Write meter number
+        byte[] meterNumberBytes = hexStringToByteArray(msg.substring(2, 14)); // Extract meter number from message
+        out.writeBytes(meterNumberBytes);
+
+        // Write second frame start
+        out.writeByte(FRAME_START);
+
+        // Write control code
+        byte controlCode = hexStringToByteArray(msg.substring(14, 16))[0]; // Extract control code from message
+        out.writeByte(controlCode);
+
+        // Write data length
+        int dataLength = Integer.parseInt(msg.substring(16, 18), 16); // Extract data length from message
+        out.writeByte(dataLength);
+
+        // Write data
+        byte[] dataBytes = hexStringToByteArray(msg.substring(18, 18 + (dataLength * 2))); // Extract data from message
+        out.writeBytes(dataBytes);
+
+        // Calculate and write checksum
+        byte checksum = calculateChecksum(msg.substring(14, 18 + (dataLength * 2))); // Calculate checksum
+        out.writeByte(checksum);
+
+        // Write frame end
+        out.writeByte(FRAME_END);
+    }
+
+    private byte[] hexStringToByteArray(String hexString) {
+        int len = hexString.length();
+        byte[] data = new byte[len / 2];
+        for (int i = 0; i < len; i += 2) {
+            data[i / 2] = (byte) ((Character.digit(hexString.charAt(i), 16) << 4)
+                    + Character.digit(hexString.charAt(i + 1), 16));
+        }
+        return data;
+    }
+
+    private byte calculateChecksum(String message) {
+        byte checksum = 0;
+        for (int i = 0; i < message.length(); i += 2) {
+            checksum += Integer.parseInt(message.substring(i, i + 2), 16);
+        }
+        return (byte) (checksum & 0xFF);
+    }
+}

user-service/src/main/java/com/mh/user/netty/fourg/handle/DataUploadServerHandler.java

@@ -0,0 +1,53 @@
+package com.mh.user.netty.fourg.handle;
+
+import com.mh.user.constants.FourthGEnum;
+import com.mh.user.factory.Protocol;
+import com.mh.user.factory.ProtocolFactory;
+import com.mh.user.netty.fourg.session.ServerSession;
+import com.mh.user.netty.fourg.session.SessionMap;
+import com.mh.user.strategy.ProtocolStrategy;
+import com.mh.user.strategy.ProtocolStrategyFactory;
+import com.mh.user.utils.ExchangeStringUtil;
+import com.mh.user.utils.MeterUtils;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class DataUploadServerHandler extends ChannelInboundHandlerAdapter {
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        //将接收到的数据转为字符串，此字符串就是客户端发送的字符串
+        String receiveStr = ExchangeStringUtil.bytesToHexString(msg);
+        // 判断当前报文是否是上报数据报文
+        if (receiveStr != null && receiveStr.endsWith("16")) {
+            if (MeterUtils.makeChecksum(receiveStr)) {
+                // 上报报文不用回复
+                if (!FourthGEnum.DATA_ACTIVE_UPLOAD.getCode().equalsIgnoreCase(MeterUtils.getControlCode(receiveStr))) {
+                    super.channelRead(ctx, msg);
+                    return;
+                }
+                // 判断是否登录，没有登录立马断开
+                String deviceCode = MeterUtils.getDeviceCode(receiveStr);
+                if (!SessionMap.inst().hasLogin(deviceCode+ctx.channel().remoteAddress())) {
+                    ServerSession.closeSession(ctx);
+                    return;
+                }
+                log.info("接收到的上报报文，并推送到rabbitmq <== {} " , receiveStr);
+                // 创建设备报文
+                Protocol protocol = ProtocolFactory.matchProtocol("4");
+                ProtocolStrategy strategy = ProtocolStrategyFactory.matchProtocolStrategy("4");
+
+                if (strategy == null) {
+                    return;
+                }
+
+                protocol.setStrategy(strategy);
+                protocol.analysisReceiveData(null, receiveStr);
+            }
+        }
+        super.channelRead(ctx, msg);
+    }
+
+}

user-service/src/main/java/com/mh/user/netty/fourg/handle/ExceptionServerHandler.java

@@ -0,0 +1,50 @@
+package com.mh.user.netty.fourg.handle;
+
+import com.mh.user.netty.fourg.session.ServerSession;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * @author LJF
+ * @version 1.0
+ * @project TAD_Server
+ * @description 异常处理
+ * @date 2023/7/5 09:02:03
+ */
+@Slf4j
+public class ExceptionServerHandler extends ChannelInboundHandlerAdapter {
+    //private final ApplicationContext applicationContext = ApplicationContextProvider.getApplicationContext();
+    //private final MeterInfoService meterInfoService = applicationContext.getBean(MeterInfoService.class);
+
+    @Override
+    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+        ctx.flush();
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        log.info(ctx.channel().remoteAddress()+"客户端断开连接！");
+        //String meterNumIp = ServerSession.closeSession(ctx);
+        //String[] split = meterNumIp.split("/");
+        //String meterNum = split[0];
+        //if (meterNum != null) {
+        //    meterInfoService.insertTestDetail(meterNum,"offline");
+        //}
+        ServerSession.closeSession(ctx);
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        if (cause instanceof UnsupportedOperationException) {
+            ServerSession.closeSession(ctx);
+        } else {
+            log.info("获得已知异常！",cause);
+            ctx.close();
+        }
+        log.info("获得已知异常！{}", cause.getMessage());
+        log.info("获得已知异常！", cause);
+        ServerSession.closeSession(ctx);
+        ctx.close();
+    }
+}

user-service/src/main/java/com/mh/user/netty/fourg/handle/FourthChannelInitializer.java

@@ -0,0 +1,34 @@
+package com.mh.user.netty.fourg.handle;
+
+import com.mh.user.netty.fourg.decoder.CustomFrameEncoder;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelInitializer;
+
+/**
+ * @author LJF
+ * @version 1.0
+ * @project TAD_Server
+ * @description 初始换一个服务端
+ * @date 2023/7/3 15:58:05
+ */
+public class FourthChannelInitializer extends ChannelInitializer<Channel> {
+    @Override
+    protected void initChannel(Channel channel) throws Exception {
+        // 电表报文分隔符，一般都是16结尾
+        //ByteBuf delimiter = Unpooled.copiedBuffer(new byte[]{22});
+        //打印分隔符
+        //channel.pipeline().addLast("frameDecoder", new DelimiterBasedFrameDecoder(1024,false, delimiter));
+        //分割报文处理解码器，防止数据粘包
+        channel.pipeline().addLast(new CustomFrameEncoder());
+        // 处理登录
+        channel.pipeline().addLast(new LoginRequestHandler());
+        // 处理心跳
+        channel.pipeline().addLast(new HeartBeatServerHandler());
+        // 处理数据上报数据
+        channel.pipeline().addLast(new DataUploadServerHandler());
+        // 异常处理
+        channel.pipeline().addLast(new ExceptionServerHandler());
+    }
+
+
+}

user-service/src/main/java/com/mh/user/netty/fourg/handle/HeartBeatServerHandler.java

@@ -0,0 +1,144 @@
+package com.mh.user.netty.fourg.handle;
+
+import com.mh.user.constants.FourthGEnum;
+import com.mh.user.netty.fourg.session.ServerSession;
+import com.mh.user.netty.fourg.task.FutureTaskScheduler;
+import com.mh.user.utils.ExchangeStringUtil;
+import com.mh.user.utils.MeterUtils;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.timeout.IdleStateEvent;
+import io.netty.handler.timeout.IdleStateHandler;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.concurrent.TimeUnit;
+
+
+/**
+ * @author LJF
+ * @version 1.0
+ * @project TAD_Server
+ * @description 心跳包设置
+ * @date 2023/7/4 13:54:27
+ */
+@Slf4j
+public class HeartBeatServerHandler extends IdleStateHandler {
+
+    //目前电表默认心跳60秒，多长时间没有心跳，就关闭连接
+    private static final int READ_IDLE_GAP = 60*5;
+
+    /**
+     *
+     * @param readerIdleTimeSeconds 最长 没有 read到心跳的时间
+     * @param writerIdleTimeSeconds 心跳包我们只是进行接收心跳，没有所谓的写操作，我们将其设置为0
+     * @param allIdleTimeSeconds 如果我们定义 多长时间没有 读或写操作，我们使用这个参数。
+     */
+    public HeartBeatServerHandler(int readerIdleTimeSeconds, int writerIdleTimeSeconds, int allIdleTimeSeconds) {
+        super(readerIdleTimeSeconds, writerIdleTimeSeconds, allIdleTimeSeconds);
+    }
+
+    public HeartBeatServerHandler(long readerIdleTime, long writerIdleTime, long allIdleTime, TimeUnit unit) {
+        super(readerIdleTime, writerIdleTime, allIdleTime, unit);
+    }
+
+    public HeartBeatServerHandler() {
+        super(READ_IDLE_GAP, 0, 0, TimeUnit.SECONDS);
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+
+        //将接收到的数据转为字符串，此字符串就是客户端发送的字符串
+        String receiveStr = ExchangeStringUtil.bytesToHexString(msg);
+        // 判断当前报文是否是登录报文
+        if (receiveStr != null && receiveStr.endsWith("16")) {
+            if (MeterUtils.makeChecksum(receiveStr)) {
+                //todo 压测放开，正式屏蔽
+                //if (!FourthGEnum.LOGIN_HEART_REQUEST.getCode().equalsIgnoreCase(MeterUtils.getControlCode(receiveStr))
+                //        || !FourthGEnum.LOGIN_TYPE.getCode().equalsIgnoreCase(MeterUtils.getControlType(receiveStr))) {
+                //    super.channelRead(ctx, msg);
+                //    return;
+                //}
+                //log.info("接收到的心跳报文 <== {}",receiveStr);
+
+                // TODO 压力测试时候屏蔽，不做压力测试放开
+                if (!FourthGEnum.LOGIN_HEART_REQUEST.getCode().equalsIgnoreCase(MeterUtils.getControlCode(receiveStr))
+                        || !FourthGEnum.HEART_TYPE.getCode().equalsIgnoreCase(MeterUtils.getControlType(receiveStr))) {
+                    super.channelRead(ctx, msg);
+                    return;
+                }
+                String meterNum = ExchangeStringUtil.changePosition(receiveStr.substring(2, 14));
+                String signalIntensity = ExchangeStringUtil.cutThree(receiveStr.substring(22, 24));
+                //log.info(ctx.channel().remoteAddress()+"地址"+"接收到的心跳报文 <== {}，表号 <=={}, 信号强度：{}", receiveStr, meterNum, signalIntensity);
+//                MeterInfoEntity meterInfo = new MeterInfoEntity();
+//                meterInfo.setMeterNum(meterNum);
+//                meterInfo.setSignal(String.valueOf(Integer.parseInt(signalIntensity,16)));
+//                meterInfo.setOnlineState(0);
+//                meterInfoService.updateMeterData(meterInfo);
+
+                // 获取表号
+                String deviceCode = MeterUtils.getDeviceCode(receiveStr);
+                //判断是否存在需要修改IP或者升级的配置 0未处理1处理中2成功   暂时屏蔽
+                //List<DeviceSettingDto> list = meterInfoService.queryDeviceSetting(deviceCode);
+                //MHThreadPoolExecutor.getInstance().execute(() -> {
+                //    if (list == null) {
+                //        return;
+                //    }
+                //    for (DeviceSettingDto dto : list) {
+                //        boolean checked = false;
+                //        switch (dto.getControlType()) {
+                //            case "update":
+                //                //修改IP接口
+                //                String cmd = MeterUtils.createControlCode(deviceCode, "13",
+                //                        null, null, null, null, dto.getTargetIp());
+                //                ctx.writeAndFlush(MeterUtils.getByteBuf(ctx, cmd));
+                //                log.info("表号:{},发送修改IP指令 ==> {}",deviceCode, cmd);
+                //                meterInfoService.updateDeviceSetting(dto.getId(),1);
+                //                checked = NettyTools.waitReceiveMsg(deviceCode);
+                //                if (checked){
+                //                    meterInfoService.updateDeviceSetting(dto.getId(),2);
+                //                }
+                //                break;
+                //            case "upGrade":
+                //                //升级
+                //                String upGrade = MeterUtils.createControlCode(deviceCode, "15",
+                //                        null, null, null, null, null);
+                //                ctx.writeAndFlush(MeterUtils.getByteBuf(ctx, upGrade));
+                //                log.info("表号:{},发送固件升级指令 ==> {}",deviceCode, upGrade);
+                //                meterInfoService.updateDeviceSetting(dto.getId(),1);
+                //                checked = NettyTools.waitReceiveMsg(deviceCode);
+                //                if (checked){
+                //                    meterInfoService.updateDeviceSetting(dto.getId(),2);
+                //                }
+                //                break;
+                //            default:
+                //                break;
+                //        }
+                //    }
+                //});
+
+                //进行心跳包的处理
+                FutureTaskScheduler.add(() -> {
+                    if(ctx.channel().isActive()) {
+                        //直接将信息 回复给客户端
+                        String responseStr = MeterUtils.createLoginOrHeartCheckCode(deviceCode, FourthGEnum.HEART_TYPE.getCode());
+                        ctx.writeAndFlush(MeterUtils.getByteBuf(ctx, responseStr));
+                        //log.info("发送心跳确认报文 ==> {}", responseStr);
+                    }
+                });
+            }
+        }
+        super.channelRead(ctx, msg);
+    }
+
+    @Override
+    protected void channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt) throws Exception {
+        log.info(ctx.channel().remoteAddress() + " 已超过 "+READ_IDLE_GAP + " 秒没有接收到心跳包，已断开链接");
+        String meterNumIp = ServerSession.closeSession(ctx);
+        String[] split = meterNumIp.split("/");
+        String meterNum = split[0];
+        if (meterNum != null) {
+        }
+        ctx.close();
+    }
+
+}

user-service/src/main/java/com/mh/user/netty/fourg/handle/LoginRequestHandler.java

@@ -0,0 +1,112 @@
+package com.mh.user.netty.fourg.handle;
+
+import com.mh.user.constants.FourthGEnum;
+import com.mh.user.netty.fourg.session.ServerSession;
+import com.mh.user.netty.fourg.session.SessionMap;
+import com.mh.user.netty.fourg.task.CallbackTask;
+import com.mh.user.netty.fourg.task.CallbackTaskScheduler;
+import com.mh.user.utils.ExchangeStringUtil;
+import com.mh.user.utils.MeterUtils;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * @author LJF
+ * @version 1.0
+ * @project TAD_Server
+ * @description 登录报文处理
+ * @date 2023/7/3 15:34:11
+ */
+@Slf4j
+public class LoginRequestHandler extends ChannelInboundHandlerAdapter {
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        //将接收到的数据转为字符串，此字符串就是客户端发送的字符串
+        String receiveStr = ExchangeStringUtil.bytesToHexString(msg);
+        // 判断当前报文是否是登录报文
+        if (receiveStr != null && receiveStr.endsWith("16")) {
+            if (MeterUtils.makeChecksum(receiveStr)) {
+                if (!FourthGEnum.LOGIN_HEART_REQUEST.getCode().equalsIgnoreCase(MeterUtils.getControlCode(receiveStr))
+                        || !FourthGEnum.LOGIN_TYPE.getCode().equalsIgnoreCase(MeterUtils.getControlType(receiveStr))) {
+                    super.channelRead(ctx, msg);
+                    return;
+                }
+                log.info("接收到的登录报文 <== {}", receiveStr);
+                // 获取表号
+                String deviceCode = MeterUtils.getDeviceCode(receiveStr);
+                String meterNum = deviceCode;
+                deviceCode = deviceCode + ctx.channel().remoteAddress();
+                //新的session的创建
+                ServerSession session = new ServerSession(ctx.channel(), deviceCode);
+
+                //进行登录逻辑处理，异步进行处理。并且需要知道 处理的结果。 callbacktask就要
+                //派上用场了
+                String finalDeviceCode = deviceCode;
+                CallbackTaskScheduler.add(new CallbackTask<Boolean>() {
+                    @Override
+                    public Boolean execute() throws Exception {
+                        //进行 login 逻辑的处理
+                        return action(session, finalDeviceCode, ctx, meterNum);
+                    }
+                    //没有异常的话，我们进行处理
+                    @Override
+                    public void onBack(Boolean result) {
+                        if(result) {
+                            log.info("设备登录成功： 设备号 = " + session.getSessionId());
+                            //ctx.pipeline().remove(LoginRequestHandler.class); //压测需要放开
+                        } else {
+                            log.info("设备刷新登录： 设备号 = " + session.getSessionId());
+                            SessionMap.inst().updateSession(finalDeviceCode ,session,meterNum);
+                            //log.info("设备登录失败： 设备号 = " + session.getSessionId());
+                            //ServerSession.closeSession(ctx);
+                            // 假如说已经在会话中了，直接断开连接
+                            //ctx.close();
+                        }
+                    }
+                    //有异常的话，我们进行处理
+                    @Override
+                    public void onException(Throwable t) {
+                        log.info("设备登录异常： 设备号 = " + session.getSessionId());
+                        ServerSession.closeSession(ctx);
+                    }
+                });
+            }
+        }
+    }
+
+    private boolean action(ServerSession session, String deviceCode, ChannelHandlerContext ctx, String meterNum) {
+        //user验证
+        boolean isValidUser = checkUser(deviceCode,session);
+        if(!isValidUser) {
+            //我们发送登录成功的报文给 客户端
+            session.bind();
+            String responseStr = MeterUtils.createLoginOrHeartCheckCode(meterNum, FourthGEnum.LOGIN_TYPE.getCode());
+            session.writeAndFlush(MeterUtils.getByteBuf(ctx, responseStr));
+            log.info("发送登录确认报文 ==> {}", responseStr);
+            return false;
+        }
+        //我们发送登录成功的报文给 客户端
+        session.bind();
+        String responseStr = MeterUtils.createLoginOrHeartCheckCode(meterNum, FourthGEnum.LOGIN_TYPE.getCode());
+        session.writeAndFlush(MeterUtils.getByteBuf(ctx, responseStr));
+        log.info("发送登录确认报文 ==> {}", responseStr);
+        return true;
+    }
+
+    private boolean checkUser(String deviceCode,ServerSession session) {
+        //当前用户已经登录
+        if(SessionMap.inst().hasLogin(deviceCode)) {
+            log.info("设备已经登录： 设备号 = " + deviceCode);
+            return false;
+        }
+        //一般情况下，我们会将 user存储到 DB中，然后对user的用户名和密码进行校验
+        //但是，我们这边没有进行db的集成，所以我们想一个别的办法进行user的校验。在我们的sessionMap进行以下校验
+        //为什么选sessionmap，因为我们user的会话，都是存储到sessionmap中的，sessionmap中只要有这个user的会话，说明就是ok的
+        return true;
+    }
+
+
+}

user-service/src/main/java/com/mh/user/netty/fourg/session/ServerSession.java

@@ -0,0 +1,66 @@
+package com.mh.user.netty.fourg.session;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.util.AttributeKey;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+
+@Data
+@Slf4j
+public class ServerSession {
+    public static final AttributeKey<ServerSession> SESSION_KEY =
+            AttributeKey.valueOf("SESSION_KEY");
+    //通道
+    private Channel channel;
+    private final String sessionId;
+    private boolean isLogin = false;
+
+    public ServerSession(Channel channel, String deviceCode){
+        this.channel = channel;
+        this.sessionId = deviceCode;
+    }
+
+    //session需要和通道进行一定的关联，他是在构造函数中关联上的；
+    //session还需要通过sessionkey和channel进行再次的关联；channel.attr方法.set当前的
+    // serverSession
+    //session需要被添加到我们的SessionMap中
+    public void bind(){
+        log.info("server Session 会话进行绑定 ：" + channel.remoteAddress());
+        channel.attr(SESSION_KEY).set(this);
+        SessionMap.inst().addSession(sessionId, this);
+        this.isLogin = true;
+    }
+
+    //通过channel获取session
+    public static ServerSession getSession(ChannelHandlerContext ctx){
+        Channel channel = ctx.channel();
+        return channel.attr(SESSION_KEY).get();
+    }
+
+    //关闭session,新增返回一个meterNum用于纪录设备下线时间2024-05-08
+    public static String closeSession(ChannelHandlerContext ctx){
+        String meterNum = null;
+        ServerSession serverSession = ctx.channel().attr(SESSION_KEY).get();
+        if(serverSession != null && serverSession.getSessionId() != null) {
+            ChannelFuture future = serverSession.channel.close();
+            future.addListener((ChannelFutureListener) future1 -> {
+                if(!future1.isSuccess()) {
+                    log.info("Channel close error!");
+                }
+            });
+            ctx.close();
+            meterNum = serverSession.sessionId;
+            SessionMap.inst().removeSession(serverSession.sessionId);
+            log.info(ctx.channel().remoteAddress()+"  "+serverSession.sessionId + "==>移除会话");
+        }
+        return meterNum;
+    }
+
+    //写消息
+    public void writeAndFlush(Object msg) {
+        channel.writeAndFlush(msg);
+    }
+}

user-service/src/main/java/com/mh/user/netty/fourg/session/SessionMap.java

@@ -0,0 +1,96 @@
+package com.mh.user.netty.fourg.session;
+
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+@Data
+@Slf4j
+public class SessionMap {
+
+    private ThreadLocal<Boolean> sceneThreadLocal = new ThreadLocal<>();
+
+    //用单例模式进行sessionMap的创建
+    private SessionMap(){}
+
+    private static SessionMap singleInstance = new SessionMap();
+
+    public static SessionMap inst() {
+        return singleInstance;
+    }
+
+    //进行会话的保存
+    //key 我们使用 sessionId；value 需要是 serverSession
+    private ConcurrentHashMap<String, ServerSession> map = new ConcurrentHashMap<>(256);
+    //添加session
+    public void addSession(String sessionId, ServerSession s) {
+        map.put(sessionId, s);
+        log.info("IP地址:"+s.getChannel().remoteAddress()+"  "+ sessionId + " 表具上线,总共表具：" + map.size());
+    }
+
+    //删除session
+    public void removeSession(String sessionId) {
+        if(map.containsKey(sessionId)) {
+            ServerSession s = map.get(sessionId);
+            map.remove(sessionId);
+            log.info("设备id下线:{},在线设备：{}", s.getSessionId(), map.size() );
+        }
+        return;
+    }
+
+    public boolean hasLogin(String sessionId) {
+        Iterator<Map.Entry<String, ServerSession>> iterator = map.entrySet().iterator();
+        while(iterator.hasNext()) {
+            Map.Entry<String, ServerSession> next = iterator.next();
+            if(sessionId != null && sessionId.equalsIgnoreCase(next.getValue().getSessionId())) {
+                return true ;
+            }
+        }
+        return false;
+    }
+
+    //如果在线，肯定有sessionMap里保存的 serverSession
+    //如果不在线，serverSession也没有。用这个来判断是否在线
+    public List<ServerSession> getSessionBy(String sessionId) {
+        return map.values().stream().
+                filter(s -> s.getSessionId().equals(sessionId)).
+                collect(Collectors.toList());
+    }
+
+    public boolean getScene() {
+        return sceneThreadLocal.get();
+    }
+
+    public void initScene(Boolean status) {
+        if (sceneThreadLocal == null) {
+            log.info("======创建ThreadLocal======");
+            sceneThreadLocal = new ThreadLocal<>();
+        }
+        log.info("设置状态==>" + status);
+        sceneThreadLocal.set(status);
+    }
+
+    public void clearScene() {
+        initScene(null);
+        sceneThreadLocal.remove();
+    }
+
+    public void updateSession(String sessionId, ServerSession session, String meterNum) {
+        Iterator<Map.Entry<String, ServerSession>> iterator = map.entrySet().iterator();
+        while(iterator.hasNext()) {
+            Map.Entry<String, ServerSession> next = iterator.next();
+            if (next.getKey().contains(meterNum)){
+                iterator.remove();
+            }
+            if(sessionId != null && sessionId.equalsIgnoreCase(next.getValue().getSessionId())) {
+                next.setValue(session);
+            }
+        }
+    }
+
+}

user-service/src/main/java/com/mh/user/netty/fourg/task/CallbackTask.java

@@ -0,0 +1,20 @@
+package com.mh.user.netty.fourg.task;
+
+/**
+ * @author LJF
+ * @version 1.0
+ * @project TAD_Server
+ * @description 回调任务
+ * @date 2023/7/3 15:34:11
+ */
+public interface CallbackTask<T> {
+    T execute() throws Exception;
+
+    /**
+     * // 执行没有 异常的情况下的 返回值
+     * @param t
+     */
+    void onBack(T t);
+
+    void onException(Throwable t);
+}

user-service/src/main/java/com/mh/user/netty/fourg/task/CallbackTaskScheduler.java

@@ -0,0 +1,76 @@
+package com.mh.user.netty.fourg.task;
+
+import com.google.common.util.concurrent.*;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+/**
+ * @author LJF
+ * @version 1.0
+ * @project TAD_Server
+ * @description 回调任务
+ * @date 2023/7/3 15:34:11
+ */
+public class CallbackTaskScheduler extends Thread {
+    private ConcurrentLinkedQueue<CallbackTask> executeTaskQueue =
+            new ConcurrentLinkedQueue<>();
+    private long sleepTime = 1000 * 10;
+    private final ExecutorService pool = Executors.newCachedThreadPool();
+    ListeningExecutorService lpool = MoreExecutors.listeningDecorator(pool);
+    private static CallbackTaskScheduler inst = new CallbackTaskScheduler();
+    private CallbackTaskScheduler() {
+        this.start();
+    }
+    //add task
+    public static <T> void add(CallbackTask<T> executeTask) {
+        inst.executeTaskQueue.add(executeTask);
+    }
+
+    @Override
+    public void run() {
+        while (true) {
+            handleTask();
+            //为了避免频繁连接服务器,但是当前连接服务器过长导致失败
+            //threadSleep(sleepTime);
+        }
+    }
+
+    private void threadSleep(long sleepTime) {
+        try {
+            Thread.sleep(sleepTime);
+        }catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    //任务执行
+    private void handleTask() {
+        CallbackTask executeTask = null;
+        while (executeTaskQueue.peek() != null) {
+            executeTask = executeTaskQueue.poll();
+            handleTask(executeTask);
+        }
+    }
+    private <T> void handleTask(CallbackTask<T> executeTask) {
+        ListenableFuture<T> future = lpool.submit(new Callable<T>() {
+            public T call() throws Exception {
+                return executeTask.execute();
+            }
+        });
+        Futures.addCallback(future, new FutureCallback<T>() {
+            @Override
+            public void onSuccess(T t) {
+                executeTask.onBack(t);
+            }
+
+            @Override
+            public void onFailure(Throwable throwable) {
+                executeTask.onException(throwable);
+            }
+        }, lpool);
+    }
+}
+

user-service/src/main/java/com/mh/user/netty/fourg/task/ExecuteTask.java

@@ -0,0 +1,6 @@
+package com.mh.user.netty.fourg.task;
+
+//不需要知道异步线程的 返回值
+public interface ExecuteTask {
+    void execute();
+}

user-service/src/main/java/com/mh/user/netty/fourg/task/FutureTaskScheduler.java

@@ -0,0 +1,67 @@
+package com.mh.user.netty.fourg.task;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+/**
+ * @author LJF
+ * @version 1.0
+ * @project TAD_Server
+ * @description 任务定时
+ * @date 2023/7/3 15:34:11
+ */
+public class FutureTaskScheduler extends Thread{
+    private ConcurrentLinkedQueue<ExecuteTask> executeTaskQueue =
+            new ConcurrentLinkedQueue<>();
+    private long sleepTime = 200;
+    private ExecutorService pool = Executors.newFixedThreadPool(10);
+    private static FutureTaskScheduler inst = new FutureTaskScheduler();
+    public FutureTaskScheduler() {
+        this.start();
+    }
+    //任务添加
+    public static void add(ExecuteTask executeTask) {
+        inst.executeTaskQueue.add(executeTask);
+    }
+
+    @Override
+    public void run() {
+        while (true) {
+            handleTask();
+            //threadSleep(sleepTime);
+        }
+    }
+
+    private void threadSleep(long sleepTime) {
+        try {
+            Thread.sleep(sleepTime);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+    //执行任务
+    private void handleTask() {
+        ExecuteTask executeTask;
+        while (executeTaskQueue.peek() != null) {
+            executeTask = executeTaskQueue.poll();
+            handleTask(executeTask);
+        }
+        //刷新心跳时间
+    }
+    private void handleTask(ExecuteTask executeTask) {
+        pool.execute(new ExecuteRunnable(executeTask));
+    }
+
+    class ExecuteRunnable implements Runnable {
+        ExecuteTask executeTask;
+        public ExecuteRunnable(ExecuteTask executeTask) {
+            this.executeTask = executeTask;
+        }
+        @Override
+        public void run() {
+            executeTask.execute();
+        }
+    }
+}

user-service/src/main/java/com/mh/user/service/DeviceCodeParamService.java

@@ -40,4 +40,6 @@ public interface DeviceCodeParamService {
      * 创建指令参数缓存
      */
     void createCodeParams(MeterManageEntity entity);
+
+    DeviceCodeParamEntity queryDeviceCodeParamByDeviceCode(String deviceCode);
 }

user-service/src/main/java/com/mh/user/service/MeterManageService.java

@@ -1,6 +1,7 @@
 package com.mh.user.service;
 
 import com.mh.user.dto.ProDevicesMonitorDTO;
+import com.mh.user.entity.DeviceCodeParamEntity;
 import com.mh.user.entity.MeterManageEntity;
 import com.mh.user.vo.ProMonitorVO;
 
@@ -22,4 +23,8 @@ public interface MeterManageService extends BaseService<MeterManageEntity> {
     int queryIsHotCollection(String systemId, String projectId);
 
     List<ProDevicesMonitorDTO> list(ProMonitorVO proMonitorVO);
+
+    void updateDataByDeviceCode(String deviceCode, String curValue);
+
+    DeviceCodeParamEntity getDeviceCodeParam(String deviceCode);
 }

user-service/src/main/java/com/mh/user/service/impl/DeviceCodeParamServiceImpl.java

@@ -44,6 +44,10 @@ public class DeviceCodeParamServiceImpl implements DeviceCodeParamService {
     @Qualifier("caffeineCache")
     private Cache caffeineCache;
 
+    @Override
+    public DeviceCodeParamEntity queryDeviceCodeParamByDeviceCode(String deviceCode) {
+        return deviceCodeParamMapper.queryDeviceCodeParamByDeviceCode(deviceCode);
+    }
 
     @Override
     public void insertDeviceCodeParam(DeviceCodeParamEntity deviceCodeParamEntity) {

user-service/src/main/java/com/mh/user/service/impl/MeterManageServiceImpl.java

@@ -10,6 +10,7 @@ import com.mh.common.page.PageResult;
 import com.mh.common.utils.StringUtils;
 import com.mh.user.constants.Constant;
 import com.mh.user.dto.ProDevicesMonitorDTO;
+import com.mh.user.entity.DeviceCodeParamEntity;
 import com.mh.user.entity.MeterManageEntity;
 import com.mh.user.entity.ProjectInfoEntity;
 import com.mh.user.mapper.MeterManageMapper;
@@ -45,6 +46,20 @@ public class MeterManageServiceImpl implements MeterManageService {
     @Autowired
     private ProjectInfoMapper projectInfoMapper;
 
+    @Override
+    public DeviceCodeParamEntity getDeviceCodeParam(String deviceCode) {
+        return deviceCodeParamService.queryDeviceCodeParamByDeviceCode(deviceCode);
+    }
+
+    @Override
+    public void updateDataByDeviceCode(String deviceCode, String curValue) {
+        meterManageMapper.update(
+                new UpdateWrapper<MeterManageEntity>()
+                        .eq("mt_code", deviceCode)
+                        .set("cur_value", curValue)
+                        .set("cur_time", new Date()));
+    }
+
     @Override
     public List<ProDevicesMonitorDTO> list(ProMonitorVO proMonitorVO) {
         if (StringUtils.isBlank(proMonitorVO.getUserName()) || StringUtils.isBlank(proMonitorVO.getSystemId())) {

user-service/src/main/java/com/mh/user/strategy/EleYYProtocolStrategy.java

@@ -0,0 +1,432 @@
+package com.mh.user.strategy;
+
+import com.mh.common.utils.StringUtils;
+import com.mh.user.config.ParamConfig;
+import com.mh.user.constants.Constant;
+import com.mh.user.constants.FourthGEnum;
+import com.mh.user.entity.DataResultEntity;
+import com.mh.user.entity.DeviceCodeParamEntity;
+import com.mh.user.entity.MeterManageEntity;
+import com.mh.user.service.DataResultService;
+import com.mh.user.service.MeterManageService;
+import com.mh.user.service.ProjectInfoService;
+import com.mh.user.utils.*;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.ApplicationContext;
+
+import java.math.BigDecimal;
+import java.text.DecimalFormat;
+import java.text.SimpleDateFormat;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.HashMap;
+
+/**
+ * @author LJF
+ * @version 1.0
+ * @project mh_esi
+ * @description 永阳4G电表协议
+ * @date 2024-09-12 14:04:01
+ */
+@Slf4j
+public class EleYYProtocolStrategy implements ProtocolStrategy {
+
+    // 调用service
+    ApplicationContext context = SpringBeanUtil.getApplicationContext();
+    DataResultService dataResultService = context.getBean(DataResultService.class);
+    ProjectInfoService projectInfoService = context.getBean(ProjectInfoService.class);
+    MeterManageService meterManageService = context.getBean(MeterManageService.class);
+    private static final ParamConfig paramConfig = ApplicationContextProvider.getApplicationContext().getBean(ParamConfig.class);
+
+    private static final SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+
+    private static class SingletonHolder {
+        private static final EleYYProtocolStrategy INSTANCE = new EleYYProtocolStrategy();
+    }
+
+    private EleYYProtocolStrategy() {
+        // 防止外部直接实例化
+    }
+
+    public static EleYYProtocolStrategy getInstance() {
+        return SingletonHolder.INSTANCE;
+    }
+
+    @Override
+    public String createOrder(MeterManageEntity meterManageEntity) {
+        //       FE FE FE FE 68 07 00 50 30 02 08 68 1C 0F 38 39 3B 35 33 33 33 33 4D 59 56 3B 4A 3A 56 4A 16
+        // 跳闸例子FE FE FE FE 68 99 99 99 99 99 99 68 1C 0F 33 33 33 33 F7 F6 F5 F4 4D 89 75 3C 38 3A 56 82 16
+//               FE FE FE FE 68 07 00 50 30 02 08 68 1C 0F 38 39 3B 35 33 33 33 33 4D 37 66 3B 4A 3A 56 38 16
+        // 0 代表前面补充0
+        // 12 代表长度为4
+        // d 代表参数为正数型
+        StringBuilder resultStr = new StringBuilder("68");
+        String str = String.format("%012d", Long.parseLong(meterManageEntity.getMtCode()));
+        // 转换位置
+        resultStr.append(ExchangeStringUtil.changePosition(str));
+        StringBuilder appendStr = new StringBuilder();
+        String controlType = "4";
+        switch (controlType) {
+            case "0": // 合闸
+                resultStr.append("681C10");
+                appendStr.append(paramConfig.getMeterPass()).append(paramConfig.getMeterOpcode()).append(FourthGEnum.CLOSE_CONTROL_TYPE.getCode()).append("00").append(TimeUtils.limitDateStr());
+                break;
+            case "1": // 跳闸
+                resultStr.append("681C10");
+                appendStr.append(paramConfig.getMeterPass()).append(paramConfig.getMeterOpcode()).append(FourthGEnum.OPEN_CONTROL_TYPE.getCode()).append("00").append(TimeUtils.limitDateStr());
+                break;
+            case "2": // 跟上报数据报文一致
+                // 响应：680700503002086891 1D 3533B437 595533333333 333333333333333333 83333333 33343333 34 33 F716
+                resultStr.append("681104");
+                appendStr.append(FourthGEnum.READ_UPLOAD_CODE.getCode());
+                break;
+            case "3": // 电表运行状态
+                // 跳闸响应：68 070050300208 68 91 06 36383337 8333（5000） 3616
+                // 68 0600200208 689106 36383337 3333 0616
+                resultStr.append("681104");
+                appendStr.append(FourthGEnum.READ_STATUS_IDENTIFY_CODE.getCode());
+                break;
+            case "4": // 电表总电量
+                // 响应：68070050300208 68 9108 33333333 75493333（42160000） EA16
+                resultStr.append("681104");
+                appendStr.append(FourthGEnum.READ_TOTAL_CHARGE_IDENTIFY_CODE.getCode());
+                break;
+            case "8":
+                // 读取限容
+                resultStr.append("681104");
+                appendStr.append(FourthGEnum.READ_LIMIT_POWER.getCode());
+                break;
+            case "9":
+                //读取软件版本
+                resultStr.append("681104");
+                appendStr.append(FourthGEnum.READ_SOFT_VERSION.getCode());
+                break;
+            case "10":
+                //读取硬件版本
+                resultStr.append("681104");
+                appendStr.append(FourthGEnum.READ_HARD_VERSION.getCode());
+                break;
+            case "11":
+                //读取ICCID
+                resultStr.append("681104");
+                appendStr.append(FourthGEnum.READ_ICCID.getCode());
+                break;
+            case "12":
+                // 读取IMEI
+                resultStr.append("681104");
+                appendStr.append(FourthGEnum.READ_IMEI.getCode());
+                break;
+//            case "13":
+//                // 写IP端口  --未测试
+//                String ipStr = "," + ip;
+//                String ipHex = ExchangeStringUtil.stringToHex(ipStr).replace(" ", "");
+//                int len = (ipHex.length() / 2) + 2;
+//                if (len > 99) {
+//                    throw new RuntimeException("长度异常");
+//                }
+//                String formattedLen = String.format("%02d", len);
+//                String string = formattedLen + ipStr;
+//                ipHex = ExchangeStringUtil.stringToHex(string).replace(" ", "");
+//                appendStr.append(FourthGEnum.WRITE_IP.getCode());
+//                appendStr.append(paramConfig.getMeterPass())
+//                        .append(paramConfig.getMeterOpcode())
+//                        .append(ipHex);
+//                resultStr.append("6814").append(ExchangeStringUtil.decToHex(String.valueOf(appendStr.length() / 2)));
+//                break;
+            case "14":
+                // 读IP端口
+                resultStr.append("681104");
+                appendStr.append(FourthGEnum.READ_IP.getCode());
+                break;
+            case "15":
+                // 升级
+                appendStr.append(FourthGEnum.UPGRADES.getCode())
+                        .append(paramConfig.getMeterPass())
+                        .append(paramConfig.getMeterOpcode())
+                        .append("01"); //写入1
+                resultStr.append("6814").append(ExchangeStringUtil.decToHex(String.valueOf(appendStr.length() / 2)));
+                break;
+            default:
+                break;
+        }
+        // 加33操作
+        resultStr.append(ExchangeStringUtil.addThree(appendStr.toString()));
+        // 检验和
+        String checkSum = ExchangeStringUtil.makeChecksum(resultStr.toString());
+        resultStr.append(checkSum).append("16");
+        resultStr.insert(0, "FEFEFEFE");
+        return resultStr.toString().toUpperCase();
+    }
+
+    @Override
+    public String analysisReceiveData(DeviceCodeParamEntity deviceCodeParamEntity, String receiveData) {
+        log.info("永阳电表97/07规约协议：工厂解析报文，接收到的报文<=={}",receiveData);
+        HashMap<String, Object> dealMeterUploadData = dealMeterUploadData(receiveData);
+        if (dealMeterUploadData == null) {
+            return Constant.FAIL;
+        }
+        String deviceCode = (String) dealMeterUploadData.get("deviceCode");
+        String curValue = (String) dealMeterUploadData.get("curValue");
+        if (StringUtils.isBlank(curValue) || Constant.FAIL.equals(curValue)) {
+            return Constant.FAIL;
+        }
+        curValue = String.valueOf(Double.valueOf(curValue)); //00010.76,去除读数前面带0的情况
+        // 解析入库
+        // 根据deviceCode查询deviceCodeParam对应参数
+        DeviceCodeParamEntity deviceCodeParam = meterManageService.getDeviceCodeParam(deviceCode);
+        if (deviceCodeParam == null) {
+            log.error("未找到对应的参数");
+            return null;
+        }
+        analysisMeterOrder485(curValue, deviceCodeParam);
+
+        // 更新meter_manage
+        meterManageService.updateDataByDeviceCode(deviceCode, curValue);
+
+        return curValue;
+    }
+
+    /**
+     * 处理数据上报报文
+     *
+     * @param receiveStr
+     */
+    public static HashMap<String, Object> dealMeterUploadData(String receiveStr) {
+        if (org.apache.commons.lang3.StringUtils.isBlank(receiveStr)) {
+            return null;
+        }
+        HashMap<String, Object> hashMap = new HashMap<>();
+        // 获取表号
+        String deviceCode = ExchangeStringUtil.getDeviceCode(receiveStr);
+        // 获取功能码
+        String funCode = receiveStr.substring(16, 18);
+        // 获取数据位大小
+        Integer dataSize = Integer.parseInt(receiveStr.substring(18, 20), 16);
+        // 获取标识码
+        String identifyCode = ExchangeStringUtil.cutThree(receiveStr.substring(20, 28));
+        if ((funCode.equals("8E") || funCode.equals("91"))
+                && "02008104".equals(identifyCode)) {
+            if ("8E".equals(funCode)) {
+                log.info("主动参数上报报文 <== {}", receiveStr);
+            } else {
+                log.info("下发指令参数上报报文 <== {}", receiveStr);
+            }
+            // 总电量（当前电表读数）
+            // 单相：68060020810208 6891 08 33333333 45493333 8A16
+            // 三相：68050080300218 6891 1D 3533B437 33333333333333333333333333333374333333333433333335E516
+            // 被动上报报文   6805008030021868 91 1D 3533B437 3333333333333333333333333333337A3333333334333333 35 E516
+            // 主动上报报文   6802001031020868 8E 1D 3533B437 345533333333333333333333333333B43433333334333333 33 BC16
+            BigDecimal curValue = new BigDecimal(String.valueOf(Integer.parseInt(ExchangeStringUtil.changePosition(ExchangeStringUtil.cutThree(receiveStr.substring(58, 66)))) / 100.00));
+            DecimalFormat decimalFormat = new DecimalFormat("#0.00");
+            String result = decimalFormat.format(curValue);
+            // 剩余电量
+            BigDecimal remainingValue = new BigDecimal(String.valueOf(Integer.parseInt(ExchangeStringUtil.changePosition(ExchangeStringUtil.cutThree(receiveStr.substring(66, 74)))) / 100.00));
+            String remaining = decimalFormat.format(remainingValue);
+            // 继电器状态
+            int deviceStatus = Integer.parseInt(ExchangeStringUtil.cutThree(receiveStr.substring(74, 76)));
+            int errorCode = Integer.parseInt(ExchangeStringUtil.cutThree(receiveStr.substring(76, 78)));
+            log.info("数据标识：{}，表号==>{}，当前读数==>{}，当前剩余量==>{}，当前表具状态==>{},故障状态字==>{}",
+                    identifyCode, deviceCode, result, remaining, deviceStatus == 1 ? "拉闸" : "合闸",
+                    errorCode == 1 ? "预警" : errorCode == 2 ? "报警" : errorCode == 3 ? "欠费停电" : errorCode == 5 ? "过载跳闸" : "其他:" + errorCode);
+
+            hashMap.put("deviceCode", deviceCode);
+            hashMap.put("curValue", result);
+            hashMap.put("remaining", remaining);
+            hashMap.put("deviceStatus", deviceStatus);
+            hashMap.put("errorCode", errorCode);
+            return hashMap;
+        } else if (funCode.equals("8E") && "03008104".equals(identifyCode)) {
+            log.info("主动故障报警上报 <== {}", receiveStr);
+            //6802001031020868 8E 1D 3533B437 345533333333333333333333333333B43433333334333333 33 BC16 参数状态上报中含有故障状态，对比故障状态是否一致
+            //6802001031020868 8E 0E 3633B437 38 34B434333333343333 9416
+            int errorCode = Integer.parseInt(ExchangeStringUtil.cutThree(receiveStr.substring(28, 30)));
+            int relayStatus = Integer.parseInt(ExchangeStringUtil.cutThree(receiveStr.substring(30, 32)));
+            String eleVal = receiveStr.substring(32, 40);
+            BigDecimal eleValue = new BigDecimal(String.valueOf(Integer.parseInt(ExchangeStringUtil.changePosition(ExchangeStringUtil.cutThree(eleVal))) / 100.00));
+            String residualEle = receiveStr.substring(40, 48);
+            BigDecimal residualEleValue = new BigDecimal(String.valueOf(Integer.parseInt(ExchangeStringUtil.changePosition(ExchangeStringUtil.cutThree(residualEle))) / 100.00));
+            log.info("故障码==>{}，表号==>{}，继电器状态==>{}，当前总电量==>{}，当前剩余量==>{}",
+                    errorCode == 1 ? "预警" : errorCode == 2 ? "报警" : errorCode == 3 ? "欠费停电" : errorCode == 5 ? "过载跳闸" : "其他:" + errorCode,
+                    deviceCode, relayStatus == 1 ? "拉闸" : "合闸", eleValue, residualEleValue);
+
+            hashMap.put("deviceCode", deviceCode);
+            hashMap.put("curValue", eleValue);
+            hashMap.put("remaining", residualEleValue);
+            hashMap.put("deviceStatus", relayStatus);
+            hashMap.put("errorCode", errorCode);
+            return hashMap;
+        } else if ("91".equals(funCode) && "03050004".equals(identifyCode)) {
+            log.info("下发指令读取运行状态上报 <== {}", receiveStr);
+            String status = ExchangeStringUtil.cutThree(receiveStr.substring(28, 32));
+            String statusHex = ExchangeStringUtil.hex2Binary(status);
+            if (org.apache.commons.lang3.StringUtils.isNotBlank(statusHex) && statusHex.length() == 16) {
+                // 运行时段：0 第一套，1 第二套
+                String timePeriod = statusHex.substring(15, 16);
+                // 供电方式：00 主电源， 01 辅助电源， 10 电池供电
+                String powerMethod = statusHex.substring(13, 15);
+                // 编程允许：0 禁止， 1 许可
+                String programmingAllows = statusHex.substring(12, 13);
+                // 继电器状态：0 通， 1 断
+                String relayStatus = statusHex.substring(11, 12);
+                // 当前运行时区： 0 第一套， 1 第二套
+                String timeZone = statusHex.substring(10, 11);
+                // 继电器命令状态：0 通， 1 断
+                String relayOrderStatus = statusHex.substring(9, 10);
+                // 预跳闸报警状态：0 通， 1 断
+                String closeAlarmStatus = statusHex.substring(8, 9);
+                // 电能表类型：00 非预付费表 01 电量型预付费表 10 电费型预付费表
+                String meterType = statusHex.substring(6, 8);
+                if (meterType.equals("00")) {
+                    meterType = "非预付费表";
+                } else if (meterType.equals("01")) {
+                    meterType = "电量型预付费表";
+                } else if (meterType.equals("10")) {
+                    meterType = "电费型预付费表";
+                }
+                // 当前运行分时费率：0 第一套， 1 第二套
+                String rate = statusHex.substring(5, 6);
+                // 当前阶梯：0 第一套， 1 第二套
+                String ladder = statusHex.substring(4, 5);
+                // 保电：0 未保电 1 保电
+                String protect = statusHex.substring(3, 4);
+                log.info("继电器状态：{}，继电器命令状态：{}，保电：{}，编程允许：{}，电能表类型：{}",
+                        relayStatus.equals("0") ? "通" : "断",
+                        relayOrderStatus.equals("0") ? "通" : "断",
+                        protect.equals("0") ? "未保电" : "保电",
+                        programmingAllows.equals("0") ? "禁止" : "许可",
+                        meterType
+                );
+            }
+        } else if ("91".equals(funCode) && "00000000".equals(identifyCode)) {
+            log.info("下发指令读取总电能上报 <== {}", receiveStr);
+            String totalPower = ExchangeStringUtil.cutThree(receiveStr.substring(28, 36));
+            totalPower = ExchangeStringUtil.changePosition(totalPower);
+            BigDecimal curValue = new BigDecimal(String.valueOf(Integer.parseInt(totalPower) / 100.00));
+            DecimalFormat decimalFormat = new DecimalFormat("#0.00");
+            String result = decimalFormat.format(curValue);
+            log.info("总电能：{}", result);
+        } else if ("91".equals(funCode) && FourthGEnum.READ_LIMIT_POWER.getCode().equals(identifyCode)) {
+            log.info("下发读取限容功率值上报 <== {}", receiveStr);
+            String limitPower = ExchangeStringUtil.cutThree(receiveStr.substring(28, 36));
+            limitPower = ExchangeStringUtil.changePosition(limitPower);
+            BigDecimal limitPV = new BigDecimal(String.valueOf(Integer.parseInt(limitPower) / 10.00));
+            String allowTimes = ExchangeStringUtil.cutThree(receiveStr.substring(36, 38));
+            allowTimes = ExchangeStringUtil.changePosition(allowTimes);
+            BigDecimal allowVal = new BigDecimal(String.valueOf(Integer.parseInt(allowTimes)));
+            String usedTimes = ExchangeStringUtil.cutThree(receiveStr.substring(38, 40));
+            usedTimes = ExchangeStringUtil.changePosition(usedTimes);
+            BigDecimal usedVal = new BigDecimal(String.valueOf(Integer.parseInt(usedTimes)));
+
+            log.info("限制功率值==> {},允许合闸次数==>{},使用次数==>{}", limitPV, allowVal, usedVal);
+            hashMap.put("deviceCode", deviceCode);
+            hashMap.put("limitPower", limitPV);
+            hashMap.put("maxSwitchOn", allowVal);
+            hashMap.put("usedSwitchOn", usedVal);
+            hashMap.put("mark", "powerData");
+            return hashMap;
+        }
+        return null;
+    }
+
+    private static String getAnalysisData(String receiveData, String checkStr, String meterId) {
+        StringBuilder stringBuilder = new StringBuilder();
+        // 累计总读数据：FE FE FE FE 68 69 04 00 70 01 08 68 81 06 43 C3 A5 33 33 33 81 16
+        // 瞬时有功功率：FE FE FE FE 68 69 04 00 70 01 08 68 81 05 63 E9 33 33 33 21 16
+        //            FE FE FE FE 68 98 01 50 04 00 00 68 C1 01 35 B4 16
+        // 判断属于哪个数据标识
+        String resultData = Constant.FAIL;
+        if (receiveData.length() > 24) {
+            String identifyCode = ExchangeStringUtil.changePosition(ExchangeStringUtil.cutThree(receiveData.substring(20, 24)));
+            switch (identifyCode) {
+                case "9010":
+                    // 累计总读数据
+                    for (int i = 0; i < 4; i++) {
+                        String data1 = checkStr.substring(32 - 2 * (i + 1), 32 - 2 * i);
+                        stringBuilder.append(data1);
+                    }
+                    resultData = stringBuilder.toString();
+                    resultData = ExchangeStringUtil.cutThree(resultData);
+                    // 0 代表前面补充0,4 代表长度为4,d 代表参数为正数型
+                    resultData = String.format("%08d", Long.parseLong(resultData));
+                    resultData = resultData.substring(0, 6) + "." + resultData.substring(6, 8);
+                    log.info("电表表号：{},电表累计总量读数：{}", meterId, resultData);
+                    break;
+                case "B630":
+                    // 瞬时有功功率
+                    for (int i = 0; i < 3; i++) {
+                        String data1 = checkStr.substring(30 - 2 * (i + 1), 30 - 2 * i);
+                        stringBuilder.append(data1);
+                    }
+                    resultData = stringBuilder.toString();
+                    resultData = ExchangeStringUtil.cutThree(resultData);
+                    // 0 代表前面补充0,4 代表长度为4,d 代表参数为正数型
+                    resultData = String.format("%06d", Long.parseLong(resultData));
+                    resultData = resultData.substring(0, 2) + "." + resultData.substring(2, 6);
+                    log.info("电表表号：{},电表瞬时有功功率读数：{}", meterId, resultData);
+                    break;
+            }
+        }
+        return resultData;
+    }
+
+    /**
+     * 解析电表返回的数据
+     *
+     * @param dataStr
+     */
+    public void analysisMeterOrder485(final String dataStr, final DeviceCodeParamEntity deviceCodeParam) {
+//        threadPoolService.execute(() -> {
+            Date date = new Date();
+        // 使用Calendar进行更安全的时间处理
+        Calendar cal = Calendar.getInstance();
+        cal.setTime(date);
+
+        if (ExchangeStringUtil.isInDate(date, "00:00:00", "00:05:00")) {
+            cal.set(Calendar.MINUTE, 0);
+        } else if (ExchangeStringUtil.isInDate(date, "00:05:00", "00:10:00")) {
+            cal.set(Calendar.MINUTE, 5);
+        } else if (ExchangeStringUtil.isInDate(date, "00:10:00", "00:15:00")) {
+            cal.set(Calendar.MINUTE, 10);
+        } else if (ExchangeStringUtil.isInDate(date, "00:15:00", "00:20:00")) {
+            cal.set(Calendar.MINUTE, 15);
+        } else if (ExchangeStringUtil.isInDate(date, "00:20:00", "00:25:00")) {
+            cal.set(Calendar.MINUTE, 20);
+        } else if (ExchangeStringUtil.isInDate(date, "00:25:00", "00:30:00")) {
+            cal.set(Calendar.MINUTE, 25);
+        } else if (ExchangeStringUtil.isInDate(date, "00:30:00", "00:35:00")) {
+            cal.set(Calendar.MINUTE, 30);
+        } else if (ExchangeStringUtil.isInDate(date, "00:35:00", "00:40:00")) {
+            cal.set(Calendar.MINUTE, 35);
+        } else if (ExchangeStringUtil.isInDate(date, "00:40:00", "00:45:00")) {
+            cal.set(Calendar.MINUTE, 40);
+        } else if (ExchangeStringUtil.isInDate(date, "00:45:00", "00:50:00")) {
+            cal.set(Calendar.MINUTE, 45);
+        } else if (ExchangeStringUtil.isInDate(date, "00:50:00", "00:55:00")) {
+            cal.set(Calendar.MINUTE, 50);
+        } else if (ExchangeStringUtil.isInDate(date, "00:55:00", "00:59:00")) {
+            cal.set(Calendar.MINUTE, 55);
+        }
+        cal.set(Calendar.SECOND, 0);
+        cal.set(Calendar.MILLISECOND, 0);
+        date = cal.getTime();
+            try {
+                DataResultEntity dataResultEntity = new DataResultEntity();
+                dataResultEntity.setDeviceAddr(deviceCodeParam.getDeviceAddr());    //通讯编号
+                dataResultEntity.setDeviceType(deviceCodeParam.getDeviceType());                              //类型
+                dataResultEntity.setProjectId(deviceCodeParam.getProjectId());      //所属项目
+                dataResultEntity.setCurValue(new BigDecimal(dataStr));              //当前读数
+                dataResultEntity.setCurDate(date);                   //当前日期
+                dataResultEntity.setGrade(deviceCodeParam.getGrade());
+                dataResultEntity.setRegisterAddr(deviceCodeParam.getRegisterAddr());
+                dataResultEntity.setRegisterName(deviceCodeParam.getRegisterName());
+                String projectName = projectInfoService.selectName(deviceCodeParam.getProjectId());
+                log.info("电表==>{},读数==>{},项目名称==>{}", deviceCodeParam.getDeviceAddr(), dataStr, projectName);
+                dataResultService.saveDataResult(dataResultEntity);
+                log.info("电表保存数据成功!项目名称:{}", projectName);
+            } catch (Exception e) {
+                log.error("保存电表数据失败!", e);
+            }
+//        });
+    }
+
+}

user-service/src/main/java/com/mh/user/utils/ApplicationContextProvider.java

@@ -0,0 +1,65 @@
+package com.mh.user.utils;
+
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.stereotype.Component;
+
+/**
+ * @Classname ApplicationContextProvider
+ * @Description: 获取bean对象的工具类
+ * @Date 2020-07-14 10:44
+ * @Created by NING MEI
+ */
+@Component
+public class ApplicationContextProvider implements ApplicationContextAware {
+    /**
+     * 上下文对象实例
+     */
+    private static ApplicationContext applicationContext;
+    @Override
+    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+        this.applicationContext = applicationContext;
+    }
+    /**
+     * 获取applicationContext
+     *
+     * @return
+     */
+    public static ApplicationContext getApplicationContext() {
+        return applicationContext;
+    }
+
+    /**
+     * 通过name获取 Bean.
+     *
+     * @param name
+     * @return
+     */
+    public static Object getBean(String name) {
+        return getApplicationContext().getBean(name);
+    }
+
+    /**
+     * 通过class获取Bean.
+     *
+     * @param clazz
+     * @param <T>
+     * @return
+     */
+    public static <T> T getBean(Class<T> clazz) {
+        return getApplicationContext().getBean(clazz);
+    }
+
+    /**
+     * 通过name,以及Clazz返回指定的Bean
+     *
+     * @param name
+     * @param clazz
+     * @param <T>
+     * @return
+     */
+    public static <T> T getBean(String name, Class<T> clazz) {
+        return getApplicationContext().getBean(name, clazz);
+    }
+}

user-service/src/main/java/com/mh/user/utils/ExchangeStringUtil.java

@@ -2,8 +2,10 @@ package com.mh.user.utils;
 
 import com.alibaba.fastjson2.JSONObject;
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
 import org.apache.commons.codec.binary.Hex;
+import org.apache.commons.lang3.StringUtils;
 import sun.misc.BASE64Encoder;
 
 import java.io.*;
@@ -383,6 +385,31 @@ public class ExchangeStringUtil {
         return stringBuilder.toString();
     }
 
+    public static String bytesToHexString(Object msg){
+        ByteBuf byteBuf = (ByteBuf)msg;
+        ByteBuf buf = Unpooled.copiedBuffer(byteBuf);
+        byte [] src = new byte[buf.readableBytes()];
+        //复制内容到字节数组bytes
+        buf.readBytes(src);
+        StringBuilder stringBuilder = new StringBuilder();
+        if (src == null || src.length <= 0) {
+            return null;
+        }
+        for (int i = 0; i < src.length; i++) {
+            int v = src[i] & 0xFF;
+            String hv = Integer.toHexString(v);
+            if (hv.length() < 2) {
+                stringBuilder.append(0);
+            }
+            stringBuilder.append(hv);
+        }
+        String rawMessage = stringBuilder.toString().toUpperCase();
+        int first68Index = rawMessage.indexOf("68");
+        if (first68Index == -1){
+        }
+        return rawMessage.substring(first68Index);
+    }
+
     /**
      * 字符串转化成为16进制字符串
      *
@@ -1351,4 +1378,83 @@ public class ExchangeStringUtil {
         return Date.from(localDateTime.toInstant(ZoneOffset.of("+8")));
     }
 
+    /**
+     * 获取设备号
+     *
+     * @param receiveStr
+     * @return
+     */
+    public static String getDeviceCode(String receiveStr) {
+        // 表号
+        String meterID = "";
+        if (StringUtils.isNotBlank(receiveStr)) {
+            meterID = ExchangeStringUtil.changePosition(receiveStr.substring(2, 14));
+        }
+        return meterID;
+    }
+
+    /**
+     * 16进制转换成二进制
+     */
+    public static String hex2Binary(String string) {
+        // TODO Auto-generated method stub
+        String s = "";
+        for (int i = 0; i < string.length(); i++) {
+            s += char2Binary(string.charAt(i));
+        }
+        return s;
+    }
+
+    private static String char2Binary(char charAt) {
+        // TODO Auto-generated method stub
+        switch (charAt) {
+            case '0':
+                return "0000";
+            case '1':
+                return "0001";
+            case '2':
+                return "0010";
+            case '3':
+                return "0011";
+            case '4':
+                return "0100";
+            case '5':
+                return "0101";
+            case '6':
+                return "0110";
+            case '7':
+                return "0111";
+            case '8':
+                return "1000";
+            case '9':
+                return "1001";
+            case 'a':
+                return "1010";
+            case 'b':
+                return "1011";
+            case 'c':
+                return "1100";
+            case 'd':
+                return "1101";
+            case 'e':
+                return "1110";
+            case 'f':
+                return "1111";
+            case 'A':
+                return "1010";
+            case 'B':
+                return "1011";
+            case 'C':
+                return "1100";
+            case 'D':
+                return "1101";
+            case 'E':
+                return "1110";
+            case 'F':
+                return "1111";
+
+        }
+        return null;
+    }
+
 }

user-service/src/main/java/com/mh/user/utils/MeterUtils.java

@@ -0,0 +1,618 @@
+package com.mh.user.utils;
+
+import com.mh.user.config.ParamConfig;
+import com.mh.user.constants.FourthGEnum;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+
+import java.math.BigDecimal;
+import java.text.DecimalFormat;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.HashMap;
+
+/**
+ * @author LJF
+ * @version 1.0
+ * @project TAD_Server
+ * @description 电表工具类
+ * @date 2023/7/4 08:45:16
+ */
+@Slf4j
+public class MeterUtils {
+
+    private static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+
+    private static final ParamConfig paramConfig = ApplicationContextProvider.getApplicationContext().getBean(ParamConfig.class);
+
+    public static String removeLeadingFE(String message) {
+        // 找到报文开头处的连续“FE”字符
+        int leadingFeCount = 0;
+        while (message.startsWith("FE", leadingFeCount)) {
+            leadingFeCount += 2;
+        }
+
+        // 删除前导字符串“FE”
+        if (leadingFeCount > 0) {
+            message = message.substring(leadingFeCount);
+        }
+
+        return message;
+    }
+
+    public static ByteBuf getByteBuf(ChannelHandlerContext ctx, String sendStr) {
+        // byte类型的数据
+//        String sendStr = "5803004900021914";          // 冷量计
+        // 申请一个数据结构存储信息
+        ByteBuf buffer = ctx.alloc().buffer();
+        // 将信息放入数据结构中
+        buffer.writeBytes(ExchangeStringUtil.hexStrToBinaryStr(sendStr));//对接需要16进制
+        return buffer;
+    }
+
+    public static ByteBuf createByteBuf(String sendStr) {
+        // byte类型的数据
+//        String sendStr = "5803004900021914";          // 冷量计
+        // 申请一个数据结构存储信息
+        ByteBuf buffer = Unpooled.buffer();
+        // 将信息放入数据结构中
+        buffer.writeBytes(ExchangeStringUtil.hexStrToBinaryStr(sendStr));//对接需要16进制
+        return buffer;
+    }
+
+    /**
+     * 校验和
+     *
+     * @param data
+     * @return
+     */
+    public static boolean makeChecksum(String data) {
+        if (data == null || data.equals("")) {
+            return false;
+        }
+        //log.info("接收到的响应报文==>{}",data);
+        //data = data.replaceAll("FE", "");
+        int dataLen = data.length();
+        // 校验和
+        String checkSumData = data.substring(0, dataLen - 4);
+        int total = 0;
+        int len = checkSumData.length();
+        int num = 0;
+        while (num < len) {
+            String s = data.substring(num, num + 2);
+            total += Integer.parseInt(s, 16);
+            num = num + 2;
+        }
+        /**
+         * 用256求余最大是255，即16进制的FF
+         */
+        int mod = total % 256;
+        String hex = Integer.toHexString(mod);
+        len = hex.length();
+        // 如果不够校验位的长度，补0,这里用的是两位校验
+        if (len < 2) {
+            hex = "0" + hex;
+        }
+//        log.info("自己计算的校验和==>" + hex);
+        // 比较校验和
+        String check = data.substring(dataLen - 4, dataLen - 2);
+//        log.info("计算前校验和==>" + check);
+        return hex.equalsIgnoreCase(check);
+
+    }
+
+    /**
+     * 获取控制码
+     *
+     * @param data
+     * @return
+     */
+    public static String getControlCode(String data) {
+        return data.substring(16, 18);
+    }
+
+    /**
+     * 获取是什么类型的心跳
+     *
+     * @param data
+     * @return
+     */
+    public static String getControlType(String data) {
+        return ExchangeStringUtil.cutThree(data.substring(20, 22));
+    }
+
+    /**
+     * 获取设备号
+     *
+     * @param receiveStr
+     * @return
+     */
+    public static String getDeviceCode(String receiveStr) {
+        // 表号
+        String meterID = "";
+        if (StringUtils.isNotBlank(receiveStr)) {
+            meterID = ExchangeStringUtil.changePosition(receiveStr.substring(2, 14));
+        }
+        return meterID;
+    }
+
+    /**
+     * 生成登录确认报文和心跳报文
+     *
+     * @return
+     */
+    public static String createLoginOrHeartCheckCode(String deviceId, String orderType) {
+        // 0 代表前面补充0
+        // 12 代表长度为4
+        // d 代表参数为正数型
+        StringBuilder resultStr = new StringBuilder("68");
+        String str = String.format("%012d", Long.parseLong(deviceId));
+        // 转换位置
+        resultStr.append(ExchangeStringUtil.changePosition(str)).append("680D0B");
+        // 加33操作
+        String appendStr = paramConfig.getMeterPass() + paramConfig.getMeterOpcode() + orderType + "0000";
+        resultStr.append(ExchangeStringUtil.addThree(appendStr));
+        // 检验和
+        String checkSum = ExchangeStringUtil.makeChecksum(resultStr.toString());
+        resultStr.append(checkSum).append("16");
+        resultStr.insert(0, "FEFEFEFE");
+//        68140060300208688D 02 34 4E 8F16 1
+//        6814006030020868 8E 1D3533B4374A5533333333333333333333333333933633333334333333331416
+        return resultStr.toString().toUpperCase();
+    }
+
+    /**
+     * 创建拉合闸确认指令
+     *
+     * @param deviceCode
+     * @return
+     */
+    public static String createControlCheckCode(String deviceCode) {
+        // 0 代表前面补充0
+        // 12 代表长度为4
+        // d 代表参数为正数型
+        StringBuilder resultStr = new StringBuilder("68");
+        String str = String.format("%012d", Long.parseLong(deviceCode));
+        // 转换位置
+        resultStr.append(ExchangeStringUtil.changePosition(str)).append("689C00");
+        // 检验和
+        String checkSum = ExchangeStringUtil.makeChecksum(resultStr.toString());
+        resultStr.append(checkSum).append("16");
+        return resultStr.toString().toUpperCase();
+    }
+
+    /**
+     * 处理数据上报报文
+     *
+     * @param receiveStr
+     */
+    public static HashMap<String, Object> dealMeterUploadData(String receiveStr) {
+        if (StringUtils.isBlank(receiveStr)) {
+            return null;
+        }
+        HashMap<String, Object> hashMap = new HashMap<>();
+        // 获取表号
+        String deviceCode = ExchangeStringUtil.getDeviceCode(receiveStr);
+        // 获取功能码
+        String funCode = receiveStr.substring(16, 18);
+        // 获取数据位大小
+        Integer dataSize = Integer.parseInt(receiveStr.substring(18, 20), 16);
+        // 获取标识码
+        String identifyCode = ExchangeStringUtil.cutThree(receiveStr.substring(20, 28));
+        if ((funCode.equals("8E") || funCode.equals("91"))
+                && "02008104".equals(identifyCode)) {
+            if ("8E".equals(funCode)) {
+                log.info("主动参数上报报文 <== {}", receiveStr);
+            } else {
+                log.info("下发指令参数上报报文 <== {}", receiveStr);
+            }
+            // 总电量（当前电表读数）
+            // 单相：68060020810208 6891 08 33333333 45493333 8A16
+            // 三相：68050080300218 6891 1D 3533B437 33333333333333333333333333333374333333333433333335E516
+            // 被动上报报文   6805008030021868 91 1D 3533B437 3333333333333333333333333333337A3333333334333333 35 E516
+            // 主动上报报文   6802001031020868 8E 1D 3533B437 345533333333333333333333333333B43433333334333333 33 BC16
+            BigDecimal curValue = new BigDecimal(String.valueOf(Integer.parseInt(ExchangeStringUtil.changePosition(ExchangeStringUtil.cutThree(receiveStr.substring(58, 66)))) / 100.00));
+            DecimalFormat decimalFormat = new DecimalFormat("#0.00");
+            String result = decimalFormat.format(curValue);
+            // 剩余电量
+            BigDecimal remainingValue = new BigDecimal(String.valueOf(Integer.parseInt(ExchangeStringUtil.changePosition(ExchangeStringUtil.cutThree(receiveStr.substring(66, 74)))) / 100.00));
+            String remaining = decimalFormat.format(remainingValue);
+            // 继电器状态
+            int deviceStatus = Integer.parseInt(ExchangeStringUtil.cutThree(receiveStr.substring(74, 76)));
+            int errorCode = Integer.parseInt(ExchangeStringUtil.cutThree(receiveStr.substring(76, 78)));
+            log.info("数据标识：{}，表号==>{}，当前读数==>{}，当前剩余量==>{}，当前表具状态==>{},故障状态字==>{}",
+                    identifyCode, deviceCode, result, remaining, deviceStatus == 1 ? "拉闸" : "合闸",
+                    errorCode == 1 ? "预警" : errorCode == 2 ? "报警" : errorCode == 3 ? "欠费停电" : errorCode == 5 ? "过载跳闸" : "其他:" + errorCode);
+
+            hashMap.put("deviceCode", deviceCode);
+            hashMap.put("curValue", result);
+            hashMap.put("remaining", remaining);
+            hashMap.put("deviceStatus", deviceStatus);
+            hashMap.put("errorCode", errorCode);
+            return hashMap;
+        } else if (funCode.equals("8E") && "03008104".equals(identifyCode)) {
+            log.info("主动故障报警上报 <== {}", receiveStr);
+            //6802001031020868 8E 1D 3533B437 345533333333333333333333333333B43433333334333333 33 BC16 参数状态上报中含有故障状态，对比故障状态是否一致
+            //6802001031020868 8E 0E 3633B437 38 34B434333333343333 9416
+            int errorCode = Integer.parseInt(ExchangeStringUtil.cutThree(receiveStr.substring(28, 30)));
+            int relayStatus = Integer.parseInt(ExchangeStringUtil.cutThree(receiveStr.substring(30, 32)));
+            String eleVal = receiveStr.substring(32, 40);
+            BigDecimal eleValue = new BigDecimal(String.valueOf(Integer.parseInt(ExchangeStringUtil.changePosition(ExchangeStringUtil.cutThree(eleVal))) / 100.00));
+            String residualEle = receiveStr.substring(40, 48);
+            BigDecimal residualEleValue = new BigDecimal(String.valueOf(Integer.parseInt(ExchangeStringUtil.changePosition(ExchangeStringUtil.cutThree(residualEle))) / 100.00));
+            log.info("故障码==>{}，表号==>{}，继电器状态==>{}，当前总电量==>{}，当前剩余量==>{}",
+                    errorCode == 1 ? "预警" : errorCode == 2 ? "报警" : errorCode == 3 ? "欠费停电" : errorCode == 5 ? "过载跳闸" : "其他:" + errorCode,
+                    deviceCode, relayStatus == 1 ? "拉闸" : "合闸", eleValue, residualEleValue);
+
+            hashMap.put("deviceCode", deviceCode);
+            hashMap.put("curValue", eleValue);
+            hashMap.put("remaining", residualEleValue);
+            hashMap.put("deviceStatus", relayStatus);
+            hashMap.put("errorCode", errorCode);
+            return hashMap;
+        } else if ("91".equals(funCode) && "03050004".equals(identifyCode)) {
+            log.info("下发指令读取运行状态上报 <== {}", receiveStr);
+            String status = ExchangeStringUtil.cutThree(receiveStr.substring(28, 32));
+            String statusHex = ExchangeStringUtil.hex2Binary(status);
+            if (StringUtils.isNotBlank(statusHex) && statusHex.length() == 16) {
+                // 运行时段：0 第一套，1 第二套
+                String timePeriod = statusHex.substring(15, 16);
+                // 供电方式：00 主电源， 01 辅助电源， 10 电池供电
+                String powerMethod = statusHex.substring(13, 15);
+                // 编程允许：0 禁止， 1 许可
+                String programmingAllows = statusHex.substring(12, 13);
+                // 继电器状态：0 通， 1 断
+                String relayStatus = statusHex.substring(11, 12);
+                // 当前运行时区： 0 第一套， 1 第二套
+                String timeZone = statusHex.substring(10, 11);
+                // 继电器命令状态：0 通， 1 断
+                String relayOrderStatus = statusHex.substring(9, 10);
+                // 预跳闸报警状态：0 通， 1 断
+                String closeAlarmStatus = statusHex.substring(8, 9);
+                // 电能表类型：00 非预付费表 01 电量型预付费表 10 电费型预付费表
+                String meterType = statusHex.substring(6, 8);
+                if (meterType.equals("00")) {
+                    meterType = "非预付费表";
+                } else if (meterType.equals("01")) {
+                    meterType = "电量型预付费表";
+                } else if (meterType.equals("10")) {
+                    meterType = "电费型预付费表";
+                }
+                // 当前运行分时费率：0 第一套， 1 第二套
+                String rate = statusHex.substring(5, 6);
+                // 当前阶梯：0 第一套， 1 第二套
+                String ladder = statusHex.substring(4, 5);
+                // 保电：0 未保电 1 保电
+                String protect = statusHex.substring(3, 4);
+                log.info("继电器状态：{}，继电器命令状态：{}，保电：{}，编程允许：{}，电能表类型：{}",
+                        relayStatus.equals("0") ? "通" : "断",
+                        relayOrderStatus.equals("0") ? "通" : "断",
+                        protect.equals("0") ? "未保电" : "保电",
+                        programmingAllows.equals("0") ? "禁止" : "许可",
+                        meterType
+                );
+            }
+        } else if ("91".equals(funCode) && "00000000".equals(identifyCode)) {
+            log.info("下发指令读取总电能上报 <== {}", receiveStr);
+            String totalPower = ExchangeStringUtil.cutThree(receiveStr.substring(28, 36));
+            totalPower = ExchangeStringUtil.changePosition(totalPower);
+            BigDecimal curValue = new BigDecimal(String.valueOf(Integer.parseInt(totalPower) / 100.00));
+            DecimalFormat decimalFormat = new DecimalFormat("#0.00");
+            String result = decimalFormat.format(curValue);
+            log.info("总电能：{}", result);
+        } else if ("91".equals(funCode) && FourthGEnum.READ_LIMIT_POWER.getCode().equals(identifyCode)) {
+            log.info("下发读取限容功率值上报 <== {}", receiveStr);
+            String limitPower = ExchangeStringUtil.cutThree(receiveStr.substring(28, 36));
+            limitPower = ExchangeStringUtil.changePosition(limitPower);
+            BigDecimal limitPV = new BigDecimal(String.valueOf(Integer.parseInt(limitPower) / 10.00));
+            String allowTimes = ExchangeStringUtil.cutThree(receiveStr.substring(36, 38));
+            allowTimes = ExchangeStringUtil.changePosition(allowTimes);
+            BigDecimal allowVal = new BigDecimal(String.valueOf(Integer.parseInt(allowTimes)));
+            String usedTimes = ExchangeStringUtil.cutThree(receiveStr.substring(38, 40));
+            usedTimes = ExchangeStringUtil.changePosition(usedTimes);
+            BigDecimal usedVal = new BigDecimal(String.valueOf(Integer.parseInt(usedTimes)));
+
+            log.info("限制功率值==> {},允许合闸次数==>{},使用次数==>{}", limitPV, allowVal, usedVal);
+            hashMap.put("deviceCode", deviceCode);
+            hashMap.put("limitPower", limitPV);
+            hashMap.put("maxSwitchOn", allowVal);
+            hashMap.put("usedSwitchOn", usedVal);
+            hashMap.put("mark", "powerData");
+            return hashMap;
+        }
+        return null;
+    }
+
+    /**
+     * 处理拉合闸响应报文
+     *
+     * @param receiveStr
+     */
+    public static String dealMeterControlData(String receiveStr, String controlCode) {
+        String result = "fail";
+        switch (controlCode) {
+            case "9C": // 拉合闸的
+                String checkOrder = createControlCheckCode(getDeviceCode(receiveStr));
+                if (StringUtils.isNotEmpty(checkOrder)
+                        && receiveStr.equalsIgnoreCase(checkOrder)) {
+                    log.info("拉合闸成功 <== {}", receiveStr);
+                    result = "success";
+                } else {
+                    log.info("拉合闸异常");
+                    result = "fail";
+                }
+                break;
+            case "91": // 读取电表的内容
+                log.info("读取电表的数据,并推送到rabbitmq <== {} ", receiveStr);
+//                sendMsgByTopic.sendEleMsg(receiveStr);
+                result = "success";
+                break;
+            case "8F": // 充值电表响应
+                log.info("充值电表响应报文 <== {} ", receiveStr);
+                result = "success";
+                break;
+            case "94": // 写电表的内容
+                result = "success";
+                break;
+            default:
+                break;
+        }
+        return result;
+    }
+
+    /**
+     * 创建跳合闸指令
+     *
+     * @param deviceCode  表号
+     * @param controlType 控制类型：1：跳闸， 0：合闸， 2：读取电表当前状态 3：读取电表总电量
+     * @param chargeValue 充值量
+     * @param chargeTime  充值次数
+     * @return
+     */
+//    public static String createControlCode(String deviceCode, String controlType,
+//                                           String chargeValue, String chargeTime,
+//                                           String powerValue, String times,
+//                                           String ip) {
+//        //       FE FE FE FE 68 07 00 50 30 02 08 68 1C 0F 38 39 3B 35 33 33 33 33 4D 59 56 3B 4A 3A 56 4A 16
+//        // 跳闸例子FE FE FE FE 68 99 99 99 99 99 99 68 1C 0F 33 33 33 33 F7 F6 F5 F4 4D 89 75 3C 38 3A 56 82 16
+////               FE FE FE FE 68 07 00 50 30 02 08 68 1C 0F 38 39 3B 35 33 33 33 33 4D 37 66 3B 4A 3A 56 38 16
+//        // 0 代表前面补充0
+//        // 12 代表长度为4
+//        // d 代表参数为正数型
+//        StringBuilder resultStr = new StringBuilder("68");
+//        String str = String.format("%012d", Long.parseLong(deviceCode));
+//        // 转换位置
+//        resultStr.append(ExchangeStringUtil.changePosition(str));
+//        StringBuilder appendStr = new StringBuilder();
+//        switch (controlType) {
+//            case "0": // 合闸
+//                resultStr.append("681C10");
+//                appendStr.append(paramConfig.getMeterPass()).append(paramConfig.getMeterOpcode()).append(FourthGEnum.CLOSE_CONTROL_TYPE.getCode()).append("00").append(TimeUtils.limitDateStr());
+//                break;
+//            case "1": // 跳闸
+//                resultStr.append("681C10");
+//                appendStr.append(paramConfig.getMeterPass()).append(paramConfig.getMeterOpcode()).append(FourthGEnum.OPEN_CONTROL_TYPE.getCode()).append("00").append(TimeUtils.limitDateStr());
+//                break;
+//            case "2": // 跟上报数据报文一致
+//                // 响应：680700503002086891 1D 3533B437 595533333333 333333333333333333 83333333 33343333 34 33 F716
+//                resultStr.append("681104");
+//                appendStr.append(FourthGEnum.READ_UPLOAD_CODE.getCode());
+//                break;
+//            case "3": // 电表运行状态
+//                // 跳闸响应：68 070050300208 68 91 06 36383337 8333（5000） 3616
+//                // 68 0600200208 689106 36383337 3333 0616
+//                resultStr.append("681104");
+//                appendStr.append(FourthGEnum.READ_STATUS_IDENTIFY_CODE.getCode());
+//                break;
+//            case "4": // 电表总电量
+//                // 响应：68070050300208 68 9108 33333333 75493333（42160000） EA16
+//                resultStr.append("681104");
+//                appendStr.append(FourthGEnum.READ_TOTAL_CHARGE_IDENTIFY_CODE.getCode());
+//                break;
+//            case "5": // 充值到表里
+//                // FEFEFEFE 68 030020310208 68 0F 19 681427 35333333333333333433343345356455B434A37735333333433A33B437353333333333333333340016
+//                // FEFEFEFE 68 050080300218 68 0F 19 35333333 33333333 34 3334 3345356455A4785469BA 33333343 D116
+//                String timeStr = ExchangeStringUtil.addZeroForNum(chargeTime, 3) + TimeUtils.dateFormat(new Date(), 17);
+//                appendStr.append(paramConfig.getMeterPass())        // 密码
+//                        .append(paramConfig.getMeterOpcode())       // 用户
+//                        .append("01")                               // 01 代表电量充值 02 代表金额充值
+//                        .append(ExchangeStringUtil.changePosition(ExchangeStringUtil.addZeroForNum(chargeTime, 4)))                             // 代表充值次数：2个字节 N2-N3
+//                        .append(ExchangeStringUtil.changePosition(timeStr))                               // 充值订单：10个字节 N4-N13
+//                        .append(ExchangeStringUtil.changePosition(ExchangeStringUtil.addZeroForNum(chargeValue, 8)));                           // 充值电量：4个字节 N14-N17
+//                int length = appendStr.length() / 2;
+//                resultStr.append("680F").append(ExchangeStringUtil.decToHex(String.valueOf(length)));
+//                break;
+//            case "6": // 写入充值次数
+//                // FEFEFEFE 68050080300218 68 14 0E 3A33B437 35333333 33333333 3333 1916
+//                appendStr.append(FourthGEnum.WRITE_CHARGE_TIME_IDENTIFY_CODE.getCode())
+//                        .append(paramConfig.getMeterPass())
+//                        .append(paramConfig.getMeterOpcode())
+//                        .append(ExchangeStringUtil.changePosition(ExchangeStringUtil.addZeroForNum(chargeTime, 4)));                             // 代表充值次数：2个字节 N2-N3
+//                resultStr.append("6814").append(ExchangeStringUtil.decToHex(String.valueOf(appendStr.length() / 2)));
+//                break;
+//            case "7":
+//                //写入限容
+//                //FEFEFEFE 68050080300218 68 14 12 43337437 33333333 33333333 33333333 33 33 1916
+//                //FEFEFEFE 68020010310208 68 14 12 6811044333B43735333333333333333363333335334333B4378016
+//                appendStr.append(FourthGEnum.WRITE_LIMIT_POWER.getCode())
+//                        .append(paramConfig.getMeterPass())
+//                        .append(paramConfig.getMeterOpcode())
+//                        .append(ExchangeStringUtil.changePosition(ExchangeStringUtil.addZeroForNum(powerValue, 8))) //300W
+//                        .append(ExchangeStringUtil.changePosition(ExchangeStringUtil.addZeroForNum(times, 2))) //跳闸后允许合闸次数
+//                        .append(ExchangeStringUtil.changePosition(ExchangeStringUtil.addZeroForNum("0", 2))); //使用次数,默认0
+//                resultStr.append("6814").append(ExchangeStringUtil.decToHex(String.valueOf(appendStr.length() / 2)));
+//                break;
+//            case "8":
+//                // 读取限容
+//                resultStr.append("681104");
+//                appendStr.append(FourthGEnum.READ_LIMIT_POWER.getCode());
+//                break;
+//            case "9":
+//                //读取软件版本
+//                resultStr.append("681104");
+//                appendStr.append(FourthGEnum.READ_SOFT_VERSION.getCode());
+//                break;
+//            case "10":
+//                //读取硬件版本
+//                resultStr.append("681104");
+//                appendStr.append(FourthGEnum.READ_HARD_VERSION.getCode());
+//                break;
+//            case "11":
+//                //读取ICCID
+//                resultStr.append("681104");
+//                appendStr.append(FourthGEnum.READ_ICCID.getCode());
+//                break;
+//            case "12":
+//                // 读取IMEI
+//                resultStr.append("681104");
+//                appendStr.append(FourthGEnum.READ_IMEI.getCode());
+//                break;
+//            case "13":
+//                // 写IP端口  --未测试
+//                String ipStr = "," + ip;
+//                String ipHex = ExchangeStringUtil.stringToHex(ipStr).replace(" ", "");
+//                int len = (ipHex.length() / 2) + 2;
+//                if (len > 99) {
+//                    throw new RuntimeException("长度异常");
+//                }
+//                String formattedLen = String.format("%02d", len);
+//                String string = formattedLen + ipStr;
+//                ipHex = ExchangeStringUtil.stringToHex(string).replace(" ", "");
+//                appendStr.append(FourthGEnum.WRITE_IP.getCode());
+//                appendStr.append(paramConfig.getMeterPass())
+//                        .append(paramConfig.getMeterOpcode())
+//                        .append(ipHex);
+//                resultStr.append("6814").append(ExchangeStringUtil.decToHex(String.valueOf(appendStr.length() / 2)));
+//                break;
+//            case "14":
+//                // 读IP端口
+//                resultStr.append("681104");
+//                appendStr.append(FourthGEnum.READ_IP.getCode());
+//                break;
+//            case "15":
+//                // 升级
+//                appendStr.append(FourthGEnum.UPGRADES.getCode())
+//                        .append(paramConfig.getMeterPass())
+//                        .append(paramConfig.getMeterOpcode())
+//                        .append("01"); //写入1
+//                resultStr.append("6814").append(ExchangeStringUtil.decToHex(String.valueOf(appendStr.length() / 2)));
+//                break;
+//            default:
+//                break;
+//        }
+//        // 加33操作
+//        resultStr.append(ExchangeStringUtil.addThree(appendStr.toString()));
+//        // 检验和
+//        String checkSum = ExchangeStringUtil.makeChecksum(resultStr.toString());
+//        resultStr.append(checkSum).append("16");
+//        resultStr.insert(0, "FEFEFEFE");
+//        log.info("发送控制指令==>{}", resultStr.toString().toUpperCase());
+//        return resultStr.toString().toUpperCase();
+//    }
+
+
+    /**
+     * 汉威4G水表保持在线指令
+     * @param meterNum
+     */
+    public static String createKeepOnline(String meterNum,String ip) {
+        StringBuilder ipCmd = getIpCmd(ip);
+        StringBuilder cmd = new StringBuilder();
+        cmd.append("C0").append("52").append("0000").append(meterNum)
+                .append("000000000000000000000000000000")
+                .append("04").append("0C").append("00")
+                .append(ipCmd)
+                .append("0500")
+                .append("50").append("00");
+        String time = sdf.format(new Date());
+        //yyyy-MM-dd HH:mm:ss
+        cmd.append(time.substring(11,13)).append(time.substring(14,16))
+                .append(time.substring(17,19));
+        String checksum = ExchangeStringUtil.makeChecksum(cmd.substring(2));
+        cmd.append(checksum).append("D0");
+
+        return cmd.toString();
+    }
+
+    /**
+     * 汉威4G水表下线指令
+     * @param meterNum
+     * @return
+     */
+    public static String createOffline(String meterNum,String ip) {
+        StringBuilder ipCmd = getIpCmd(ip);
+        StringBuilder cmd = new StringBuilder();
+        cmd.append("C0").append("52").append("0000").append(meterNum)
+                .append("000000000000000000000000000000")
+                .append("04").append("0C").append("00")
+                .append(ipCmd)
+                .append("0500")
+                .append("50").append("80");
+        String time = sdf.format(new Date());
+        //yyyy-MM-dd HH:mm:ss
+        cmd.append(time.substring(11,13)).append(time.substring(14,16))
+                .append(time.substring(17,19));
+        String checksum = ExchangeStringUtil.makeChecksum(cmd.substring(2));
+        cmd.append(checksum).append("D0");
+
+        return cmd.toString();
+    }
+
+    /**
+     * 汉威4G冷水表控制阀门指令
+     * @param meterNum
+     * @param valve
+     * @return
+     */
+    public static String createControlValve(String meterNum,String valve,String ip){
+        StringBuilder ipCmd = getIpCmd(ip);
+
+        StringBuilder cmd = new StringBuilder();
+        cmd.append("C0").append("57").append("0000").append(meterNum)
+                .append("000000000000000000000000000000")
+                .append("03").append("0C").append("00")
+                .append(ipCmd)
+                //.append("7B154C177E1A") //IP+端口
+                .append("0900")
+                .append("50").append("00").append("10").append("3000")
+                .append("0200");
+        if (valve.equals("1")){
+            cmd.append("0200"); //关阀
+        }else if (valve.equals("0")){
+            cmd.append("0100"); //开阀
+        }else {
+            throw new RuntimeException("阀门参数不正确");
+        }
+
+        String checksum = ExchangeStringUtil.makeChecksum(cmd.substring(2));
+        cmd.append(checksum).append("D0");
+        return cmd.toString();
+    }
+
+    /**
+     * 解析获取到16进制的IP地址
+     * @param ip
+     * @return
+     */
+    private static StringBuilder getIpCmd(String ip) {
+        String[] split = ip.split(":");
+        if (split.length != 2){
+            throw new RuntimeException("目标源地址格式不正确");
+        }
+        String host = split[0];
+        String port = split[1];
+        String[] hostArray = host.split("\\.");
+        if (hostArray.length != 4){
+            throw new RuntimeException("IP格式不正确");
+        }
+        StringBuilder ipCmd = new StringBuilder();
+        for (String s : hostArray) {
+            ipCmd.append(ExchangeStringUtil.decToHex(s));
+        }
+        ipCmd.append(ExchangeStringUtil.decToHex(port));
+        return ipCmd;
+    }
+}

user-service/src/main/java/com/mh/user/utils/NettyTools.java

@@ -0,0 +1,77 @@
+package com.mh.user.utils;
+
+
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.Objects;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author LJF
+ * @version 1.0
+ * @project TAD_Server
+ * @description 缓存等待数据
+ * @date 2023/7/4 08:45:16
+ */
+@Slf4j
+public class NettyTools {
+
+    /**
+     * 响应消息缓存
+     */
+    private static final Cache<String, BlockingQueue<String>> responseMsgCache = CacheBuilder.newBuilder()
+            .maximumSize(50000)
+            .expireAfterWrite(1000, TimeUnit.SECONDS)
+            .build();
+
+
+    /**
+     * 等待响应消息
+     * @param key 消息唯一标识
+     * @return ReceiveDdcMsgVo
+     */
+    public static boolean waitReceiveMsg(String key) {
+
+        try {
+            //设置超时时间
+            String vo = Objects.requireNonNull(responseMsgCache.getIfPresent(key))
+                .poll(1000 * 10, TimeUnit.MILLISECONDS);
+
+            //删除key
+            responseMsgCache.invalidate(key);
+            return StringUtils.isNotBlank(vo);
+        } catch (Exception e) {
+            log.error("获取数据异常,sn={},msg=null",key);
+            return false;
+        }
+
+    }
+
+    /**
+     * 初始化响应消息的队列
+     * @param key 消息唯一标识
+     */
+    public static void initReceiveMsg(String key) {
+        responseMsgCache.put(key,new LinkedBlockingQueue<String>(1));
+    }
+
+    /**
+     * 设置响应消息
+     * @param key 消息唯一标识
+     */
+    public static void setReceiveMsg(String key, String msg) {
+
+        if(responseMsgCache.getIfPresent(key) != null){
+            responseMsgCache.getIfPresent(key).add(msg);
+            return;
+        }
+
+        log.warn("sn {}不存在",key);
+    }
+
+}

user-service/src/main/java/com/mh/user/utils/TimeUtils.java

@@ -0,0 +1,85 @@
+package com.mh.user.utils;
+
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Calendar;
+import java.util.Date;
+
+/**
+ * @author 铭汉科技——LJL
+ * @date 2020-10-16 09:44
+ * @Description
+ */
+public class TimeUtils {
+
+    /**
+     * 线程安全的 DateTimeFormatter
+     * 推荐使用，因为该类是不可变的，并且是线程安全的
+     */
+    //SimpleDateFormat对象
+    private static final SimpleDateFormat simpleDateFormat = new SimpleDateFormat("ssmmHHddMMyy");
+
+    private static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+
+    public static void main(String[] args) {
+        System.out.println(limitDateStr());
+    }
+
+    public static String date2String() {
+        synchronized (sdf) {
+            return sdf.format(new Date());
+        }
+    }
+    static String UTCToCST(String UTCStr, String format) throws ParseException {
+        Date date = null;
+        SimpleDateFormat sdf = new SimpleDateFormat(format);
+        UTCStr = UTCStr.substring(0, 4) + "-" + UTCStr.substring(4,6) + "-" + UTCStr.substring(6,7)
+                + UTCStr.substring(7,11) + ":" + UTCStr.substring(11,13) + ":" + UTCStr.substring(13,UTCStr.length());
+        System.out.println(UTCStr);
+        date = sdf.parse(UTCStr);
+//        System.out.println("UTC时间: " + date);
+        Calendar calendar = Calendar.getInstance();
+        calendar.setTime(date);
+        calendar.set(Calendar.HOUR, calendar.get(Calendar.HOUR) + 8);
+        //calendar.getTime() 返回的是Date类型，也可以使用calendar.getTimeInMillis()获取时间戳
+//        System.out.println("北京时间: " + calendar.getTime());
+        SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+        String dateString = formatter.format(calendar.getTime());
+//        System.out.println(dateString);
+        return dateString;
+    }
+
+    public static String limitDateStr() {
+        try {
+            synchronized (simpleDateFormat) {
+                // 指令五秒内有效
+                Date date = new Date(new Date().getTime() + 5);
+                return simpleDateFormat.format(date);
+            }
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public static String dateFormat(Date time, Integer timeTypeLen) {
+        String returnStr = "";
+        switch (timeTypeLen) {
+            case 17:
+                returnStr = cn.hutool.core.date.DateUtil.format(time, "yyyyMMddHHmmssSSS");
+                break;
+            case 10:
+                returnStr = cn.hutool.core.date.DateUtil.format(time, "yyyy-MM-dd");
+                break;
+            case 7:
+                returnStr = cn.hutool.core.date.DateUtil.format(time, "yyyy-MM");
+                break;
+            case 4:
+                returnStr = cn.hutool.core.date.DateUtil.format(time, "yyyy");
+                break;
+            default:
+                break;
+        }
+        return returnStr;
+    }
+
+}

user-service/src/main/resources/application-dev.yml

@@ -1,5 +1,5 @@
 server:
-  port: 8762 #8764新珠江,8766广合科技,8767广大科技,8768珠江国际,8769保利山庄,8770东莞迎宾馆
+  port: 8766 #8764新珠江,8766广合科技,8767广大科技,8768珠江国际,8769保利山庄,8770东莞迎宾馆
 spring:
   application:
     name: jnd-user-service
@@ -118,6 +118,14 @@ wechat:
   pushUrl: https://api.weixin.qq.com/cgi-bin/message/template/send?access_token=ACCESS_TOKEN
   baseUrl: https://35gm72cu2458.vicp.fun
 
+device:
+  meter:
+    #    password: "02080605"
+    #    opcode: "00000000"
+    password: "02000000"
+    opcode: "00000000"
+  port: 20001
+
 
 
 

user-service/src/main/resources/application-prod.yml

@@ -117,3 +117,9 @@ wechat:
   # 消息推送url
   pushUrl: https://api.weixin.qq.com/cgi-bin/message/template/send?access_token=ACCESS_TOKEN
   baseUrl: http://jnd2.mhwsh.net:8766
+
+device:
+  meter:
+    password: "02000000"
+    opcode: "00000000"
+  port: 20001

user-service/src/main/resources/application-test.yml

@@ -117,3 +117,11 @@ wechat:
   # 消息推送url
   pushUrl: https://api.weixin.qq.com/cgi-bin/message/template/send?access_token=ACCESS_TOKEN
   baseUrl: https://35gm72cu2458.vicp.fun
+
+device:
+  meter:
+    #    password: "02080605"
+    #    opcode: "00000000"
+    password: "02000000"
+    opcode: "00000000"
+  port: 20001