commit 02e9bacb88069939f728113a01d82be49ae41657 Author: qiminbao Date: Wed Aug 20 16:50:00 2025 +0800 初始化 diff --git a/.mvn/wrapper/maven-wrapper.properties b/.mvn/wrapper/maven-wrapper.properties new file mode 100644 index 0000000..12fbe1e --- /dev/null +++ b/.mvn/wrapper/maven-wrapper.properties @@ -0,0 +1,19 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +wrapperVersion=3.3.2 +distributionType=only-script +distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.11/apache-maven-3.9.11-bin.zip diff --git a/pom.xml b/pom.xml new file mode 100644 index 0000000..d2ed828 --- /dev/null +++ b/pom.xml @@ -0,0 +1,135 @@ + + + 4.0.0 + + org.springframework.boot + spring-boot-starter-parent + 2.5.6 + + + com.tongran + agent-tcp + 0.0.1-SNAPSHOT + agent-tcp + agent-tcp + + 1.8 + + + + org.springframework.boot + spring-boot-starter-web + + + + org.projectlombok + lombok + true + + + org.springframework.boot + spring-boot-starter-test + test + + + + org.springframework.boot + spring-boot-starter-reactor-netty + + + + cn.hutool + hutool-all + 5.7.11 + + + + org.apache.commons + commons-lang3 + + + + org.springdoc + springdoc-openapi-ui + 1.5.5 + + + + org.springframework.boot + spring-boot-configuration-processor + + + + org.springframework.boot + spring-boot-starter-aop + + + + org.springframework.boot + spring-boot-starter-validation + + + + org.openpnp + opencv + 4.5.1-2 + + + + com.github.oshi + oshi-core + 6.4.4 + + + + org.fusesource + sigar + 1.6.4 + + + + com.github.docker-java + docker-java + 3.3.0 + + + + org.snmp4j + snmp4j + 3.7.0 + + + + com.alibaba.fastjson2 + fastjson2 + 2.0.31 + + + + org.apache.rocketmq + rocketmq-spring-boot-starter + 2.2.3 + + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + org.apache.maven.plugins + maven-surefire-plugin + ${maven-surefire-plugin.version} + + true + + + + + + diff --git a/src/main/java/com/tongran/agenttcp/AgentTcpApplication.java b/src/main/java/com/tongran/agenttcp/AgentTcpApplication.java new file mode 100644 index 0000000..d34c67d --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/AgentTcpApplication.java @@ -0,0 +1,13 @@ +package com.tongran.agenttcp; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class AgentTcpApplication { + + public static void main(String[] args) { + SpringApplication.run(AgentTcpApplication.class, args); + } + +} diff --git a/src/main/java/com/tongran/agenttcp/exception/ServerException.java b/src/main/java/com/tongran/agenttcp/exception/ServerException.java new file mode 100644 index 0000000..3f7f464 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/exception/ServerException.java @@ -0,0 +1,27 @@ +package com.tongran.agenttcp.exception; + +import cn.hutool.core.util.StrUtil; +import com.tongran.agenttcp.exception.base.BaseException; +import com.tongran.agenttcp.exception.code.ErrorCode; + +public class ServerException extends BaseException { + + private static final long serialVersionUID = -519977195881081739L; + + public ServerException(ErrorCode code) { + super(code.getCode(), code.getMsg()); + } + + public ServerException(Integer code, String msg) { + super(code, msg); + } + + public ServerException(String msg) { + super(500, msg); + } + + public ServerException(String msg, Object... arguments) { + super(500, StrUtil.format(msg, arguments)); + } + +} diff --git a/src/main/java/com/tongran/agenttcp/exception/base/BaseException.java b/src/main/java/com/tongran/agenttcp/exception/base/BaseException.java new file mode 100644 index 0000000..ca1fee3 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/exception/base/BaseException.java @@ -0,0 +1,42 @@ +package com.tongran.agenttcp.exception.base; + + +import com.tongran.agenttcp.exception.code.ErrorCode; + +public class BaseException extends RuntimeException implements ErrorCode { + + private static final long serialVersionUID = 1966249840643379123L; + + private Integer code; + + private String msg; + + public BaseException() { + } + + public BaseException(String msg, Throwable cause) { + super(msg, cause); + } + + public BaseException(Integer code, String msg) { + super(msg); + this.code = code; + this.msg = msg; + } + + public BaseException(Integer code, String msg, Throwable cause) { + super(msg, cause); + this.code = code; + this.msg = msg; + } + + @Override + public Integer getCode() { + return code; + } + + @Override + public String getMsg() { + return msg; + } +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agenttcp/exception/code/ErrorCode.java b/src/main/java/com/tongran/agenttcp/exception/code/ErrorCode.java new file mode 100644 index 0000000..98127fe --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/exception/code/ErrorCode.java @@ -0,0 +1,16 @@ +package com.tongran.agenttcp.exception.code; + + +import com.tongran.agenttcp.utils.R; + +public interface ErrorCode { + + Integer getCode(); + + String getMsg(); + + default R toResult() { + return R.error(getMsg(), getCode()); + } + +} diff --git a/src/main/java/com/tongran/agenttcp/exception/code/GlobalErrorCode.java b/src/main/java/com/tongran/agenttcp/exception/code/GlobalErrorCode.java new file mode 100644 index 0000000..3f3e520 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/exception/code/GlobalErrorCode.java @@ -0,0 +1,20 @@ +package com.tongran.agenttcp.exception.code; + +import lombok.AllArgsConstructor; +import lombok.Getter; + +@Getter +@AllArgsConstructor +public enum GlobalErrorCode implements ErrorCode { + + SUCCESS(200, "成功"), + + NOT_FOUND(404, "未找到相关资源"), + + ERROR(500, "系统错误"); + + private final Integer code; + + private final String msg; + +} diff --git a/src/main/java/com/tongran/agenttcp/rocketmq/AgentProducer.java b/src/main/java/com/tongran/agenttcp/rocketmq/AgentProducer.java new file mode 100644 index 0000000..5eae646 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/rocketmq/AgentProducer.java @@ -0,0 +1,55 @@ +package com.tongran.agenttcp.rocketmq; + + +import com.tongran.agenttcp.rocketmq.core.RocketMqService; +import com.tongran.agenttcp.rocketmq.core.model.MqMsg; +import com.tongran.agenttcp.utils.AssertLog; +import org.apache.rocketmq.client.producer.SendCallback; +import org.apache.rocketmq.client.producer.SendResult; +import org.apache.rocketmq.spring.core.RocketMQTemplate; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.stereotype.Component; + +import javax.annotation.Resource; + +@Component +public class AgentProducer implements RocketMqService { + + @Resource + private RocketMQTemplate rocketMQTemplate; + + @Override + public void send(MqMsg msg) { + AssertLog.info("send发送消息:==>{}", msg); + rocketMQTemplate.send(msg.getTopic(), MessageBuilder.withPayload(msg.getContent()).build()); + } + + @Override + public void asyncSend(MqMsg msg) { + AssertLog.info("asyncSend发送消息:==>{}", msg); + rocketMQTemplate.asyncSend(msg.getTopic(), msg.getContent(), new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { +// AssertLog.info("asyncSend发送消息成功:==>{}", sendResult.getSendStatus()); + } + + @Override + public void onException(Throwable throwable) { + AssertLog.error("asyncSend发送消息失败:==>{}", throwable.getMessage()); + } + }); + } + + @Override + public void syncSendOrderly(MqMsg msg) { + AssertLog.info("syncSendOrderly发送消息:==>{}", msg); + rocketMQTemplate.sendOneWay(msg.getTopic(), msg.getContent()); + } + + @Override + public void delayedSendOrderly(MqMsg msg) { + AssertLog.info("delayedSendOrderly发送消息:==>{}", msg); + rocketMQTemplate.syncSend(msg.getTopic(), MessageBuilder.withPayload(msg.getContent()).build(), 30000, msg.getLevel()); + } + +} diff --git a/src/main/java/com/tongran/agenttcp/rocketmq/config/AgentMqConfig.java b/src/main/java/com/tongran/agenttcp/rocketmq/config/AgentMqConfig.java new file mode 100644 index 0000000..03d91c6 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/rocketmq/config/AgentMqConfig.java @@ -0,0 +1,14 @@ +package com.tongran.agenttcp.rocketmq.config; + +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.context.annotation.Configuration; + +@Data +@Configuration +@ConfigurationProperties(prefix = "rocketmq.producer") +public class AgentMqConfig { + + private String agentTopic; + +} diff --git a/src/main/java/com/tongran/agenttcp/rocketmq/core/RocketMqService.java b/src/main/java/com/tongran/agenttcp/rocketmq/core/RocketMqService.java new file mode 100644 index 0000000..25c9faf --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/rocketmq/core/RocketMqService.java @@ -0,0 +1,45 @@ +package com.tongran.agenttcp.rocketmq.core; + + +import com.tongran.agenttcp.rocketmq.core.model.MqMsg; + +/** + * Rocket MQ 对应服务封装 + * @author BAO + * + */ +public interface RocketMqService { + + /** + * 同步发送消息
+ *

+ * 当发送的消息很重要是，且对响应时间不敏感的时候采用sync方式; + * + * @param msg 发送消息实体类 + */ + void send(MqMsg msg); + + /** + * 异步发送消息，异步返回消息结果 当发送的消息很重要，且对响应时间非常敏感的时候采用async方式； + * + * @param msg 发送消息实体类 + */ + void asyncSend(MqMsg msg); + + /** + * 直接发送发送消息，不关心返回结果，容易消息丢失，适合日志收集、不精确统计等消息发送; 发送的消息不重要时，采用one-way方式，以提高吞吐量； + * + * @param msg 发送消息实体类 + */ + void syncSendOrderly(MqMsg msg); + + /** + * 发送延时消息，不关心返回结果
+ *

+ * 当发送的消息不重要时，采用one-way方式，以提高吞吐量； + * + * @param msg 发送消息实体类 + */ + default void delayedSendOrderly(MqMsg msg){}; + +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agenttcp/rocketmq/core/model/MqMsg.java b/src/main/java/com/tongran/agenttcp/rocketmq/core/model/MqMsg.java new file mode 100644 index 0000000..21d2eb3 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/rocketmq/core/model/MqMsg.java @@ -0,0 +1,38 @@ +package com.tongran.agenttcp.rocketmq.core.model; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.io.Serializable; + +/** + * RocketMQ消息发送对象 + * + * @author BAO + */ +@Data +@Builder +@AllArgsConstructor +@NoArgsConstructor +public class MqMsg implements Serializable { + + private static final long serialVersionUID = 4164379745748817325L; + + /** + * 消息topic + */ + private String topic; + + /** + * 消息content + */ + private Object content; + + /** + * 消息延时等级 + */ + private Integer level; + +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agenttcp/rocketmq/listener/AgentConsumerDownListener.java b/src/main/java/com/tongran/agenttcp/rocketmq/listener/AgentConsumerDownListener.java new file mode 100644 index 0000000..800ceab --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/rocketmq/listener/AgentConsumerDownListener.java @@ -0,0 +1,52 @@ +package com.tongran.agenttcp.rocketmq.listener; + +import cn.hutool.core.util.ObjectUtil; +import com.alibaba.fastjson2.JSON; +import com.tongran.agenttcp.server.annotation.AgentDispatcher; +import com.tongran.agenttcp.server.basics.AgentDispatcherManager; +import com.tongran.agenttcp.server.basics.AgentMsgHandler; +import com.tongran.agenttcp.server.model.Message; +import com.tongran.agenttcp.server.netty.core.session.SessionManager; +import com.tongran.agenttcp.utils.AssertLog; +import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; +import org.apache.rocketmq.spring.core.RocketMQListener; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.stereotype.Component; + +import javax.annotation.Resource; +import java.util.Objects; + +@Component +@ConditionalOnProperty(name = "rocketmq.enabled", havingValue = "true") +@RocketMQMessageListener(topic = "agent_down", consumerGroup = "agent_down_group") +public class AgentConsumerDownListener implements RocketMQListener { + + protected final SessionManager sessionManager; + + @Resource + private AgentDispatcherManager agentDispatcherManager; + + public AgentConsumerDownListener() { + this.sessionManager = SessionManager.getInstance(); + } + + @Override + public void onMessage(String message) { + AssertLog.info("consumer==> received down message: {}", message); + try { + Message dto = JSON.parseObject(message, Message.class); + AgentMsgHandler msgHandler = agentDispatcherManager.getHandler(dto.getDataType() + "&" + AgentDispatcher.VersionEnum.V1.value); + if (ObjectUtil.isNotEmpty(msgHandler)) { + String msg = msgHandler.downHandle(dto.getData()); + Message chargeMessage = Message.builder().clientId(dto.getClientId()).data(msg).build(); + if (Objects.nonNull(sessionManager.getSessionById(dto.getClientId()))) { + sessionManager.writeAndFlush(sessionManager.getSessionById(dto.getClientId()).getChannel(), chargeMessage); + } + } + } catch (Exception e) { + AssertLog.error("=====rocketmq-exception{}=====" + e.getMessage()); + } + + } + +} diff --git a/src/main/java/com/tongran/agenttcp/server/AgentNettyConfig.java b/src/main/java/com/tongran/agenttcp/server/AgentNettyConfig.java new file mode 100644 index 0000000..5a8c997 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/server/AgentNettyConfig.java @@ -0,0 +1,22 @@ +package com.tongran.agenttcp.server; + +import com.tongran.agenttcp.server.netty.config.BaseNettyConfig; +import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.ToString; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.context.annotation.Configuration; + +/** + * Netty配置 + */ +@Data +@Configuration +@ToString(callSuper = true) +@EqualsAndHashCode(callSuper = true) +@ConfigurationProperties(prefix = "tcp.netty.charge") +public class AgentNettyConfig extends BaseNettyConfig { + + private static final long serialVersionUID = -4284214721383107291L; + +} diff --git a/src/main/java/com/tongran/agenttcp/server/AgentNettyServer.java b/src/main/java/com/tongran/agenttcp/server/AgentNettyServer.java new file mode 100644 index 0000000..e0a2892 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/server/AgentNettyServer.java @@ -0,0 +1,58 @@ +package com.tongran.agenttcp.server; + + +import com.tongran.agenttcp.server.handler.AgentDecoderHandler; +import com.tongran.agenttcp.server.handler.AgentDispatcherHandler; +import com.tongran.agenttcp.server.handler.AgentEncoderHandler; +import com.tongran.agenttcp.server.netty.BaseNettyServer; +import com.tongran.agenttcp.server.netty.core.handler.TCPListenHandler; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.handler.timeout.IdleStateHandler; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import javax.annotation.Resource; + +@Configuration +public class AgentNettyServer extends BaseNettyServer { + + @Resource + private AgentNettyConfig config; + + @Resource + private TCPListenHandler tcpListenHandler; + + @Resource + private AgentDecoderHandler decoderHandler; + + @Resource + private AgentEncoderHandler encoderHandler; + + @Resource + private AgentDispatcherHandler dispatcherHandler; + + + protected AgentNettyServer(AgentNettyConfig config) { + super(config); + } + + @Bean(name = "NettyCharge", initMethod = "start", destroyMethod = "stop") + BaseNettyServer run() { + config.setHander(new ChannelInitializer() { + @Override + public void initChannel(NioSocketChannel ch) throws Exception { + ch.pipeline().addLast(new IdleStateHandler(config.readerIdleTime, config.writerIdleTime, config.allIdleTime));// 心跳 + ch.pipeline().addLast(tcpListenHandler);// 监听器 + //入栈 + ch.pipeline().addLast(decoderHandler);//解码器 + //出栈 + ch.pipeline().addLast(encoderHandler);//加码器 + // 业务分发是入站最后一个处理器, 出站第一个处理器, 位置要放在最后 + ch.pipeline().addLast(businessGroup, dispatcherHandler);//业务分发 这里使用业务线程组 + } + }); + return new AgentNettyServer(config); + } + +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agenttcp/server/annotation/AgentDispatcher.java b/src/main/java/com/tongran/agenttcp/server/annotation/AgentDispatcher.java new file mode 100644 index 0000000..4bc4d92 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/server/annotation/AgentDispatcher.java @@ -0,0 +1,47 @@ +package com.tongran.agenttcp.server.annotation; + +import com.tongran.agenttcp.server.netty.enums.MsgEnum; +import lombok.AllArgsConstructor; +import lombok.Getter; +import org.springframework.stereotype.Component; + +import java.lang.annotation.*; + +/** + * 指明类为充电消息 + */ +@Component +@Documented +@Target({ElementType.TYPE}) +@Retention(RetentionPolicy.RUNTIME) +public @interface AgentDispatcher { + + @Getter + @AllArgsConstructor + enum VersionEnum { + V1("2025"); + public final String value; + } + + /** + * 消息ID + * + * @return + */ + MsgEnum msgId(); + + /** + * 消息版本 默认版本2025 + * + * @return + */ + VersionEnum version() default VersionEnum.V1; + + /** + * 描述 + * + * @return + */ + String desc() default ""; + +} diff --git a/src/main/java/com/tongran/agenttcp/server/basics/AgentDispatcherManager.java b/src/main/java/com/tongran/agenttcp/server/basics/AgentDispatcherManager.java new file mode 100644 index 0000000..ab3e8b8 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/server/basics/AgentDispatcherManager.java @@ -0,0 +1,46 @@ +package com.tongran.agenttcp.server.basics; + +import com.tongran.agenttcp.server.annotation.AgentDispatcher; +import org.springframework.beans.BeansException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.stereotype.Component; +import org.springframework.util.CollectionUtils; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +@Component +public class AgentDispatcherManager implements ApplicationContextAware { + + /** + * 所有实现的包处理器 + */ + private static Map MSG_HANDLER_MAP; + + /** + * 唤醒时 初始化 packHandlerMap + */ + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + // 仅一次性初始化完成 + if (MSG_HANDLER_MAP == null) { + MSG_HANDLER_MAP = new ConcurrentHashMap<>(); + Map handlers = applicationContext.getBeansWithAnnotation(AgentDispatcher.class); + if (!CollectionUtils.isEmpty(handlers)) { + handlers.values().forEach(tempHandler -> { + boolean result = tempHandler.getClass().isAnnotationPresent(AgentDispatcher.class); + if (result) { + AgentDispatcher annotation = tempHandler.getClass().getAnnotation(AgentDispatcher.class); + MSG_HANDLER_MAP.put(annotation.msgId().getValue() + "&" + annotation.version().value, (AgentMsgHandler) tempHandler); + } + }); + } + } + } + + public AgentMsgHandler getHandler(String msgIdVersion) { + return MSG_HANDLER_MAP == null ? null : MSG_HANDLER_MAP.get(msgIdVersion); + } + +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agenttcp/server/basics/AgentMsgHandler.java b/src/main/java/com/tongran/agenttcp/server/basics/AgentMsgHandler.java new file mode 100644 index 0000000..18a2c57 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/server/basics/AgentMsgHandler.java @@ -0,0 +1,23 @@ +package com.tongran.agenttcp.server.basics; + + +import com.tongran.agenttcp.server.model.UpMsgResponse; + +public interface AgentMsgHandler { + + /** + * 处理终端传入的消息, 然后进行返回 + * + * @return 需要发送给终端的消息 + */ + + default UpMsgResponse upHandle(String data) { + return null; + } + + default String downHandle(String data) { + return ""; + } + + +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agenttcp/server/enpoint/AgentEndpoint.java b/src/main/java/com/tongran/agenttcp/server/enpoint/AgentEndpoint.java new file mode 100644 index 0000000..907ad52 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/server/enpoint/AgentEndpoint.java @@ -0,0 +1,265 @@ +package com.tongran.agenttcp.server.enpoint; + +import com.alibaba.fastjson2.JSONObject; +import com.tongran.agenttcp.server.annotation.AgentDispatcher; +import com.tongran.agenttcp.server.basics.AgentMsgHandler; +import com.tongran.agenttcp.server.model.UpMsgResponse; +import com.tongran.agenttcp.server.netty.enums.MsgEnum; +import org.springframework.stereotype.Component; + +@Component +public class AgentEndpoint { + +// @Resource +// private AgentMqConfig agentMqConfig; +// +// @Resource +// private AgentProducer agentProducer; + + @AgentDispatcher(msgId = MsgEnum.登录认证) + public class LoginHandler implements AgentMsgHandler { + @Override + public UpMsgResponse upHandle(String data) { + JSONObject jsonObject = JSONObject.parseObject(data); + String clientId = jsonObject.getString("clientId"); +// MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(data).build(); +// agentProducer.asyncSend(mqMsg); + JSONObject json = new JSONObject(); + json.put("resCode",1); + return UpMsgResponse.builder().clientId(clientId).content(json.toString()).build(); + } + } + + @AgentDispatcher(msgId = MsgEnum.登录认证应答) + public class LoginRspHandler implements AgentMsgHandler { + @Override + public String downHandle(String data) { +// T0x02 obj = JSONObject.parseObject(data, T0x02.class); + StringBuilder res = new StringBuilder(); + StringBuilder signData = new StringBuilder(); + res.append("68");// 起始字符 +// res.append(obj.getLen()); +// signData.append(obj.getSerialNumber()); +// signData.append("00"); +// signData.append(YkcUtil.coveringStr(Integer.toHexString(CmdEnum.登录认证应答.getValue()))); +// signData.append(obj.getPileNumber()); +// signData.append(YkcUtil.coveringStr(Integer.toHexString(obj.getLoginResult()))); +// res.append(signData); +// String crc = EscapeUtil.fill(Integer.toHexString(CrcUtil.calcCrc16(signData.toString())), 4); +// res.append(crc); + return res.toString(); + } + } + + @AgentDispatcher(msgId = MsgEnum.心跳包) + public class HeartBeatHandler implements AgentMsgHandler { + @Override + public UpMsgResponse upHandle(String data) { + JSONObject jsonObject = JSONObject.parseObject(data); + String clientId = jsonObject.getString("clientId"); +// MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(data).build(); +// agentProducer.asyncSend(mqMsg); + +// T0x04 obj = new T0x04(); +// StringBuilder res = new StringBuilder(); +// StringBuilder signData = new StringBuilder(); +// res.append("68");// 起始字符 +// res.append(obj.getLen()); +// signData.append(data.getSerialNumber()); +// signData.append("00"); +// signData.append(YkcUtil.coveringStr(Integer.toHexString(CmdEnum.心跳包应答.getValue()))); +// signData.append(data.getPileNumber()); +// signData.append(YkcUtil.coveringStr(Integer.toHexString(data.getGunNo()))); +// signData.append(YkcUtil.coveringStr(Integer.toHexString(0))); +// res.append(signData); +// String crc = EscapeUtil.fill(Integer.toHexString(CrcUtil.calcCrc16(signData.toString())), 4); +// res.append(crc); + JSONObject json = new JSONObject(); + json.put("resCode",1); + return UpMsgResponse.builder().clientId(clientId).content(json.toString()).build(); + } + } + + @AgentDispatcher(msgId = MsgEnum.心跳包应答) + public class HeartBeatRspHandler implements AgentMsgHandler { + @Override + public String downHandle(String data) { +// T0x04 obj = JSONObject.parseObject(data, T0x04.class); + StringBuilder res = new StringBuilder(); +// StringBuilder signData = new StringBuilder(); +// res.append("68");// 起始字符 +// res.append(obj.getLen()); +// signData.append(obj.getSerialNumber()); +// signData.append("00"); +// signData.append(YkcUtil.coveringStr(Integer.toHexString(CmdEnum.心跳包应答.getValue()))); +// signData.append(obj.getPileNumber()); +// signData.append(YkcUtil.coveringStr(Integer.toHexString(obj.getGunNo()))); +// signData.append(YkcUtil.coveringStr(Integer.toHexString(obj.getHeartbeat()))); +// res.append(signData); +// String crc = EscapeUtil.fill(Integer.toHexString(CrcUtil.calcCrc16(signData.toString())), 4); +// res.append(crc); + return res.toString(); + } + } + + + @AgentDispatcher(msgId = MsgEnum.CPU包) + public class CpuHandler implements AgentMsgHandler { + @Override + public UpMsgResponse upHandle(String data) { + JSONObject jsonObject = JSONObject.parseObject(data); + String clientId = jsonObject.getString("clientId"); + JSONObject json = new JSONObject(); + json.put("resCode",1); + return UpMsgResponse.builder().clientId(clientId).content(json.toString()).build(); + } + } + + @AgentDispatcher(msgId = MsgEnum.CPU包应答) + public class CpuRspHandler implements AgentMsgHandler { + @Override + public String downHandle(String data) { + StringBuilder res = new StringBuilder(); + + return res.toString(); + } + } + + @AgentDispatcher(msgId = MsgEnum.磁盘包) + public class DiskHandler implements AgentMsgHandler { + @Override + public UpMsgResponse upHandle(String data) { + JSONObject jsonObject = JSONObject.parseObject(data); + String clientId = jsonObject.getString("clientId"); + JSONObject json = new JSONObject(); + json.put("resCode",1); + return UpMsgResponse.builder().clientId(clientId).content(json.toString()).build(); + } + } + + @AgentDispatcher(msgId = MsgEnum.磁盘包应答) + public class DiskRspHandler implements AgentMsgHandler { + @Override + public String downHandle(String data) { + StringBuilder res = new StringBuilder(); + + return res.toString(); + } + } + + @AgentDispatcher(msgId = MsgEnum.内存包) + public class MemoryHandler implements AgentMsgHandler { + @Override + public UpMsgResponse upHandle(String data) { + JSONObject jsonObject = JSONObject.parseObject(data); + String clientId = jsonObject.getString("clientId"); + JSONObject json = new JSONObject(); + json.put("resCode",1); + return UpMsgResponse.builder().clientId(clientId).content(json.toString()).build(); + } + } + + @AgentDispatcher(msgId = MsgEnum.内存包应答) + public class MemoryRspHandler implements AgentMsgHandler { + @Override + public String downHandle(String data) { + StringBuilder res = new StringBuilder(); + + return res.toString(); + } + } + + @AgentDispatcher(msgId = MsgEnum.网卡包) + public class NetHandler implements AgentMsgHandler { + @Override + public UpMsgResponse upHandle(String data) { + JSONObject jsonObject = JSONObject.parseObject(data); + String clientId = jsonObject.getString("clientId"); + JSONObject json = new JSONObject(); + json.put("resCode",1); + return UpMsgResponse.builder().clientId(clientId).content(json.toString()).build(); + } + } + + @AgentDispatcher(msgId = MsgEnum.网卡包应答) + public class NetRspHandler implements AgentMsgHandler { + @Override + public String downHandle(String data) { + StringBuilder res = new StringBuilder(); + + return res.toString(); + } + } + + @AgentDispatcher(msgId = MsgEnum.挂载点包) + public class PointHandler implements AgentMsgHandler { + @Override + public UpMsgResponse upHandle(String data) { + JSONObject jsonObject = JSONObject.parseObject(data); + String clientId = jsonObject.getString("clientId"); + JSONObject json = new JSONObject(); + json.put("resCode",1); + return UpMsgResponse.builder().clientId(clientId).content(json.toString()).build(); + } + } + + @AgentDispatcher(msgId = MsgEnum.挂载点包应答) + public class PointRspHandler implements AgentMsgHandler { + @Override + public String downHandle(String data) { + StringBuilder res = new StringBuilder(); + + return res.toString(); + } + } + + @AgentDispatcher(msgId = MsgEnum.系统包) + public class SystemHandler implements AgentMsgHandler { + @Override + public UpMsgResponse upHandle(String data) { + JSONObject jsonObject = JSONObject.parseObject(data); + String clientId = jsonObject.getString("clientId"); + JSONObject json = new JSONObject(); + json.put("resCode",1); + return UpMsgResponse.builder().clientId(clientId).content(json.toString()).build(); + } + } + + @AgentDispatcher(msgId = MsgEnum.系统包应答) + public class SystemRspHandler implements AgentMsgHandler { + @Override + public String downHandle(String data) { + StringBuilder res = new StringBuilder(); + + return res.toString(); + } + } + + @AgentDispatcher(msgId = MsgEnum.交换机包) + public class SwitchBoardHandler implements AgentMsgHandler { + @Override + public UpMsgResponse upHandle(String data) { + JSONObject jsonObject = JSONObject.parseObject(data); + String clientId = jsonObject.getString("clientId"); + JSONObject json = new JSONObject(); + json.put("resCode",1); + return UpMsgResponse.builder().clientId(clientId).content(json.toString()).build(); + } + } + + @AgentDispatcher(msgId = MsgEnum.交换机包应答) + public class SwitchBoardRspHandler implements AgentMsgHandler { + @Override + public String downHandle(String data) { + StringBuilder res = new StringBuilder(); + + return res.toString(); + } + } + + + + + + +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agenttcp/server/handler/AgentDecoderHandler.java b/src/main/java/com/tongran/agenttcp/server/handler/AgentDecoderHandler.java new file mode 100644 index 0000000..2589f18 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/server/handler/AgentDecoderHandler.java @@ -0,0 +1,126 @@ +package com.tongran.agenttcp.server.handler; + +import cn.hutool.cache.Cache; +import cn.hutool.cache.CacheUtil; +import cn.hutool.core.date.DateUnit; +import com.alibaba.fastjson2.JSONObject; +import com.tongran.agenttcp.server.annotation.AgentDispatcher; +import com.tongran.agenttcp.server.basics.AgentDispatcherManager; +import com.tongran.agenttcp.server.basics.AgentMsgHandler; +import com.tongran.agenttcp.server.model.Message; +import com.tongran.agenttcp.server.model.UpMsgResponse; +import com.tongran.agenttcp.server.netty.core.session.SessionManager; +import com.tongran.agenttcp.utils.AssertLog; +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.util.CharsetUtil; +import org.springframework.stereotype.Component; + +import javax.annotation.Resource; + +@Component +@ChannelHandler.Sharable +public class AgentDecoderHandler extends ChannelInboundHandlerAdapter { + + protected final SessionManager sessionManager; + + @Resource + private AgentDispatcherManager agentDispatcherManager; + + public AgentDecoderHandler() { + this.sessionManager = SessionManager.getInstance(); + } + + // 用来临时保留没有处理过的请求报文 + private final Cache lruCache = CacheUtil.newLRUCache(3000); + + /** + * 消息解码器 + */ + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) { + // 检查是否为 ByteBuf（未解码的原始数据） + if (msg instanceof ByteBuf) { + ByteBuf byteBuf = (ByteBuf) msg; + String messages = byteBuf.toString(CharsetUtil.UTF_8); // 指定字符集解码 +// AssertLog.info("<<[up1]:IP:{},[up-content]==>{}", sessionManager.client(ctx),messages); + + String content = ""; + boolean isClear = false; + String tempMsg = lruCache.get(sessionManager.client(ctx)); + tempMsg = tempMsg == null ? "" : tempMsg; + int tmpMsgSize = tempMsg.length(); + + boolean startsWith = messages.startsWith("agent-tcp:"); + boolean endsWith = messages.endsWith("@tong-ran"); + String ms = messages.replaceAll("agent-tcp:",""); + String[] arr_msg = ms.split("@tong-ran"); + if(startsWith){ + //判定是否整包 + if(arr_msg.length == 1){ + //判定是否拆包 + if(endsWith){ + lruCache.remove(sessionManager.client(ctx)); + tmpMsgSize = 0; + content = arr_msg[0]+"@tong-ran"; + }else{ + lruCache.remove(sessionManager.client(ctx)); + tmpMsgSize = 0; + content = arr_msg[0]; + lruCache.put(sessionManager.client(ctx), content, DateUnit.SECOND.getMillis() * 3000); + } + } + //判定是否粘包 + if(arr_msg.length > 1){ + if(!endsWith){ + lruCache.remove(sessionManager.client(ctx)); + tmpMsgSize = 0; + content = arr_msg[0]+"@tong-ran"; + lruCache.put(sessionManager.client(ctx), arr_msg[1], DateUnit.SECOND.getMillis() * 3000); + }else{ + lruCache.remove(sessionManager.client(ctx)); + tmpMsgSize = 0; + content = String.join("@tong-ran", arr_msg); + } + } + } + if (tmpMsgSize > 0) { + content = tempMsg + messages; + isClear = true; + } + AssertLog.info("<<[up]:IP:{},[up-content]==>{}", sessionManager.client(ctx),content); + endsWith = content.endsWith("@tong-ran"); + //判断是否是整包 + if(endsWith){ + String[] arr = content.split("@tong-ran"); + for (String message : arr) { + JSONObject jsonObject = JSONObject.parseObject(message); + String dataType = jsonObject.getString("dataType"); + AgentMsgHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value); + UpMsgResponse response = msgHandler.upHandle(message); + AssertLog.info("<<[up-after-handle]:clientId:{},type={},[handle-content]={}", response.getClientId(), dataType, message); + Message agentMessage = Message.builder().build(); + agentMessage.setClientId(response.getClientId()); + agentMessage.setDataType(dataType); + agentMessage.setData(response.getContent()); + ctx.fireChannelRead(agentMessage);//传递到下一个handler + } + } + if (isClear) { + lruCache.remove(sessionManager.client(ctx)); + } + byteBuf.release(); // 释放 ByteBuf 资源（重要！） + } else { + System.out.println("Unexpected message type: " + msg.getClass()); + } + } + + //拆包 + public byte[] subByte(byte[] b, int off, int length) { + byte[] bytes = new byte[length]; + System.arraycopy(b, off, bytes, 0, length); + return bytes; + } +} diff --git a/src/main/java/com/tongran/agenttcp/server/handler/AgentDispatcherHandler.java b/src/main/java/com/tongran/agenttcp/server/handler/AgentDispatcherHandler.java new file mode 100644 index 0000000..745228d --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/server/handler/AgentDispatcherHandler.java @@ -0,0 +1,38 @@ +package com.tongran.agenttcp.server.handler; + +import com.tongran.agenttcp.server.model.Message; +import com.tongran.agenttcp.server.netty.core.session.Session; +import com.tongran.agenttcp.server.netty.core.session.SessionManager; +import com.tongran.agenttcp.utils.AssertLog; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; +import org.apache.commons.lang3.StringUtils; +import org.springframework.stereotype.Component; + +@Component +@ChannelHandler.Sharable +public class AgentDispatcherHandler extends SimpleChannelInboundHandler { + + protected final SessionManager sessionManager; + + public AgentDispatcherHandler() { + this.sessionManager = SessionManager.getInstance(); + } + + @Override + protected void channelRead0(ChannelHandlerContext ctx, Message msg) { +// AssertLog.info(">>>>>>[要处理的终端数据] {}", msg.toString()); + String clientId = msg.getClientId(); + if (StringUtils.isBlank(clientId)) { + AssertLog.error("<<<<<<错误的信息from:{}", ctx.channel().remoteAddress()); + return; + } + Session session = Session.buildSession(ctx, clientId); + sessionManager.put(clientId, session); + if (StringUtils.isNotBlank(msg.getData())) { +// AssertLog.info(">>>>>>[getSessionById的终端数据] {}", sessionManager.getSessionById(clientId)); + sessionManager.writeAndFlush(sessionManager.getSessionById(clientId).getChannel(), msg); + } + } +} diff --git a/src/main/java/com/tongran/agenttcp/server/handler/AgentEncoderHandler.java b/src/main/java/com/tongran/agenttcp/server/handler/AgentEncoderHandler.java new file mode 100644 index 0000000..5555cf4 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/server/handler/AgentEncoderHandler.java @@ -0,0 +1,48 @@ +package com.tongran.agenttcp.server.handler; + +import com.alibaba.fastjson2.JSON; +import com.tongran.agenttcp.server.model.Message; +import com.tongran.agenttcp.server.netty.core.session.SessionManager; +import com.tongran.agenttcp.utils.AssertLog; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelOutboundHandlerAdapter; +import io.netty.channel.ChannelPromise; +import org.apache.commons.lang3.StringUtils; +import org.springframework.stereotype.Component; + +import java.nio.charset.StandardCharsets; + +@Component +@ChannelHandler.Sharable +public class AgentEncoderHandler extends ChannelOutboundHandlerAdapter { + + protected final SessionManager sessionManager; + + public AgentEncoderHandler() { + this.sessionManager = SessionManager.getInstance(); + } + + /** + * 消息解码器 + */ + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { + if (msg instanceof Message) { + Message entity = (Message) msg; + if (StringUtils.isBlank(entity.getData())) { + AssertLog.error(">>[down]:IP:{},errorContent:{}", sessionManager.client(ctx), msg); + return; + } + AssertLog.info(">>[down]:IP:{},[content]==>{}", sessionManager.client(ctx), entity.getData()); + String json = JSON.toJSONString(entity); + byte[] bytes = json.getBytes(StandardCharsets.UTF_8); // 显式指定 UTF-8 +// byte[] bytes = EscapeUtil.hexStringToByteArray(entity.getContent()); + ByteBuf buf = Unpooled.wrappedBuffer(bytes); + ctx.write(buf, promise); + } + } + +} diff --git a/src/main/java/com/tongran/agenttcp/server/model/Message.java b/src/main/java/com/tongran/agenttcp/server/model/Message.java new file mode 100644 index 0000000..5f45ca3 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/server/model/Message.java @@ -0,0 +1,37 @@ +package com.tongran.agenttcp.server.model; + + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.io.Serializable; + +/** + * @author egrias + */ +@Data +@Builder +@AllArgsConstructor +@NoArgsConstructor +public class Message implements Serializable { + + private static final long serialVersionUID = -1267013167162440610L; + + /** + * clientId + */ + private String clientId; + + /** + * type + */ + private String dataType; + + /** + * 发送内容 + */ + private String data; + +} diff --git a/src/main/java/com/tongran/agenttcp/server/model/UpMsgResponse.java b/src/main/java/com/tongran/agenttcp/server/model/UpMsgResponse.java new file mode 100644 index 0000000..02315a5 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/server/model/UpMsgResponse.java @@ -0,0 +1,18 @@ +package com.tongran.agenttcp.server.model; + +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; +import lombok.experimental.SuperBuilder; + +@Data +@SuperBuilder +@AllArgsConstructor +@NoArgsConstructor +public class UpMsgResponse { + + private String clientId;// 终端号 + + private String content; + +} diff --git a/src/main/java/com/tongran/agenttcp/server/netty/BaseNettyServer.java b/src/main/java/com/tongran/agenttcp/server/netty/BaseNettyServer.java new file mode 100644 index 0000000..6ac2e55 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/server/netty/BaseNettyServer.java @@ -0,0 +1,117 @@ +package com.tongran.agenttcp.server.netty; + +import com.tongran.agenttcp.server.netty.config.BaseNettyConfig; +import com.tongran.agenttcp.utils.AssertLog; +import io.netty.bootstrap.AbstractBootstrap; +import io.netty.bootstrap.Bootstrap; +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.NioChannelOption; +import io.netty.channel.socket.nio.NioDatagramChannel; +import io.netty.channel.socket.nio.NioServerSocketChannel; +import io.netty.util.ResourceLeakDetector; +import io.netty.util.concurrent.DefaultEventExecutorGroup; +import io.netty.util.concurrent.DefaultThreadFactory; +import io.netty.util.concurrent.EventExecutorGroup; +import io.netty.util.concurrent.Future; + +/** + * 基础Netty服务 + */ +public abstract class BaseNettyServer { + + protected boolean isRunning; + + protected BaseNettyConfig config; + + protected EventLoopGroup bossGroup; + + protected EventLoopGroup workerGroup; + + protected EventExecutorGroup businessGroup; + + protected BaseNettyServer(BaseNettyConfig config) { + this.config = config; + } + + protected AbstractBootstrap initializeTcp() { + bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory(config.name, Thread.MAX_PRIORITY)); + workerGroup = new NioEventLoopGroup(config.workerCore, new DefaultThreadFactory(config.name, Thread.MAX_PRIORITY)); + if (config.businessCore > 0) { + businessGroup = new DefaultEventExecutorGroup(config.businessCore); + } + ServerBootstrap serverBootstrap = new ServerBootstrap(); + serverBootstrap.group(bossGroup, workerGroup) + .channel(NioServerSocketChannel.class) + .childOption(ChannelOption.SO_REUSEADDR, true) + .option(ChannelOption.SO_BACKLOG, 1024) + .childOption(NioChannelOption.TCP_NODELAY, true) + .childHandler(config.hander); + //内存泄漏检测 开发推荐PARANOID 线上SIMPLE + ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.SIMPLE); + return serverBootstrap; + } + + protected AbstractBootstrap initializeUdp() { + bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory(config.name, Thread.MAX_PRIORITY)); + if (config.businessCore > 0) { + businessGroup = new DefaultEventExecutorGroup(config.businessCore); + } + return new Bootstrap() + .group(bossGroup).channel(NioDatagramChannel.class) + .option(NioChannelOption.SO_REUSEADDR, true) + .option(NioChannelOption.SO_RCVBUF, 1024 * 1024 * 50) + .handler(config.hander); + } + + public synchronized boolean start() { + if (!config.enable) { + return false; + } + if (isRunning) { + AssertLog.info("======{}已经启动,port:{}======", config.name, config.port); + return isRunning; + } + AbstractBootstrap bootstrap = config.isTcp ? initializeTcp() : initializeUdp(); + ChannelFuture future = bootstrap.bind(config.port).awaitUninterruptibly(); + future.channel().closeFuture().addListener(f -> { + if (isRunning) { + stop(); + } + }); + if (future.cause() != null) { + AssertLog.error("===启动失败===", future.cause()); + } + if (isRunning = future.isSuccess()) { + AssertLog.info("\n\n\t\t\t\t\t\t\t\t======{}启动成功,port:{}======\n", config.name, config.port); + } + return isRunning; + } + + public synchronized void stop() { + if (!config.enable) { + return; + } + isRunning = false; + try { + Future future = this.workerGroup.shutdownGracefully().await(); + if (!future.isSuccess()) { + AssertLog.error("workerGroup 无法正常停止:{}", future.cause()); + } + future = this.bossGroup.shutdownGracefully().await(); + if (!future.isSuccess()) { + AssertLog.error("bossGroup 无法正常停止:{}", future.cause()); + } + future = this.businessGroup.shutdownGracefully().await(); + if (!future.isSuccess()) { + AssertLog.error("businessGroup 无法正常停止:{}", future.cause()); + } + } catch (InterruptedException e) { + e.printStackTrace(); + } + AssertLog.info("\n\n\t\t\t\t\t\t\t\t======{} 已经停止,port:{}======\n", config.name, config.port); + } +} diff --git a/src/main/java/com/tongran/agenttcp/server/netty/config/BaseNettyConfig.java b/src/main/java/com/tongran/agenttcp/server/netty/config/BaseNettyConfig.java new file mode 100644 index 0000000..cb8b380 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/server/netty/config/BaseNettyConfig.java @@ -0,0 +1,58 @@ +package com.tongran.agenttcp.server.netty.config; + +import com.tongran.agenttcp.server.netty.core.session.SessionManager; +import io.netty.channel.ChannelHandler; +import io.netty.util.NettyRuntime; +import lombok.Data; + +import java.io.Serializable; + +/** + * 基础Netty配置 + */ +@Data +public class BaseNettyConfig implements Serializable { + + private static final long serialVersionUID = -2736718499972710895L; + + /** + * 是否开启 + */ + public Boolean enable = false; + + /** + * 是否TCP + */ + public Boolean isTcp = true; + + /** + * 服务名称 + */ + public String name = "Netty"; + + /** + * 端口号 + */ + public Integer port = 11111; + + public Integer workerCore = NettyRuntime.availableProcessors() * 2; + + public Integer businessCore = Math.max(1, NettyRuntime.availableProcessors() >> 1); + + public Integer readerIdleTime = 240; + + public Integer writerIdleTime = 0; + + public Integer allIdleTime = 0; + + /** + * netty Session管理 + */ + public SessionManager sessionManager; + + /** + * netty Channel处理列表 + */ + public ChannelHandler hander; + +} diff --git a/src/main/java/com/tongran/agenttcp/server/netty/core/handler/TCPListenHandler.java b/src/main/java/com/tongran/agenttcp/server/netty/core/handler/TCPListenHandler.java new file mode 100644 index 0000000..a318288 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/server/netty/core/handler/TCPListenHandler.java @@ -0,0 +1,60 @@ +package com.tongran.agenttcp.server.netty.core.handler; + +import com.tongran.agenttcp.server.netty.core.session.SessionManager; +import com.tongran.agenttcp.utils.AssertLog; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.handler.timeout.IdleState; +import io.netty.handler.timeout.IdleStateEvent; +import org.springframework.stereotype.Component; + +import java.io.IOException; + +/** + * @Description: TCP消息适配器 + */ +@Component +@ChannelHandler.Sharable +public class TCPListenHandler extends ChannelInboundHandlerAdapter { + + private final SessionManager sessionManager; + + public TCPListenHandler() { + this.sessionManager = SessionManager.getInstance(); + } + + @Override + public void channelActive(ChannelHandlerContext ctx) { + AssertLog.info("<<<<<<[终端连接]{}", ctx.channel().remoteAddress()); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) { + AssertLog.info(">>>>>>[断开连接]{}", sessionManager.client(ctx)); + sessionManager.remove(ctx.channel()); + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) { + if (e instanceof IOException) { + AssertLog.info("<<<<<<[终端断开连接]{} {}", sessionManager.client(ctx), e.getMessage()); + } else { + AssertLog.info(">>>>>>[消息处理异常]" + sessionManager.client(ctx), e); + } + } + + @Override + public void userEventTriggered(ChannelHandlerContext ctx, Object evt) { + if (evt instanceof IdleStateEvent) { + IdleStateEvent event = (IdleStateEvent) evt; + IdleState state = event.state(); + if (state == IdleState.READER_IDLE || state == IdleState.WRITER_IDLE || state == IdleState.ALL_IDLE) { + AssertLog.warn(">>>>>>[终端心跳超时]{} {}", state, sessionManager.client(ctx)); + sessionManager.remove(ctx.channel()); + ctx.close(); + } + } + } + +} diff --git a/src/main/java/com/tongran/agenttcp/server/netty/core/handler/UDPListenHandler.java b/src/main/java/com/tongran/agenttcp/server/netty/core/handler/UDPListenHandler.java new file mode 100644 index 0000000..f0257b3 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/server/netty/core/handler/UDPListenHandler.java @@ -0,0 +1,24 @@ +package com.tongran.agenttcp.server.netty.core.handler; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.socket.DatagramPacket; +import org.springframework.stereotype.Component; + +/** + * @Description: UDP消息适配器 + */ +@Component +@ChannelHandler.Sharable +public class UDPListenHandler extends ChannelInboundHandlerAdapter { + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) { + DatagramPacket packet = (DatagramPacket) msg; + ByteBuf buf = packet.content(); + buf.clear();// 暂未实现 + } + +} diff --git a/src/main/java/com/tongran/agenttcp/server/netty/core/session/Session.java b/src/main/java/com/tongran/agenttcp/server/netty/core/session/Session.java new file mode 100644 index 0000000..ec67d26 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/server/netty/core/session/Session.java @@ -0,0 +1,85 @@ +package com.tongran.agenttcp.server.netty.core.session; + +import io.netty.channel.ChannelHandlerContext; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; +import lombok.experimental.Accessors; + +import java.io.Serializable; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * SESSION + * + */ +@Data +@Builder +@Accessors(chain = true) +@AllArgsConstructor +@NoArgsConstructor +public class Session implements Serializable { + + private static final long serialVersionUID = 1L; + + /** + * sessionId + */ + private String sessionId; + + /** + * 客户端唯一标识 + */ + private String clientId; + + /** + * Seesion额外参数 + */ + private ConcurrentHashMap extraMap; + + /** + * 是否鉴权 + */ + @Builder.Default + private boolean isAuthenticated = false; + + @Builder.Default + private long createTime = System.currentTimeMillis(); + + @Builder.Default + private long updateTime = System.currentTimeMillis(); + + /** + * 消息渠道 + */ + private ChannelHandlerContext channel; + + @Builder.Default + private AtomicInteger serialNo = new AtomicInteger(0); + + public static String buildSessionId(ChannelHandlerContext channel) { + return channel.channel().id().asLongText(); + } + + public static Session buildSession(ChannelHandlerContext channel, String clientId) { + return Session.builder().channel(channel).sessionId(buildSessionId(channel)).clientId(clientId).build(); + } + + /** + * 自生成流水号 + * + * @return + */ + public int nextSerialNo() { + int current; + int next; + do { + current = serialNo.get(); + next = current > 0xffff ? 0 : current; + } while (!serialNo.compareAndSet(current, next + 1)); + return next; + } + +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agenttcp/server/netty/core/session/SessionManager.java b/src/main/java/com/tongran/agenttcp/server/netty/core/session/SessionManager.java new file mode 100644 index 0000000..ff2d38d --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/server/netty/core/session/SessionManager.java @@ -0,0 +1,137 @@ +package com.tongran.agenttcp.server.netty.core.session; + + +import cn.hutool.cache.Cache; +import cn.hutool.cache.CacheUtil; +import cn.hutool.core.bean.BeanUtil; +import cn.hutool.core.util.ObjectUtil; +import com.tongran.agenttcp.exception.ServerException; +import com.tongran.agenttcp.utils.AssertLog; +import io.netty.channel.Channel; +import io.netty.channel.ChannelHandlerContext; +import io.netty.util.Attribute; +import io.netty.util.AttributeKey; +import lombok.Data; +import org.apache.commons.lang3.StringUtils; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +@Data +public class SessionManager { + + private static volatile SessionManager instance = null; + + // clientId-Session + private final Map sessionMap = new ConcurrentHashMap<>(); + + private final Cache sessionCache = CacheUtil.newTimedCache(10 * 60 * 1000); + + public static SessionManager getInstance() { + if (instance == null) { + synchronized (SessionManager.class) { + if (instance == null) { + instance = new SessionManager(); + } + } + } + return instance; + } + + public synchronized void put(String clientId, Session session) { + if (StringUtils.isNotBlank(session.getClientId())) { + sessionMap.put(session.getClientId(), session); + } + } + + public synchronized void remove(String clientId) { + if (StringUtils.isNotBlank(clientId)) { + sessionMap.remove(clientId); + } + } + + public synchronized void remove(Session session) { + if (session != null && StringUtils.isNotBlank(session.getClientId())) { + sessionMap.remove(session.getClientId(), session); + } + } + + public synchronized void remove(Channel channel) { + remove(getSessionByChannel(channel)); + } + + public Session getSessionById(String clientId) { + return sessionMap.get(clientId); + } + + public Session getSessionByChannel(Channel channel) { + Session session = new Session(); + sessionMap.values().forEach(s -> { + if (s.getChannel().channel() == channel) { + BeanUtil.copyProperties(s, session, true); + } + }); + return session; + } + + public boolean containsSession(String clientId) { + return sessionMap.containsKey(clientId); + } + + public boolean containsSession(Session session) { + return sessionMap.containsValue(session); + } + + public void setSessionCache(String clientId, Object value) { + sessionCache.put(clientId, value); + } + + public Object getSessionCache(String clientId) { + return sessionCache.get(clientId); + } + + public String client(ChannelHandlerContext ctx) { + Channel channel = ctx.channel(); + Session session = this.getSessionByChannel(channel); + if (ObjectUtil.isNotNull(session) && StringUtils.isNotBlank(session.getClientId())) { + return channel.remoteAddress().toString() + "/" + session.getClientId(); + } + return channel.remoteAddress().toString(); + } + + /** + * 根据channel生成流水号 + * + * @param channel + * @return + */ + public short getSerialNumber(Channel channel, AttributeKey serialNumber) { + Attribute flowIdAttr = channel.attr(serialNumber); + Short flowId = flowIdAttr.get(); + if (flowId == null) { + flowId = 0; + } else { + flowId++; + } + flowIdAttr.set(flowId); + return flowId; + } + + public void writeAndFlush(String clientId, Object msg) { + Session session = this.getSessionById(clientId); + if (ObjectUtil.isNotNull(session)) { + this.writeAndFlush(session.getChannel(), msg); + } else { + throw new ServerException(500, "终端:" + clientId + "离线或者不存在"); + } + } + + public void writeAndFlush(ChannelHandlerContext ctx, Object msg) { + ctx.writeAndFlush(msg).addListener(future -> { + if (!future.isSuccess()) { + AssertLog.error("消息发送失败:{}", future.cause()); + } + }); + } + +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agenttcp/server/netty/enums/MsgEnum.java b/src/main/java/com/tongran/agenttcp/server/netty/enums/MsgEnum.java new file mode 100644 index 0000000..bd87c86 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/server/netty/enums/MsgEnum.java @@ -0,0 +1,61 @@ +package com.tongran.agenttcp.server.netty.enums; + +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.NoArgsConstructor; + +/** + * 控制码 + */ +@Getter +@AllArgsConstructor +@NoArgsConstructor +public enum MsgEnum { + + 登录认证("LOGIN"), + + 登录认证应答("LOGIN_RSP"), + + 心跳包("HEARTBEAT"), + + 心跳包应答("HEARTBEAT_RSP"), + + CPU包("CPU"), + + CPU包应答("CPU_RSP"), + + 内存包("MEMORY"), + + 内存包应答("MEMORY_RSP"), + + 系统包("SYSTEM"), + + 系统包应答("SYSTEM_RSP"), + + 磁盘包("DISK"), + + 磁盘包应答("DISK_RSP"), + + 网卡包("NET"), + + 网卡包应答("NET_RSP"), + + 挂载点包("POINT"), + + 挂载点包应答("POINT_RSP"), + + 容器包("DOCKER"), + + 容器包应答("DOCKER_RSP"), + + 交换机包("SWITCHBOARD"), + + 交换机包应答("SWITCHBOARD_RSP"); + + private String value; + +// private String start; + +// private String serialNumber; + +} diff --git a/src/main/java/com/tongran/agenttcp/utils/AssertLog.java b/src/main/java/com/tongran/agenttcp/utils/AssertLog.java new file mode 100644 index 0000000..2757a60 --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/utils/AssertLog.java @@ -0,0 +1,73 @@ +package com.tongran.agenttcp.utils; + +import lombok.extern.slf4j.Slf4j; + +/** + * 日志断言类 + * + * @author LJ + * + */ +@Slf4j +public class AssertLog { + + /** + * 打印Info 日志 + * + * @param format + * @param arguments + */ + public static void info(String format, Object... arguments) { + if (log.isInfoEnabled()) { + log.info(format, arguments); + } + } + + /** + * 打印Debug 日志 + * + * @param format + * @param arguments + */ + public static void debug(String format, Object... arguments) { + if (log.isDebugEnabled()) { + log.debug(format, arguments); + } + } + + /** + * 打印Error 日志 + * + * @param format + * @param arguments + */ + public static void error(String format, Object... arguments) { + if (log.isErrorEnabled()) { + log.error(format, arguments); + } + } + + /** + * 打印Trace 日志 + * + * @param format + * @param arguments + */ + public static void trace(String format, Object... arguments) { + if (log.isTraceEnabled()) { + log.trace(format, arguments); + } + } + + /** + * 打印Warn 日志 + * + * @param format + * @param arguments + */ + public static void warn(String format, Object... arguments) { + if (log.isWarnEnabled()) { + log.warn(format, arguments); + } + } +} diff --git a/src/main/java/com/tongran/agenttcp/utils/R.java b/src/main/java/com/tongran/agenttcp/utils/R.java new file mode 100644 index 0000000..7de445b --- /dev/null +++ b/src/main/java/com/tongran/agenttcp/utils/R.java @@ -0,0 +1,76 @@ +package com.tongran.agenttcp.utils; + +import com.tongran.agenttcp.exception.code.ErrorCode; +import com.tongran.agenttcp.exception.code.GlobalErrorCode; +import io.swagger.v3.oas.annotations.media.Schema; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.io.Serializable; + +/** + * @author LJ + */ +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +@Schema(description = "接口交互统一数据返回标准") +public class R implements Serializable { + + private static final long serialVersionUID = 1L; + + @Schema(description = "返回代码") + private Integer code; + + @Schema(description = "消息描述") + private String msg; + + @Schema(description = "结果对象") + private T data; + + public static R success(String msg, T t) { + R r = new R<>(); + r.setData(t); + r.setMsg(msg); + r.setCode(GlobalErrorCode.SUCCESS.getCode()); + return r; + } + + public static R success(T t) { + return R.success(GlobalErrorCode.SUCCESS.getMsg(), t); + } + + public static R success() { + return R.success(null); + } + + public static R error(String msg, Integer code) { + R r = new R<>(); + r.setMsg(msg); + r.setCode(code); + return r; + } + + public static R error(Integer code, String msg) { + R r = new R<>(); + r.setMsg(msg); + r.setCode(code); + return r; + } + + public static R error(ErrorCode err) { + return R.error(err.getMsg(), err.getCode()); + } + + public static R error() { + return R.error(GlobalErrorCode.ERROR.getMsg(), GlobalErrorCode.ERROR.getCode()); + } + + public static R error(String msg) { + return R.error(msg, GlobalErrorCode.ERROR.getCode()); + } + +} \ No newline at end of file diff --git a/src/main/resources/application-dev.yml b/src/main/resources/application-dev.yml new file mode 100644 index 0000000..70d846f --- /dev/null +++ b/src/main/resources/application-dev.yml @@ -0,0 +1,27 @@ +server: + port: 7010 + servlet: + context-path: /agent-tcp + +# 接口文档配置 +knife4j: + enable: true + production: false # 开启屏蔽文档资源 + +# 日志配置 +logging: + file: + path: /data/agent-tcp/logs + +rocketmq: + name-server: 172.16.15.103:9876 + enabled: true + + +tcp: + netty: + charge: + enable: true + name: AGENT-TCP-服务 + port: 6610 + readerIdleTime: 300 \ No newline at end of file diff --git a/src/main/resources/application-prod.yml b/src/main/resources/application-prod.yml new file mode 100644 index 0000000..16cca1d --- /dev/null +++ b/src/main/resources/application-prod.yml @@ -0,0 +1,27 @@ +server: + port: 7010 + servlet: + context-path: /agent-tcp + +# 接口文档配置 +knife4j: + enable: true + production: false # 开启屏蔽文档资源 + +# 日志配置 +logging: + file: + path: /data/agent-tcp/logs + +rocketmq: + name-server: 172.16.15.103:9876 + enabled: true + + +tcp: + netty: + charge: + enable: true + name: AGENT-TCP-服务 + port: 6610 + readerIdleTime: 300 \ No newline at end of file diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml new file mode 100644 index 0000000..87d3eec --- /dev/null +++ b/src/main/resources/application.yml @@ -0,0 +1,43 @@ +spring: + profiles: + active: dev + mvc: + pathmatch: + matching-strategy: ant_path_matcher + application: + name: agent-tcp + version: 1.0 + web: + resources: + static-locations: classpath*:/META-INF/resources/ +logging: + config: classpath:logback/logback-${spring.profiles.active}.xml + +# springdoc-openapi项目配置 +knife4j: + setting: + enable-footer-custom: true + footer-custom-content: Apache License 2.0 | Copyright © 2025-[] +springdoc: + config: + title: AGENTTCP服务 + description: AGENTTCP服务接口文档 + contact: AGENTTCP + email: + version: ${spring.application.version} + group-configs: + - group: 'AGENTTCP服务' + paths-to-match: '/**' + packages-to-scan: com.tongran.agenttcp + +rocketmq: + producer: + group: tongran-group + topic: default_topic + agent-group: agent_up_group + agent-topic: agent_up + consumer: + group: tongran-group + topic: default_topic + agent-group: agent_down_group + agent-topic: agent_down diff --git a/src/main/resources/logback/logback-dev.xml b/src/main/resources/logback/logback-dev.xml new file mode 100644 index 0000000..cf1e7db --- /dev/null +++ b/src/main/resources/logback/logback-dev.xml @@ -0,0 +1,116 @@ + + + + + ${appName} + + + + + + + [requestId:%X{requestId}] [time:%d{yyyy-MM-dd HH:mm:ss.SSS}] [thread:%thread] [level:%-5level] [appName:${appName}] msg- %msg%n + + + + + + + ${LOG_HOME}/debug.log + + + + ${LOG_HOME}/debug.%d{yyyy-MM-dd}.%i.AssertLog.zip + + 30 + + 10GB + + + + 10MB + + + + + + [requestId:%X{requestId}] [time:%d{yyyy-MM-dd HH:mm:ss.SSS}] [thread:%thread] [level:%-5level] [appName:${appName}] msg- %msg%n + + + + + + + + INFO + ACCEPT + DENY + + + ${LOG_HOME}/info.log + + + + ${LOG_HOME}/info.%d{yyyy-MM-dd}.%i.AssertLog.zip + + 30 + + 10GB + + + + 10MB + + + + + + [requestId:%X{requestId}] [time:%d{yyyy-MM-dd HH:mm:ss.SSS}] [thread:%thread] [level:%-5level] [appName:${appName}] msg- %msg%n + + + + + + + + ERROR + ACCEPT + DENY + + ${LOG_HOME}/error.log + + + + ${LOG_HOME}/error.%d{yyyy-MM-dd}.%i.AssertLog.zip + + 30 + + 10GB + + + + 10MB + + + + + + [requestId:%X{requestId}] [time:%d{yyyy-MM-dd HH:mm:ss.SSS}] [thread:%thread] [level:%-5level] [appName:${appName}] msg- %msg%n + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/src/main/resources/logback/logback-prod.xml b/src/main/resources/logback/logback-prod.xml new file mode 100644 index 0000000..859494f --- /dev/null +++ b/src/main/resources/logback/logback-prod.xml @@ -0,0 +1,115 @@ + + + + + ${appName} + + + + + + + [requestId:%X{requestId}] [time:%d{yyyy-MM-dd HH:mm:ss.SSS}] [thread:%thread] [level:%-5level] [appName:${appName}] msg- %msg%n + + + + + + + ${LOG_HOME}/debug.log + + + + ${LOG_HOME}/debug.%d{yyyy-MM-dd}.%i.AssertLog.zip + + 30 + + 10GB + + + + 10MB + + + + + + [requestId:%X{requestId}] [time:%d{yyyy-MM-dd HH:mm:ss.SSS}] [thread:%thread] [level:%-5level] [appName:${appName}] msg- %msg%n + + + + + + + + INFO + ACCEPT + DENY + + + ${LOG_HOME}/info.log + + + + ${LOG_HOME}/info.%d{yyyy-MM-dd}.%i.AssertLog.zip + + 30 + + 10GB + + + + 100MB + + + + + + [requestId:%X{requestId}] [time:%d{yyyy-MM-dd HH:mm:ss.SSS}] [thread:%thread] [level:%-5level] [appName:${appName}] msg- %msg%n + + + + + + + + ERROR + ACCEPT + DENY + + ${LOG_HOME}/error.log + + + + ${LOG_HOME}/error.%d{yyyy-MM-dd}.%i.AssertLog.zip + + 30 + + 10GB + + + + 10MB + + + + + + [requestId:%X{requestId}] [time:%d{yyyy-MM-dd HH:mm:ss.SSS}] [thread:%thread] [level:%-5level] [appName:${appName}] msg- %msg%n + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/src/test/java/com/tongran/agenttcp/AgentTcpApplicationTests.java b/src/test/java/com/tongran/agenttcp/AgentTcpApplicationTests.java new file mode 100644 index 0000000..30b371a --- /dev/null +++ b/src/test/java/com/tongran/agenttcp/AgentTcpApplicationTests.java @@ -0,0 +1,13 @@ +package com.tongran.agenttcp; + +import org.junit.jupiter.api.Test; +import org.springframework.boot.test.context.SpringBootTest; + +@SpringBootTest +class AgentTcpApplicationTests { + + @Test + void contextLoads() { + } + +}