Netty实现数据上下行

Netty实现数据上下行

使用LVS+NGinx+Netty实现数据接入
在数据上行的时候，通过使用车辆唯一标识码（vin）和连接通道绑定

Netty一些配置参数如下:

#netty项目使用的端口
server.port=8017
#使用启用epoll（在Linux上拥有更好的传输性能）
netty.server.use-epoll=true
#netty项目监听的两个端口
netty.server.port1=8035
netty.server.port2=8055# 使用plunlog 查看运行日志，需要和plumelog服务端的redis配置一致
plumelog.appName=netty
plumelog.redisHost=xxx.xx.xxx.xx:xxxx
plumelog.redisAuth=
plumelog.redisDb=0#Spring boot admin Client
management.endpoints.web.exposure.include=*:
#这个配置项用于控制哪些Actuator端点应该通过Web接口暴露。*表示暴露所有端点。这包括健康检查、环境信息、度量信息等。在生产环境中，你可能需要根据需要暴露特定的端点，而不是全部暴露，以提高安全性。
management.endpoint.health.show-details=always:
#这个配置项用于控制健康端点（/actuator/health）返回的细节级别。always表示总是显示详细的健康信息，包括所有子系统的健康状态。其他选项包括never（不显示任何细节）和when-authorized（在授权用户请求时显示细节）。这有助于监控和诊断应用的健康状态。
management.health.redis.enabled=false:
#这个配置项用于控制是否将Redis的健康检查包含在应用的整体健康检查中。设置为false表示不将Redis的健康状态作为应用健康状态的一部分。这在你不想或不需要监控Redis服务状态时很有用。
management.server.port=8109:
#这个配置项用于设置管理端点的端口。默认情况下，管理端点（Actuator端点）与主应用共享相同的端口。但你可以通过此配置将它们分离到不同的端口上，以便更好地管理和监控。在这个例子中，管理端点被配置为监听8109端口。
spring.boot.admin.client.url=http://xxx.xx.xxx.xx:xxxx:
#这个配置项用于指定Spring Boot Admin服务器的URL。Spring Boot Admin是一个用于管理和监控Spring Boot应用的工具。通过此配置，你的Spring Boot应用可以注册到Spring Boot Admin服务器，从而允许你通过Web界面监控和管理应用。你需要将xxx.xx.xxx.xx:xxxx替换为实际的Spring Boot Admin服务器地址和端口。
spring.boot.admin.client.instance.prefer-ip=true:
这个配置项用于控制在Spring Boot Admin中注册应用实例时，是否优先使用IP地址而不是主机名。设置为true表示使用IP地址。这有助于在DNS解析可能存在问题或延迟时，确保Spring Boot Admin能够准确地识别和管理应用实例。

1、Handler

Netty的Handler 提供了一系列对数据的处理方法，与登录handle与数据处理handle分开。

在登陆成功后移除登陆handle

            if ("01".equals(logMsg) || "C2".equals(logMsg)) {boolean loginCheck = redisTemplateNew.exists(vinAsc);StringBuilder loginMsg = new StringBuilder(decryptData);if (!loginCheck) {//上报时间;发送登陆失败的kafka消息String dateSend = getDate16();kafkaTemplate.send("sendFailNS", kafkaKey, dateSend + decryptData);log.error(vinAsc + " loginFail");} else {//登陆成功标识loginMsg.replace(6, 8, "01");//把vin码做绑定channelActive(ctx);String clientId = vinAsc.trim();ChannelStore.bind(ctx, clientId);           }//加密回复报文String encryptData = commonUntils1.getEncryptRSA(loginMsg, vinAsc);byte[] odls = ByteBufUtil.decodeHexDump(encryptData);ctx.writeAndFlush(Unpooled.wrappedBuffer(odls));if (!loginCheck) {//关闭连接  `channelInactive(ctx);ctx.channel().close();} else {//移除鉴权ctx.pipeline().remove(this);}//上报时间String dateSend = getDate16();if ("01".equals(logMsg)) {kafkaTemplate.send("sendLAFS", kafkaKey, dateSend + decryptData);} else if ("C2".equals(logMsg)) {kafkaTemplate.send("sendLAFS", kafkaKey, dateSend + decryptData);}}

2、调用下行API

因为Netty服务器是集群，会把客户端的唯一标识符和连接服务器的地址存在数据库里，在平台调用下行命令时，先找到netty连接的服务器。

    /*** 发送消息的控制器方法。** @param downVO 包含消息详情的请求体对象，包括VIN（车辆识别码）、类型（设防、解防、启动指令等）和流水号。* @return 返回一个响应实体，包含消息发送的状态码和描述。* @throws Exception 如果校验失败或发送消息时遇到异常，则抛出异常。*/@PostMapping(value = "/sendMsg")public ResponseEntity<String> sendMSg(@RequestBody DownVO downVO) throws Exception {// 校验VINif (StringUtils.isBlank(downVO.getVin())) {ApiResponse apiResponse = new ApiResponse(400, "VIN参数不合法");return new ResponseEntity(apiResponse, HttpStatus.BAD_REQUEST);}// 校验类型if (StringUtils.isBlank(downVO.getType()) || !"1".equals(downVO.getType()) && !"2".equals(downVO.getType()) && !"3".equals(downVO.getType())) {ApiResponse apiResponse = new ApiResponse(400, "远控类型参数不合法");return new ResponseEntity(apiResponse, HttpStatus.BAD_REQUEST);}// 校验流水号if (StringUtils.isBlank(downVO.getSerialNumber()) || downVO.getSerialNumber().length() != 16) {ApiResponse apiResponse = new ApiResponse(400, "流水号参数不合法");return new ResponseEntity(apiResponse, HttpStatus.BAD_REQUEST);}try {log.info(downVO.getVin() +" 下行命令：" + downVO.getVin() + ", " + downVO.getType() + ", " + downVO.getSerialNumber());String remoteMsg = getRemoteData(downVO.getVin(), downVO.getType(), downVO.getSerialNumber());Channel channel = ChannelStore.getChannel(downVO.getVin());log.info(downVO.getVin() + " 下行通道：" + channel + "  下行组装消息：" + remoteMsg);// 连接状态if ((channel == null) || !channel.isActive()) {log.error(downVO.getVin() +" 555连接已断开：" + redisTemplateNew.get(downVO.getVin()+"_IP"));redisTemplateNew.delete(downVO.getVin()+"_IP");ApiResponse apiResponse = new ApiResponse(555, "连接已断开");return new ResponseEntity(apiResponse, HttpStatus.BAD_REQUEST);}log.error( downVO.getVin() +" 消息发送成功：" +   downVO.getType() + ", " + downVO.getSerialNumber());byte[] odls = ByteBufUtil.decodeHexDump(remoteMsg);channel.writeAndFlush(Unpooled.wrappedBuffer(odls));ApiResponse apiResponse = new ApiResponse(200, "消息发送成功");return new ResponseEntity(apiResponse, HttpStatus.OK);} catch (Exception e) {// 处理异常，返回错误信息String errorMessage = "消息发送失败: " + e.getMessage();log.error(downVO.getVin() +" 消息发送失败：" +  e.getMessage());return new ResponseEntity(new ApiResponse(500, errorMessage), HttpStatus.INTERNAL_SERVER_ERROR);}}

3、任务监控

日志使用pumelog查看

<?xml version="1.0" encoding="UTF-8"?><configuration scan="true" scanPeriod="60 seconds" debug="false">    <!-- 日志存放路径 -->   <property name="log.path" value="logs/bi-veos" />   <!-- 日志输出格式 -->   <property name="log.pattern" value="%d{HH:mm:ss.SSS} [%X{traceId}][%thread] %-5level %logger{20} - [%method,%line] - %msg%n" />    <!-- 控制台输出 -->   <appender name="console" class="ch.qos.logback.core.ConsoleAppender">      <encoder>         <pattern>${log.pattern}</pattern>      </encoder>   </appender>    <!-- 系统日志输出 -->   <appender name="file_info" class="ch.qos.logback.core.rolling.RollingFileAppender">       <file>${log.path}/info.log</file>        <!-- 循环政策：基于时间创建日志文件 -->      <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">            <!-- 日志文件名格式 -->         <fileNamePattern>${log.path}/info.%d{yyyy-MM-dd}.log</fileNamePattern>         <!-- 日志最大的历史 30天 -->         <maxHistory>30</maxHistory>      </rollingPolicy>      <encoder>         <pattern>${log.pattern}</pattern>      </encoder>      <filter class="ch.qos.logback.classic.filter.LevelFilter">            <!-- 过滤的级别 -->            <level>INFO</level>            <!-- 匹配时的操作：接收（记录） -->            <onMatch>ACCEPT</onMatch>            <!-- 不匹配时的操作：拒绝（不记录） -->            <onMismatch>DENY</onMismatch>        </filter>   </appender>    <appender name="file_error" class="ch.qos.logback.core.rolling.RollingFileAppender">       <file>${log.path}/error.log</file>        <!-- 循环政策：基于时间创建日志文件 -->        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">            <!-- 日志文件名格式 -->            <fileNamePattern>${log.path}/error.%d{yyyy-MM-dd}.log</fileNamePattern>         <!-- 日志最大的历史 30天 -->         <maxHistory>30</maxHistory>        </rollingPolicy>        <encoder>            <pattern>${log.pattern}</pattern>        </encoder>        <filter class="ch.qos.logback.classic.filter.LevelFilter">            <!-- 过滤的级别 -->            <level>ERROR</level>         <!-- 匹配时的操作：接收（记录） -->            <onMatch>ACCEPT</onMatch>         <!-- 不匹配时的操作：拒绝（不记录） -->            <onMismatch>DENY</onMismatch>        </filter>    </appender>    <!-- 环境配置 -->    <springProperty scope="context" name="plumelog.appName" source="plumelog.appName"/>    <springProperty scope="context" name="plumelog.redisHost" source="plumelog.redisHost"/>    <springProperty scope="context" name="plumelog.redisAuth" source="plumelog.redisAuth"/>    <springProperty scope="context" name="plumelog.redisDb" source="plumelog.redisDb"/>    <springProperty scope="context" name="plumelog.env" source="spring.profiles.active"/>    <!-- plumelog -->    <appender name="plumelog" class="com.plumelog.logback.appender.RedisAppender">        <appName>${plumelog.appName}</appName>        <redisHost>${plumelog.redisHost}</redisHost>        <redisAuth>${plumelog.redisAuth}</redisAuth>        <redisDb>${plumelog.redisDb}</redisDb>        <env>${plumelog.env}</env>    </appender><!--    &lt;!&ndash; 系统模块日志级别控制  &ndash;&gt;--><!--   <logger name="com.chery.bi.modules.veos" level="warn" />--><!--   &lt;!&ndash; Spring日志级别控制  &ndash;&gt;--><!--   <logger name="org.springframework" level="warn" />-->    <!--系统操作日志-->    <root level="info">        <appender-ref ref="console" />        <appender-ref ref="file_info" />        <appender-ref ref="file_error" />        <appender-ref ref="plumelog"/>    </root></configuration>

任务监控 使用springbootadmin查看

只需要在pom中引入依赖，配置如开始所示

   <dependency><groupId>de.codecentric</groupId><artifactId>spring-boot-admin-starter-client</artifactId><version>2.7.3</version></dependency><!-- actuator start--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId><exclusions><exclusion><artifactId>logback-classic</artifactId><groupId>ch.qos.logback</groupId></exclusion></exclusions></dependency>

新建spring boot server项目 并配置邮件告警

Spring Boot应用名称和端口
spring.application.name=nettyAdminServer：设置Spring Boot应用的名称为nettyAdminServer，这个名称在Spring Boot Admin或其他服务发现机制中很重要，用于识别和区分不同的服务。
server.port=8110：设置应用监听的端口号为8110。
健康检查细节
management.endpoint.health.show-details=always：配置健康检查端点以总是显示详细信息。这有助于在监控和诊断时获取更多关于应用健康状态的信息。
邮件配置
spring.mail.smtp.auth=true：启用SMTP认证。
spring.mail.properties.mail.smtp.ssl.trust=${spring.mail.host}：设置SMTP SSL/TLS连接的信任主机为spring.mail.host配置的值，即smtp.feishu.cn。这有助于建立安全的邮件连接。
spring.mail.properties.mail.smtp.ssl.enable=true：启用SMTP连接的SSL/TLS加密。
spring.mail.host=smtp..cn：设置SMTP服务器的主机名为smtp..cn，即SMTP服务器。
spring.mail.username和spring.mail.password：分别设置SMTP服务器的登录用户名和密码，用于认证。
spring.mail.port=465：设置SMTP服务器的端口号为465，这是SMTPS（SMTP over SSL）的标准端口。
Spring Boot Admin邮件通知
spring.boot.admin.notify.mail.to和spring.boot.admin.notify.mail.from：分别设置Spring Boot Admin发送邮件通知的收件人和发件人地址
Spring Boot Admin忽略的变更
spring.boot.admin.notify.discord.ignore-changes={"UNKNOWN:UP","OFFLINE:UP"}：这个配置项实际上与Discord通知相关，但配置中提到了spring.boot.admin.notify.discord，而你的配置环境看起来并没有启用Discord通知。不过，这里的意思是配置Spring Boot Admin在哪些状态变更时不发送Discord通知，比如从UNKNOWN到UP或从OFFLINE到UP

spring.application.name=nettyAdminServer
server.port=8110
management.endpoint.health.show-details=alwaysspring.mail.smtp.auth=true
spring.mail.properties.mail.smtp.ssl.trust=${spring.mail.host}
spring.mail.properties.mail.smtp.ssl.enable=true
spring.mail.host=smtp.cn
spring.mail.username=
spring.mail.password=
spring.mail.port=465
spring.boot.admin.notify.mail.to=
spring.boot.admin.notify.mail.from=
spring.boot.admin.notify.discord.ignore-changes={"UNKNOWN:UP","OFFLINE:UP"}

Spring bootadmin监控界面