diff --git a/tongran-modules/pom.xml b/tongran-modules/pom.xml index 30b5e99..2312349 100644 --- a/tongran-modules/pom.xml +++ b/tongran-modules/pom.xml @@ -14,6 +14,7 @@ tongran-job tongran-file tongran-mtragent + tongran-frp tongran-modules diff --git a/tongran-modules/tongran-frp/pom.xml b/tongran-modules/tongran-frp/pom.xml new file mode 100644 index 0000000..05f3a47 --- /dev/null +++ b/tongran-modules/tongran-frp/pom.xml @@ -0,0 +1,116 @@ + + 4.0.0 + + com.tongran + tongran-modules + 3.6.6 + + + tongran-modules-frp + + + + UTF-8 + + + + + junit + junit + 3.8.1 + test + + + + org.apache.rocketmq + rocketmq-client + 4.9.0 + + + + + com.alibaba.cloud + spring-cloud-starter-alibaba-nacos-discovery + + + + + com.alibaba.cloud + spring-cloud-starter-alibaba-nacos-config + + + com.alibaba.nacos + nacos-client + + + + + com.alibaba.cloud + spring-cloud-starter-alibaba-sentinel + + + + + org.springframework.boot + spring-boot-starter-actuator + + + + + com.mysql + mysql-connector-j + + + + + com.tongran + tongran-common-log + + + + + com.tongran + tongran-common-swagger + + + + + com.tongran + tongran-common-security + + + org.projectlombok + lombok + + + + + org.yaml + snakeyaml + 1.28 + + + + + org.springframework.boot + spring-boot-starter-web + + + + ${project.artifactId} + + + org.springframework.boot + spring-boot-maven-plugin + + + + repackage + + + + + + + diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/TongRanFrpApplication.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/TongRanFrpApplication.java new file mode 100644 index 0000000..6516d91 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/TongRanFrpApplication.java @@ -0,0 +1,25 @@ +package com.tongran.frp; + +import com.tongran.common.security.annotation.EnableCustomConfig; +import com.tongran.common.security.annotation.EnableRyFeignClients; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.scheduling.annotation.EnableAsync; + +/** + * 平台管理模块 + * + * @author tongran + */ +@EnableCustomConfig +@EnableRyFeignClients +@SpringBootApplication +@EnableAsync +public class TongRanFrpApplication +{ + public static void main(String[] args) + { + SpringApplication.run(com.tongran.frp.TongRanFrpApplication.class, args); + System.out.println("(♥◠‿◠)ﾉﾞ TongRanFrp模块启动成功 ლ(´ڡ`ლ)ﾞ"); + } +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/config/ConsumerConfig.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/config/ConsumerConfig.java new file mode 100644 index 0000000..f4588f3 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/config/ConsumerConfig.java @@ -0,0 +1,77 @@ +package com.tongran.frp.config; + +import com.tongran.frp.consumer.RocketMsgListener; +import com.tongran.frp.enums.MessageTopic; +import com.tongran.frp.model.ConsumerMode; +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; +import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.common.consumer.ConsumeFromWhere; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cloud.context.config.annotation.RefreshScope; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import java.util.List; + +/** + * 消费者配置 + */ +@RefreshScope +@Configuration +@Slf4j +public class ConsumerConfig { + + @Autowired + private ConsumerMode consumerMode; + + @Autowired + private RocketMsgListener rocketMsgListener; + + @Bean + public DefaultMQPushConsumer getRocketMQConsumer() { + //构建客户端连接 + DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerMode.getAgentGroup()); + // + consumer.setNamesrvAddr(consumerMode.getNamesrvAddr()); + consumer.setConsumeThreadMin(consumerMode.getConsumeThreadMin()); + consumer.setConsumeThreadMax(consumerMode.getConsumeThreadMax()); + consumer.registerMessageListener(rocketMsgListener); + /** + * 1. CONSUME_FROM_LAST_OFFSET：第一次启动从队列最后位置消费，后续再启动接着上次消费的进度开始消费 + * 2. CONSUME_FROM_FIRST_OFFSET：第一次启动从队列初始位置消费，后续再启动接着上次消费的进度开始消费 + * 3. CONSUME_FROM_TIMESTAMP：第一次启动从指定时间点位置消费，后续再启动接着上次消费的进度开始消费 + * 以上所说的第一次启动是指从来没有消费过的消费者，如果该消费者消费过，那么会在broker端记录该消费者的消费位置，如果该消费者挂了再启动，那么自动从上次消费的进度开始 + */ + consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); + /** + * CLUSTERING (集群模式) ：默认模式，同一个ConsumerGroup(groupName相同)每个consumer只消费所订阅消息的一部分内容，同一个ConsumerGroup里所有的Consumer消息加起来才是所 + * 订阅topic整体，从而达到负载均衡的目的 + * BROADCASTING (广播模式) ：同一个ConsumerGroup每个consumer都消费到所订阅topic所有消息，也就是一个消费会被多次分发，被多个consumer消费。 + * 需要注意的是，在广播模式下，每个Consumer都会独立地处理相同的消息副本。这可能会导致一些潜在的问题，例如消息重复处理或者资源浪费。因此，在使用广播模式时，请确保消息的处理逻辑是幂等的，并仔细考虑系统资源的消耗。 + */ + // consumer.setMessageModel(MessageModel.BROADCASTING); + + consumer.setVipChannelEnabled(false); + consumer.setConsumeMessageBatchMaxSize(consumerMode.getConsumeMessageBatchMaxSize()); + try { + /** + * 订阅topic，可以对指定消息进行过滤，例如："TopicTest","tagl||tag2||tag3",*或null表示topic所有消息 + * 由于官方并没有给直接订阅全部消息示例 所以使用list列表循环订阅所有topic + */ + // 获取所有topic列表 + MessageTopic messageTopic = new MessageTopic(); + List allTopics = messageTopic.RocketMQTopicList(); + //订阅所有topic + for (String topic : allTopics) { + consumer.subscribe(topic,"*"); + } + consumer.start(); + log.info("消费者初始化成功：{}", consumer); + } catch (MQClientException e) { + e.printStackTrace(); + log.error("消费者初始化失败：{}",e.getMessage()); + } + return consumer; + } +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/config/ProducerConfig.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/config/ProducerConfig.java new file mode 100644 index 0000000..54fa8c9 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/config/ProducerConfig.java @@ -0,0 +1,56 @@ +package com.tongran.frp.config; + +import com.tongran.frp.model.ProducerMode; +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.client.producer.DefaultMQProducer; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + + +/** + * mq搭建地址连接 + * 生产者初者连接信息 具体看nacos配置 + */ +@Configuration +@Slf4j +public class ProducerConfig { + + /** + * 远程调用连接信息 + */ + public static DefaultMQProducer producer; + + /** + * 连接客户端信息配置 具体看nacos配置 + */ + @Autowired + private ProducerMode producerMode; + + @Bean + public DefaultMQProducer getRocketMQProducer() { + producer = new DefaultMQProducer(producerMode.getAgentGroup()); + producer.setNamesrvAddr(producerMode.getNamesrvAddr()); + //如果需要同一个jvm中不同的producer往不同的mq集群发送消息，需要设置不同的instanceName + if(producerMode.getMaxMessageSize()!=null){ + producer.setMaxMessageSize(producerMode.getMaxMessageSize()); + } + if(producerMode.getSendMsgTimeout()!=null){ + producer.setSendMsgTimeout(producerMode.getSendMsgTimeout()); + } + //如果发送消息失败，设置重试次数，默认为2次 + if(producerMode.getRetryTimesWhenSendFailed()!=null){ + producer.setRetryTimesWhenSendFailed(producerMode.getRetryTimesWhenSendFailed()); + } + producer.setVipChannelEnabled(false); + try { + producer.start(); + log.info("生产者初始化成功：{}",producer.toString()); + } catch (MQClientException e) { + log.error("生产者初始化失败：{}",e.getMessage()); + } + return producer; + } + +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/consumer/RocketMsgListener.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/consumer/RocketMsgListener.java new file mode 100644 index 0000000..15a5485 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/consumer/RocketMsgListener.java @@ -0,0 +1,94 @@ +package com.tongran.frp.consumer; + +import com.alibaba.fastjson.JSON; +import com.tongran.frp.domain.DeviceMessage; +import com.tongran.frp.enums.MessageCodeEnum; +import com.tongran.frp.handler.MessageHandler; +import com.tongran.frp.producer.ConsumeException; +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; +import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; +import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; +import org.apache.rocketmq.common.message.MessageExt; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; +import org.springframework.util.CollectionUtils; + +import java.io.UnsupportedEncodingException; +import java.util.List; + +/** + * 消息监听 + */ +@Slf4j +@Component +public class RocketMsgListener implements MessageListenerConcurrently { + @Autowired + private MessageHandler messageHandler; + + /** + * 消费消息 + * @param list msgs.size() >= 1 + * DefaultMQPushConsumer.consumeMessageBatchMaxSize=1，you can modify here + * 这里只设置为1，当设置为多个时，list中只要有一条消息消费失败，就会整体重试 + * @param consumeConcurrentlyContext 上下文信息 + * @return 消费状态 成功（CONSUME_SUCCESS）或者 重试 (RECONSUME_LATER) + */ + @Override + public ConsumeConcurrentlyStatus consumeMessage(List list, ConsumeConcurrentlyContext consumeConcurrentlyContext) { + try{ + //消息不等于空情况 + if (!CollectionUtils.isEmpty(list)) { + //获取topic + for (MessageExt messageExt : list) { + // 解析消息内容 + String body = new String(messageExt.getBody()); + log.info("接受到的消息为：{}", body); + String tags = messageExt.getTags(); + String topic = messageExt.getTopic(); + String msgId = messageExt.getMsgId(); + String keys = messageExt.getKeys(); + int reConsume = messageExt.getReconsumeTimes(); + // 消息已经重试了3次，如果不需要再次消费，则返回成功 + if (reConsume == 3) { + // TODO 补偿信息 + log.error("消息消费三次失败，消息内容：{}", body); + return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//根据业务返回是否正常 + } + if(MessageCodeEnum.TR_MTRAGENT_UP.getCode().equals(topic)){ + // 拿到信息 + DeviceMessage message = JSON.parseObject(body, DeviceMessage.class); + // 处理消息 + messageHandler.handleMessage(message); + return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//业务处理成功 + } + // 根据不同的topic处理不同的业务 这里以订单消息为例子 + } + } + // 消息消费失败 + //broker会根据设置的messageDelayLevel发起重试，默认16次 + return ConsumeConcurrentlyStatus.RECONSUME_LATER; + } catch (Exception e) { + // 调用 handleException 方法处理异常并返回处理结果 + return handleException(e); + } + } + + /** + * 异常处理 + * + * @param e 捕获的异常 + * @return 消息消费结果 + */ + private static ConsumeConcurrentlyStatus handleException(final Exception e) { + Class exceptionClass = e.getClass(); + if (exceptionClass.equals(UnsupportedEncodingException.class)) { + log.error(e.getMessage()); + } else if (exceptionClass.equals(ConsumeException.class)) { + log.error(e.getMessage()); + } else{ + log.error(e.getMessage()); + } + return ConsumeConcurrentlyStatus.RECONSUME_LATER; + } +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/consumer/RocketMsgTransactionListenerImpl.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/consumer/RocketMsgTransactionListenerImpl.java new file mode 100644 index 0000000..b89c8d4 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/consumer/RocketMsgTransactionListenerImpl.java @@ -0,0 +1,30 @@ +package com.tongran.frp.consumer; + +import org.apache.rocketmq.client.producer.LocalTransactionState; +import org.apache.rocketmq.client.producer.TransactionListener; +import org.apache.rocketmq.common.message.Message; +import org.apache.rocketmq.common.message.MessageExt; + +/** + * 事物消息监听 + */ +public class RocketMsgTransactionListenerImpl implements TransactionListener { + + @Override + public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { + // 在这里执行本地事务，比如数据库操作等 + // 如果本地事务执行成功，返回 COMMIT_MESSAGE + // 如果本地事务执行失败，返回 ROLLBACK_MESSAGE + // 如果本地事务执行中，可以返回 UNKNOW，RocketMQ 将会检查事务状态，并根据状态处理消息 + return LocalTransactionState.COMMIT_MESSAGE; // 根据实际情况返回对应的状态 + } + + @Override + public LocalTransactionState checkLocalTransaction(MessageExt msg) { + // 检查本地事务状态，如果本地事务执行成功，返回 COMMIT_MESSAGE + // 如果本地事务执行失败，返回 ROLLBACK_MESSAGE + // 如果本地事务仍在执行中，返回 UNKNOW，RocketMQ 将会继续检查事务状态 + return LocalTransactionState.COMMIT_MESSAGE; // 根据实际情况返回对应的状态 + } + +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/controller/RocketMqController.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/controller/RocketMqController.java new file mode 100644 index 0000000..82a2f9e --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/controller/RocketMqController.java @@ -0,0 +1,139 @@ +package com.tongran.frp.controller; + + +import com.tongran.common.security.annotation.InnerAuth; +import com.tongran.frp.producer.MessageProducer; +import org.apache.rocketmq.client.producer.SendResult; +import org.apache.rocketmq.common.message.Message; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.RestController; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * 消息测试类Controller + */ +@RestController +@RequestMapping("/api/rocketMessage") +public class RocketMqController { + + + + /** + * 发送同步消息 + */ + @PostMapping("/sendSynchronizeMessage") + private Map sendSynchronizeMessage(){ + MessageProducer messageProducer = new MessageProducer(); + //调用MessageProducer配置好的消息方法 + SendResult sendResult = messageProducer.sendSynchronizeMessage("order-message","order_message_tag","title","content"); + Map result = new HashMap<>(); + result.put("data",sendResult); + return result; + } + + + + /** + * 发送单向消息 + */ + @PostMapping("/sendOnewayMessage") + private Map sendOnewayMessage(@RequestParam("topic") String topic,@RequestParam("tag") String tag,@RequestParam("key") String key,@RequestParam("value") String value){ + MessageProducer messageProducer = new MessageProducer(); + //调用MessageProducer配置好的消息方法 topic需要你根据你们业务定制相应的 + messageProducer.sendOnewayMessage("order-message","order_timeout_tag","title","content"); + Map result = new HashMap<>(); + result.put("msg","发送成功"); + result.put("code",200); + return result; + } + + + /** + * 批量发送消息 + */ + @PostMapping("/sendBatchMessage") + private Map sendBatchMessage(){ + // 根据实际需求创建消息列表并返回 + List messages = new ArrayList<>(); + // 添加消息到列表 + messages.add(new Message("order-message", "order_timeout_tag", "Message 1".getBytes())); + messages.add(new Message("order-message", "order_timeout_tag", "Message 2".getBytes())); + messages.add(new Message("order-message", "order_timeout_tag", "Message 3".getBytes())); + MessageProducer messageProducer = new MessageProducer(); + //调用MessageProducer配置好的消息方法 topic需要你根据你们业务定制相应的 + SendResult sendResult = messageProducer.sendBatchMessage(messages); + Map result = new HashMap<>(); + result.put("data",sendResult); + return result; + } + + + /** + * 发送事物消息 + */ + @PostMapping("/sendThingMessage") + private Map sendThingMessage(@RequestParam("topic") String topic,@RequestParam("tag") String tag,@RequestParam("key") String key,@RequestParam("value") String value){ + MessageProducer messageProducer = new MessageProducer(); + //调用MessageProducer配置好的消息方法 topic需要你根据你们业务定制相应的 + SendResult sendResult = messageProducer.sendThingMessage("order-message","order_timeout_tag","title","content"); + Map result = new HashMap<>(); + result.put("data",sendResult); + return result; + } + + + /** + * 发送有序的消息 + */ + @PostMapping("/sendOrderlyMessage") + private Map sendOrderlyMessage(){ + // 根据实际需求创建消息列表并返回 + List messages = new ArrayList<>(); + // 添加消息到列表 + messages.add(new Message("order-message", "order_timeout_tag", "Message 1".getBytes())); + messages.add(new Message("order-message", "order_timeout_tag", "Message 2".getBytes())); + messages.add(new Message("order-message", "order_timeout_tag", "Message 3".getBytes())); + Integer messageQueueNumber = 3; + MessageProducer messageProducer = new MessageProducer(); + //调用MessageProducer配置好的消息方法 topic需要你根据你们业务定制相应的 + SendResult sendResult = messageProducer.sendOrderlyMessage(messages,messageQueueNumber); + Map result = new HashMap<>(); + result.put("data",sendResult); + return result; + } + + /** + * 发送延迟消息 + */ + @PostMapping("/sendDelayMessage") + private Map sendDelayMessage(@RequestParam("topic") String topic,@RequestParam("tag") String tag,@RequestParam("key") String key,@RequestParam("value") String value){ + MessageProducer messageProducer = new MessageProducer(); + //调用MessageProducer配置好的消息方法 topic需要你根据你们业务定制相应的 + SendResult sendResult = messageProducer.sendDelayMessage("order-message","order_timeout_tag","title","content",4); + Map result = new HashMap<>(); + result.put("data",sendResult); + return result; + } + + + /** + * 发送异步的消息 + */ + @InnerAuth + @PostMapping("/sendAsyncProducerMessage") + private Map sendAsyncProducerMessage(@RequestParam("topic") String topic,@RequestParam("tag") String tag,@RequestParam("key") String key,@RequestParam("value") String value){ + MessageProducer messageProducer = new MessageProducer(); + //调用MessageProducer配置好的消息方法 topic需要你根据你们业务定制相应的 + SendResult sendResult = messageProducer.sendAsyncProducerMessage(topic,tag,key,value); + Map result = new HashMap<>(); + result.put("data",sendResult); + return result; + } + +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/DeviceMessage.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/DeviceMessage.java new file mode 100644 index 0000000..9f9e747 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/DeviceMessage.java @@ -0,0 +1,10 @@ +package com.tongran.frp.domain; + +import lombok.Data; + +@Data +public class DeviceMessage { + private String clientId; + private String dataType; + private String data; +} \ No newline at end of file diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/AgentUpdateMsgVo.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/AgentUpdateMsgVo.java new file mode 100644 index 0000000..e56c7b8 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/AgentUpdateMsgVo.java @@ -0,0 +1,15 @@ +package com.tongran.frp.domain.vo; + +import lombok.Data; + +@Data +public class AgentUpdateMsgVo { + /**文件地址，外网HTTP（S）地址 */ + private String fileUrl; + /** 文件MD5 */ + private String fileMd5; + /** 执行方式：0、立即执行；1、定时执行； */ + private Integer method; + /** 定时时间，执行方式为1、定时执行时该字段必传 */ + private long policyTime; +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/HopInfoVO.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/HopInfoVO.java new file mode 100644 index 0000000..aa99f9f --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/HopInfoVO.java @@ -0,0 +1,16 @@ +package com.tongran.frp.domain.vo; + +import lombok.Data; + +@Data +public class HopInfoVO { + private int hopNumber; // 跳数 + private String ipAddress; // IP地址 + private String hostname; // 主机名（如果有） + private double lossPercent; // 丢包率 + private double avgLatency; // 平均延迟(ms) + private String country; // 国家 + private String province; // 省份 + private String city; // 城市 + private String isp; // 运营商 +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/MessageVo.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/MessageVo.java new file mode 100644 index 0000000..771d78f --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/MessageVo.java @@ -0,0 +1,13 @@ +package com.tongran.frp.domain.vo; + +import lombok.Data; + +@Data +public class MessageVo { + + private String clientId; + + private String dataType; + + private String data; +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/MtrResultVo.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/MtrResultVo.java new file mode 100644 index 0000000..1acd0ab --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/MtrResultVo.java @@ -0,0 +1,19 @@ +package com.tongran.frp.domain.vo; + +import lombok.Data; + +import java.util.List; + +@Data +public class MtrResultVo { + private String targetIp; // 目标IP + private String clientId; // 目标IP + private Long policyId; // 策略ID + private double firstLossPercent; // 第一次探测丢包率 + private double finalLossPercent; // 最终丢包率 + private long timestamp; // 探测时间戳 + private boolean hasRetry; // 是否重试 + private String errorMsg; // 错误信息 + private List hopInfos; // 每跳路由信息列表 + +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/PolicyTypeVo.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/PolicyTypeVo.java new file mode 100644 index 0000000..9de648a --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/PolicyTypeVo.java @@ -0,0 +1,22 @@ +package com.tongran.frp.domain.vo; + +import lombok.Data; + +import java.time.Instant; + +@Data +public class PolicyTypeVo { + + /** 服务器监控策略信息 */ + private String monitors; + /** 服务器脚本策略信息 */ + private String scripts; + /** agent更新信息 */ + private String versions; + /** 路由信息 */ + private String routes; + /** mtr策略 */ + private String mtrPolicys; + /** 时间戳 */ + private Long timestamp = Instant.now().getEpochSecond(); +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/PolicyVo.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/PolicyVo.java new file mode 100644 index 0000000..bcb9e1a --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/PolicyVo.java @@ -0,0 +1,14 @@ +package com.tongran.frp.domain.vo; + +import lombok.Data; + +import java.time.Instant; +import java.util.List; + +@Data +public class PolicyVo { + /** 更新时间戳 */ + private Long upTime = Instant.now().getEpochSecond(); + /** 更新内容 */ + private List contents; +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/RmMtrPolicyConfigVo.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/RmMtrPolicyConfigVo.java new file mode 100644 index 0000000..70d3f8f --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/RmMtrPolicyConfigVo.java @@ -0,0 +1,31 @@ +package com.tongran.frp.domain.vo; + +import com.fasterxml.jackson.annotation.JsonFormat; +import lombok.Data; + +import java.util.Date; +import java.util.List; +import java.util.Map; + +/** + * mtr探测策略配置对象 rm_mtr_policy_config + * + * @author gyt + * @date 2025-11-18 + */ +@Data +public class RmMtrPolicyConfigVo +{ + /** 开始时间 */ + @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss") + private Date startTime; + + /** 结束时间 */ + @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss") + private Date endTime; + + /** 探测频率(秒) */ + private Long probeFrequency; + /** clientId和ip对应集合 */ + private Map> clientIdToIpsMap; +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/RspVo.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/RspVo.java new file mode 100644 index 0000000..eb1a3ef --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/domain/vo/RspVo.java @@ -0,0 +1,29 @@ +package com.tongran.frp.domain.vo; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import lombok.Data; + +import java.time.Instant; + +@Data +@JsonIgnoreProperties(ignoreUnknown = true) +public class RspVo { + /** + * 状态码，0、失败；1、成功 + */ + private Integer resCode; + /** + * 描述 + */ + private String resMag; + /** + * 描述 + */ + private String resMsg; + + private String result; + /** + * 时间戳 + */ + private Long timestamp = Instant.now().getEpochSecond(); +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/enums/AlarmTypeEnum.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/enums/AlarmTypeEnum.java new file mode 100644 index 0000000..8a4c68e --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/enums/AlarmTypeEnum.java @@ -0,0 +1,16 @@ +package com.tongran.frp.enums; + +import lombok.Getter; + +@Getter +public enum AlarmTypeEnum { + 服务器下线("1", "服务器下线"), + 交换机下线("2", "交换机下线"), + mtrAgent下线("3", "mtrAgent下线"); + private final String code; + private final String msg; + AlarmTypeEnum(String code, String msg){ + this.code = code; + this.msg = msg; + } +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/enums/MessageCodeEnum.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/enums/MessageCodeEnum.java new file mode 100644 index 0000000..32f51a0 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/enums/MessageCodeEnum.java @@ -0,0 +1,67 @@ +package com.tongran.frp.enums; + + +import lombok.Getter; + +/** + * 用于传递topic和 tag + * 也用于接收消息后判断不同的消息处理不同的业务 + */ +@Getter +public enum MessageCodeEnum { + + /** + * agent数据采集的信息 + */ + AGENT_MESSAGE_TOPIC("agent_up","agent数据采集的信息topic"), + + TONGRAN_AGENT_UP("tongran_agent_up","agent数据采集的信息topic"), + + TR_AGENT_UP("tr_agent_up","agent数据采集的信息topic v1.1"), + + TR_MTRAGENT_UP("tr_mtragent_up","mtragent监测丢包率的信息topic"), + + /** + * 系统消息 + */ + NOTE_MESSAGE_TOPIC("system-message","系统消息服务模块topic名称"), + /** + * 用户消息 + */ + USER_MESSAGE_TOPIC("user-message","用户消息服务模块topic名称"), + + /** + * 订单消息 + */ + ORDER_MESSAGE_TOPIC("order-message","订单消息服务模块topic名称"), + + /** + * 用户消息tag + */ + USER_MESSAGE_TAG("user_message_tag","用户消息推送"), + + /** + * 系统消息tag + */ + NOTE_MESSAGE_TAG("system_message_tag","系统消息推送"), + + /** + * 订单消息 + */ + ORDER_MESSAGE_TAG("order_message_tag","订单消息推送"), + + /** + * 订单处理编号 + */ + ORDER_TIMEOUT_TAG("order_timeout_tag","订单超时处理"); + + + private final String code; + private final String msg; + + MessageCodeEnum(String code, String msg){ + this.code = code; + this.msg = msg; + } + +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/enums/MessageTopic.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/enums/MessageTopic.java new file mode 100644 index 0000000..e2f9d00 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/enums/MessageTopic.java @@ -0,0 +1,21 @@ +package com.tongran.frp.enums; + + +import java.util.ArrayList; +import java.util.List; + +/** + * 定义topic列表 + */ +public class MessageTopic { + + //在这里添加topic 用于批量订阅 + public List RocketMQTopicList(){ + List getTopicLists=new ArrayList<>(); + // agent采集消息 +// getTopicLists.add("agent_up"); + getTopicLists.add("tr_frp_up"); + return getTopicLists; + } + +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/enums/PushMethodEnum.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/enums/PushMethodEnum.java new file mode 100644 index 0000000..0e1b15b --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/enums/PushMethodEnum.java @@ -0,0 +1,14 @@ +package com.tongran.frp.enums; + +import lombok.Getter; + +@Getter +public enum PushMethodEnum { + 企业微信("1", "企业微信"); + private final String code; + private final String msg; + PushMethodEnum(String code, String msg){ + this.code = code; + this.msg = msg; + } +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/enums/ServerLogoEnum.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/enums/ServerLogoEnum.java new file mode 100644 index 0000000..31a9a76 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/enums/ServerLogoEnum.java @@ -0,0 +1,21 @@ +package com.tongran.frp.enums; + +import lombok.Getter; + +@Getter +public enum ServerLogoEnum { + 交换卷文件的可用空间("systemSwapSizeFreeCollect", "交换卷/文件的可用空间（字节）"), + 内存利用率("memoryUtilizationCollect", "内存利用率"), + 可用交换空间百分比("systemSwapSizePercentCollect", "可用交换空间百分比"), + 可用内存("memorySizeAvailableCollect", "可用内存"), + 可用内存百分比("memorySizePercentCollect", "可用内存百分比"), + 正在运行的进程数("procNumRunCollect", "正在运行的进程数"), + 登录用户数("systemUsersNumCollect", "登录用户数"), + 进程数("procNumCollect", "进程数"); + private final String code; + private final String msg; + ServerLogoEnum(String code, String msg){ + this.code = code; + this.msg = msg; + } +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/enums/SwitchLogo.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/enums/SwitchLogo.java new file mode 100644 index 0000000..89aa563 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/enums/SwitchLogo.java @@ -0,0 +1,17 @@ +package com.tongran.frp.enums; + +import lombok.Getter; + +@Getter +public enum SwitchLogo { + 设备CPU使用率("hwEntityCpuUsage", "设备CPU使用率"), + 设备内存使用率("hwEntityMemUsage", "设备内存使用率(%)"), + 系统平均功率("hwAveragePower", "系统平均功率(%)"), + 系统实时功率("hwCurrentPower", "系统实时功率(%)"); + private final String code; + private final String msg; + SwitchLogo(String code, String msg){ + this.code = code; + this.msg = msg; + } +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/handler/MessageHandler.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/handler/MessageHandler.java new file mode 100644 index 0000000..e1de0b0 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/handler/MessageHandler.java @@ -0,0 +1,114 @@ +package com.tongran.frp.handler; + +import com.tongran.common.core.enums.MsgEnum; +import com.tongran.frp.domain.DeviceMessage; +import com.tongran.frp.domain.vo.RspVo; +import com.tongran.frp.model.ProducerMode; +import com.tongran.frp.producer.MessageProducer; +import com.tongran.frp.utils.JsonDataParser; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.scheduling.annotation.EnableScheduling; +import org.springframework.stereotype.Component; + +import javax.annotation.PostConstruct; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.function.Consumer; + +/** + * 设备消息处理器 + */ +@Slf4j +@Component +@EnableScheduling +public class MessageHandler { + + private final Map> messageHandlers = new HashMap<>(); + // 心跳状态 + private static final String HEARTBEAT_STATUS_PREFIX = "mtr:heartbeat:status:"; + // 心跳时间 + private static final String HEARTBEAT_TIME_PREFIX = "mtr:heartbeat:time:"; + // 心跳告警 + private static final String HEARTBEAT_ALERT_PREFIX = "mtr:heartbeat:alert:"; + String HEARTBEAT_RECOVERY_COUNT_PREFIX = "mtr:heartbeat:recovery:count:"; + + String HEARTBEAT_COUNT_PREFIX = "mtr:heartbeat:count:"; + private static final long HEARTBEAT_TIMEOUT = 30000; // 3分钟超时 + + + @Autowired + private RedisTemplate redisTemplate; + @Autowired + private ProducerMode producerMode; + + + /** + * 初始化处理器映射 + */ + @PostConstruct + public void init() { + registerHandler(MsgEnum.Agent版本更新应答.getValue(), this::handleAgentUpdateRspMessage); + + // 其他类型消息可以单独注册处理器 + registerHandler(MsgEnum.注册.getValue(), this::handleRegisterMessage); + registerHandler(MsgEnum.获取最新策略.getValue(), this::handleNewPolicyMessage); + } + + + private void handleRegisterMessage(DeviceMessage message) { + MessageProducer messageProducer = new MessageProducer(); + } + /** + * agent更新响应 + * @param message + */ + private void handleAgentUpdateRspMessage(DeviceMessage message) { + List rspVoList = JsonDataParser.parseJsonData(message.getData(), RspVo.class); + if (!rspVoList.isEmpty()) { + RspVo rsp = rspVoList.get(0); + if(rsp.getResCode() == 1){ + }else{ + } + } + } + + /** + * 注册消息处理器 + */ + private void registerHandler(String dataType, Consumer handler) { + messageHandlers.put(dataType, handler); + } + + /** + * 处理设备消息（对外暴露的主方法） + */ + public void handleMessage(DeviceMessage message) { + String dataType = message.getDataType(); + Consumer handler = messageHandlers.get(dataType); + + if (handler != null) { + handler.accept(message); + } else { + log.warn("未知数据类型：{}", dataType); + } + } + + // ========== 具体的消息处理方法 ========== + /** + * 获取最新策略 + * @param deviceMessage + */ + private void handleNewPolicyMessage(DeviceMessage deviceMessage) { + MessageProducer messageProducer = new MessageProducer(); + } + + + // 更新资源状态的公共方法 + private void updateResourceStatus(String clientId, String status) { + log.info("开启更新资源状态========"); + } + +} \ No newline at end of file diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/model/ConsumerMode.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/model/ConsumerMode.java new file mode 100644 index 0000000..7123120 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/model/ConsumerMode.java @@ -0,0 +1,30 @@ +package com.tongran.frp.model; + +import lombok.Data; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Configuration; +import org.springframework.stereotype.Component; + +/** + * 消费者初始化 + * 消费者连接信息 具体看nacos配置 + */ +@Data +@Configuration +@Component +public class ConsumerMode { + @Value("${suning.rocketmq.namesrvAddr}") + private String namesrvAddr; + @Value("${suning.rocketmq.conumer.groupName}") + private String groupName ; + @Value("${suning.rocketmq.conumer.consumeThreadMin}") + private int consumeThreadMin; + @Value("${suning.rocketmq.conumer.consumeThreadMax}") + private int consumeThreadMax; + @Value("${suning.rocketmq.conumer.consumeMessageBatchMaxSize}") + private int consumeMessageBatchMaxSize; + @Value("${suning.rocketmq.conumer.agentTopic}") + private String agentTopic; + @Value("${suning.rocketmq.conumer.agentGroup}") + private String agentGroup; +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/model/ProducerMode.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/model/ProducerMode.java new file mode 100644 index 0000000..9d88259 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/model/ProducerMode.java @@ -0,0 +1,29 @@ +package com.tongran.frp.model; + +import lombok.Data; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.cloud.context.config.annotation.RefreshScope; +import org.springframework.context.annotation.Configuration; + +/** + * 生产者初始化 + */ +@RefreshScope +@Data +@Configuration +public class ProducerMode { + @Value("${suning.rocketmq.producer.groupName}") + private String groupName; + @Value("${suning.rocketmq.producer.agentTopic}") + private String agentTopic; + @Value("${suning.rocketmq.producer.agentGroup}") + private String agentGroup; + @Value("${suning.rocketmq.namesrvAddr}") + private String namesrvAddr; + @Value("${suning.rocketmq.producer.maxMessageSize}") + private Integer maxMessageSize; + @Value("${suning.rocketmq.producer.sendMsgTimeout}") + private Integer sendMsgTimeout; + @Value("${suning.rocketmq.producer.retryTimesWhenSendFailed}") + private Integer retryTimesWhenSendFailed; +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/producer/ConsumeException.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/producer/ConsumeException.java new file mode 100644 index 0000000..0094dd8 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/producer/ConsumeException.java @@ -0,0 +1,23 @@ +package com.tongran.frp.producer; + +/** + * @author 影子 + * 用于捕捉异常非受检异常（unchecked exception） + * RuntimeException 和其子类的异常在编译时不需要进行强制性的异常处理，可以选择在运行时进行捕获和处理 + * 可选择使用 + */ +public class ConsumeException extends RuntimeException{ + private static final long serialVersionUID = 4093867789628938836L; + + public ConsumeException(String message) { + super(message); + } + + public ConsumeException(Throwable cause) { + super(cause); + } + + public ConsumeException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/producer/MessageProducer.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/producer/MessageProducer.java new file mode 100644 index 0000000..1a102c0 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/producer/MessageProducer.java @@ -0,0 +1,225 @@ +package com.tongran.frp.producer; + +import com.alibaba.fastjson.JSON; +import com.tongran.frp.consumer.RocketMsgTransactionListenerImpl; +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.client.exception.MQBrokerException; +import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.client.producer.SendCallback; +import org.apache.rocketmq.client.producer.SendResult; +import org.apache.rocketmq.client.producer.TransactionMQProducer; +import org.apache.rocketmq.client.producer.TransactionSendResult; +import org.apache.rocketmq.common.message.Message; +import org.apache.rocketmq.remoting.common.RemotingHelper; +import org.apache.rocketmq.remoting.exception.RemotingException; + +import java.io.UnsupportedEncodingException; +import java.text.DateFormat; +import java.text.SimpleDateFormat; +import java.util.Date; +import java.util.List; + +import static com.tongran.frp.config.ProducerConfig.producer; + + +/** + * 消息发送 + */ +@Slf4j +public class MessageProducer { + + + /** + * 同步发送消息 + * @param topic 主题 + * @param tag 标签 + * @param key 自定义的key，根据业务来定 + * @param value 消息的内容 + * 通过调用 send() 方法发送消息，阻塞等待服务器响应。 + */ + public SendResult sendSynchronizeMessage(String topic, String tag, String key, String value){ + String body = "topic：【"+topic+"】, tag：【"+tag+"】, key：【"+key+"】, value：【"+value+"】"; + try { + Message msg = new Message(topic,tag,key, value.getBytes(RemotingHelper.DEFAULT_CHARSET)); + System.out.println("生产者发送消息:"+ JSON.toJSONString(value)); + SendResult result = producer.send(msg); + return result; + } catch (UnsupportedEncodingException e) { + log.error("消息初始化失败！body：{}",body); + + } catch (MQClientException | InterruptedException | RemotingException | MQBrokerException e) { + log.error("消息发送失败! body：{}",body); + } + return null; + } + + /** + * 单向发送消息 + * @param topic 主题 + * @param tag 标签 + * @param key 自定义的key，根据业务来定 + * @param value 消息的内容 + * 单向发送：通过调用 sendOneway() 方法发送消息，不关心发送结果，适用于对可靠性要求不高的场景。 + */ + public void sendOnewayMessage(String topic, String tag, String key, String value){ + String body = "topic：【"+topic+"】, tag：【"+tag+"】, key：【"+key+"】, value：【"+value+"】"; + try { + Message msg = new Message(topic,tag,key, value.getBytes(RemotingHelper.DEFAULT_CHARSET)); + System.out.println("生产者发送消息:"+ JSON.toJSONString(value)); + producer.sendOneway(msg); + } catch (UnsupportedEncodingException e) { + log.error("消息初始化失败！body：{}",body); + + } catch (MQClientException | InterruptedException | RemotingException e) { + log.error("消息发送失败! body：{}",body); + } + } + + + /** + * 批量发送消息 + * @param messages 消息列表 + * 批量发送：通过调用 send() 方法并传入多条消息，实现批量发送消息。 + */ + public SendResult sendBatchMessage(List messages){ + String body = messages.toString(); + try { + System.out.println("生产者发送消息:"+ messages); + // 发送批量消息 + SendResult sendResult = producer.send(messages); + return sendResult; + } catch (MQClientException | InterruptedException | RemotingException e) { + log.error("消息发送失败! body：{}",body); + } catch (MQBrokerException e) { + throw new RuntimeException(e); + } + return null; + } + + + /** + * 事务消息发送 + * @param topic 主题 + * @param tag 标签 + * @param key 自定义的key，根据业务来定 + * @param value 消息的内容 + * 事务消息发送：通过使用事务监听器实现本地事务执行和消息发送的一致性。 + */ + public SendResult sendThingMessage(String topic, String tag, String key, String value){ + String body = "topic：【"+topic+"】, tag：【"+tag+"】, key：【"+key+"】, value：【"+value+"】"; + try { + // 实例化事务生产者 + TransactionMQProducer transactionMQProducer = new TransactionMQProducer(producer.getProducerGroup()); + transactionMQProducer.setNamesrvAddr(producer.getNamesrvAddr()); + // 设置事务监听器 + transactionMQProducer.setTransactionListener(new RocketMsgTransactionListenerImpl()); + Message msg = new Message(topic,tag,key, value.getBytes(RemotingHelper.DEFAULT_CHARSET)); + System.out.println("生产者发送消息:"+ JSON.toJSONString(value)); + // 发送事务消息 + TransactionSendResult sendResult = transactionMQProducer.sendMessageInTransaction(msg, null); + return sendResult; + } catch (UnsupportedEncodingException e) { + log.error("消息初始化失败！body：{}",body); + + } catch (MQClientException e) { + log.error("消息发送失败! body：{}",body); + } + return null; + } + + + /** + * 发送有序的消息 + * @param messagesList Message集合 + * @param messageQueueNumber 消息队列数量，根据实际情况设定 + * 顺序发送： messageQueueNumber 表示消息的业务标识，可以根据具体需求进行设置来保证消息按顺序发送。 + */ + public SendResult sendOrderlyMessage(List messagesList, Integer messageQueueNumber) { + SendResult result = null; + for (Message message : messagesList) { + try { + result = producer.send(message, (list, msg, arg) -> { + Integer queueNumber = (Integer) arg; + //int queueIndex = queueNumber % list.size(); + return list.get(queueNumber); + }, messageQueueNumber);//根据编号取模，选择消息队列 + } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) { + log.error("发送有序消息失败"); + return result; + } + } + return result; + } + + /** + * 发送延迟消息 + * @param topic 主题 + * @param tag 标签 + * @param key 自定义的key，根据业务来定 + * @param value 消息的内容 + * 延迟发送：通过设置延迟级别来实现延迟发送消息。 + */ + public SendResult sendDelayMessage(String topic, String tag, String key, String value, Integer level) + { + SendResult result = null; + try + { + Message msg = new Message(topic,tag,key, value.getBytes(RemotingHelper.DEFAULT_CHARSET)); + log.info("生产者发送消息:"+ JSON.toJSONString(value)); + //设置消息延迟级别，我这里设置5，对应就是延时一分钟 + // "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h" + msg.setDelayTimeLevel(level); + // 发送消息到一个Broker + result = producer.send(msg); + // 通过sendResult返回消息是否成功送达 + log.info("发送延迟消息结果：======sendResult：{}", result); + DateFormat format =new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); + log.info("发送时间：{}", format.format(new Date())); + return result; + } + catch (Exception e) + { + e.printStackTrace(); + log.error("延迟消息队列推送消息异常:{},推送内容:{}", e.getMessage(), result); + } + return result; + } + /** + * 发送异步的消息 + * @param topic 主题 + * @param tag 标签 + * @param key 自定义的key，根据业务来定 + * @param value 消息的内容 + * 通过调用 send() 方法，并传入一个 SendCallback 对象，在发送消息的同时可以继续处理其他逻辑，消息发送结果通过回调函数通知。 + */ + public SendResult sendAsyncProducerMessage(String topic, String tag, String key, String value){ + + try { + //创建一个消息实例，指定主题、标签和消息体。 + Message msg = new Message(topic,tag,key, value.getBytes(RemotingHelper.DEFAULT_CHARSET)); + log.info("生产者发送消息:"+ JSON.toJSONString(value)); + producer.send(msg,new SendCallback() { + // 异步回调的处理 + @Override + public void onSuccess(SendResult sendResult) { + System.out.printf("%-10d 异步发送消息成功 %s %n", msg, sendResult.getMsgId()); + } + @Override + public void onException(Throwable e) { + System.out.printf("%-10d 异步发送消息失败 %s %n", msg, e); + e.printStackTrace(); + } + }); + } catch (MQClientException e) { + e.printStackTrace(); + } catch (RemotingException e) { + e.printStackTrace(); + } catch (InterruptedException e) { + e.printStackTrace(); + } catch (UnsupportedEncodingException e) { + throw new RuntimeException(e); + } + return null; + } + +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/utils/JsonDataParser.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/utils/JsonDataParser.java new file mode 100644 index 0000000..80882af --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/utils/JsonDataParser.java @@ -0,0 +1,41 @@ +package com.tongran.frp.utils; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.springframework.util.StringUtils; + +import java.util.ArrayList; +import java.util.List; + +public class JsonDataParser { + private static final ObjectMapper objectMapper = new ObjectMapper(); + + /** + * 通用JSON解析方法（兼容对象和数组） + * @param jsonStr JSON字符串 + * @param valueType 目标实体类类型 + * @return 实体类List集合 + */ + public static List parseJsonData(String jsonStr, Class valueType) { + if (!StringUtils.hasText(jsonStr)) { + return new ArrayList<>(); + } + + try { + JsonNode rootNode = objectMapper.readTree(jsonStr); + + if (rootNode.isArray()) { + // 处理数组格式JSON + return objectMapper.readValue(jsonStr, + objectMapper.getTypeFactory().constructCollectionType(List.class, valueType)); + } else { + // 处理单个对象格式JSON + List result = new ArrayList<>(1); + result.add(objectMapper.readValue(jsonStr, valueType)); + return result; + } + } catch (Exception e) { + throw new RuntimeException("JSON解析失败: " + e.getMessage(), e); + } + } +} \ No newline at end of file diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/utils/SwitchJsonDataParser.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/utils/SwitchJsonDataParser.java new file mode 100644 index 0000000..0a69933 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/utils/SwitchJsonDataParser.java @@ -0,0 +1,149 @@ +package com.tongran.frp.utils; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.springframework.util.StringUtils; + +import java.util.ArrayList; +import java.util.List; + +public class SwitchJsonDataParser { + private static final ObjectMapper objectMapper = new ObjectMapper(); + + /** + * 通用JSON解析方法（兼容对象和数组） + * @param jsonStr JSON字符串 + * @param valueType 目标实体类类型 + * @return 实体类List集合 + */ + public static List parseJsonData(String jsonStr, Class valueType) { + if (!StringUtils.hasText(jsonStr)) { + return new ArrayList<>(); + } + + try { + JsonNode rootNode = objectMapper.readTree(jsonStr); + + if (rootNode.isArray()) { + // 处理数组格式JSON + if (isStringArrayContainingJsonObjects((ArrayNode) rootNode)) { + // 处理包含JSON对象字符串的数组 - 转换为真正的对象数组 + ArrayNode processedArray = processStringJsonArrayToObjectArray((ArrayNode) rootNode); + return convertJsonArrayToList(processedArray, valueType); + } else { + // 处理普通JSON数组 + processJsonArray((ArrayNode) rootNode); + return convertJsonArrayToList((ArrayNode) rootNode, valueType); + } + } else { + // 处理单个对象格式JSON + if (rootNode.isObject()) { + processJsonObject((ObjectNode) rootNode); + } + List result = new ArrayList<>(1); + result.add(objectMapper.treeToValue(rootNode, valueType)); + return result; + } + } catch (Exception e) { + throw new RuntimeException("JSON解析失败: " + e.getMessage(), e); + } + } + + /** + * 将JsonArray转换为List + */ + private static List convertJsonArrayToList(ArrayNode arrayNode, Class valueType) throws Exception { + List result = new ArrayList<>(); + for (int i = 0; i < arrayNode.size(); i++) { + JsonNode element = arrayNode.get(i); + result.add(objectMapper.treeToValue(element, valueType)); + } + return result; + } + + /** + * 判断是否是包含JSON对象字符串的字符串数组 + */ + private static boolean isStringArrayContainingJsonObjects(ArrayNode arrayNode) { + if (arrayNode.size() == 0) return false; + + JsonNode firstElement = arrayNode.get(0); + if (firstElement.isTextual()) { + try { + String strValue = firstElement.textValue(); + // 检查是否是JSON对象格式的字符串 + if (strValue.startsWith("{") && strValue.endsWith("}")) { + objectMapper.readTree(strValue); + return true; + } + } catch (Exception e) { + return false; + } + } + return false; + } + + /** + * 处理包含JSON对象字符串的字符串数组，转换为真正的对象数组 + */ + private static ArrayNode processStringJsonArrayToObjectArray(ArrayNode arrayNode) { + ArrayNode resultArray = objectMapper.createArrayNode(); + + for (int i = 0; i < arrayNode.size(); i++) { + JsonNode element = arrayNode.get(i); + if (element.isTextual()) { + try { + String jsonString = element.textValue(); + JsonNode jsonNode = objectMapper.readTree(jsonString); + + if (jsonNode.isObject()) { + // 处理JSON对象中的 noSuchInstance + processJsonObject((ObjectNode) jsonNode); + resultArray.add(jsonNode); + } else { + resultArray.add(element); + } + } catch (Exception e) { + // 如果解析失败，保持原样 + resultArray.add(element); + } + } else { + resultArray.add(element); + } + } + + return resultArray; + } + + /** + * 处理JSON数组 + */ + private static void processJsonArray(ArrayNode arrayNode) { + for (int i = 0; i < arrayNode.size(); i++) { + JsonNode element = arrayNode.get(i); + if (element.isObject()) { + processJsonObject((ObjectNode) element); + } else if (element.isArray()) { + processJsonArray((ArrayNode) element); + } + } + } + + /** + * 处理JSON对象 + */ + private static void processJsonObject(ObjectNode objectNode) { + objectNode.fields().forEachRemaining(entry -> { + JsonNode value = entry.getValue(); + if (value.isTextual() && "noSuchInstance".equals(value.textValue())) { + objectNode.putNull(entry.getKey()); + } else if (value.isObject()) { + processJsonObject((ObjectNode) value); + } else if (value.isArray()) { + processJsonArray((ArrayNode) value); + } + }); + } +} \ No newline at end of file diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/utils/TableRouterUtil.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/utils/TableRouterUtil.java new file mode 100644 index 0000000..9d60fb0 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/utils/TableRouterUtil.java @@ -0,0 +1,99 @@ +package com.tongran.frp.utils; + +import java.time.LocalDateTime; +import java.time.format.DateTimeFormatter; +import java.util.LinkedHashSet; +import java.util.Set; + +public class TableRouterUtil { + + // 日期格式 + private static final DateTimeFormatter YEAR_MONTH_FORMAT = + DateTimeFormatter.ofPattern("yyyy_MM"); + private static final DateTimeFormatter DATE_TIME_FORMATTER = + DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); + // 表名前缀 + private static final String TABLE_PREFIX = "eps_initial_traffic"; + // 表名前缀 + private static final String TABLE_PREFIX_INITIAL = "initial_bandwidth_traffic"; + + + /** + * 根据创建时间获取表名 + * @param createTime 记录创建时间 + * @return 对应的分表名称 + * @throws IllegalArgumentException 如果createTime为null + * + * 示例： + * 2023-08-05 14:30:00 → eps_initial_traffic_2023_08_1_10 + * 2023-08-15 09:15:00 → eps_initial_traffic_2023_08_11_20 + * 2023-08-25 18:45:00 → eps_initial_traffic_2023_08_21_31 + */ + public static String getTableName(LocalDateTime createTime) { + if (createTime == null) { + throw new IllegalArgumentException("创建时间不能为null"); + } + + String yearMonth = createTime.format(YEAR_MONTH_FORMAT); + int day = createTime.getDayOfMonth(); + + return String.format("%s_%s_%s", + TABLE_PREFIX_INITIAL, + yearMonth, + getDayRange(day)); + } + + /** + * 获取时间范围内涉及的所有表名 + * @param startTime 开始时间 (格式: "yyyy-MM-dd HH:mm:ss") + * @param endTime 结束时间 (格式: "yyyy-MM-dd HH:mm:ss") + * @return 按时间顺序排列的表名集合 + */ + public static Set getTableNamesBetween(String startTime, String endTime) { + LocalDateTime start = parseDateTime(startTime); + LocalDateTime end = parseDateTime(endTime); + validateTimeRange(start, end); + + Set tableNames = new LinkedHashSet<>(); + LocalDateTime current = start.withHour(0).withMinute(0).withSecond(0); + + while (!current.isAfter(end)) { + tableNames.add(getTableName(current)); + current = current.plusDays(1); + } + + return tableNames; + } + // 解析字符串为LocalDateTime + private static LocalDateTime parseDateTime(String dateTimeStr) { + if (dateTimeStr == null || dateTimeStr.trim().isEmpty()) { + throw new IllegalArgumentException("时间字符串不能为空"); + } + try { + return LocalDateTime.parse(dateTimeStr, DATE_TIME_FORMATTER); + } catch (Exception e) { + throw new IllegalArgumentException("时间格式必须为: yyyy-MM-dd HH:mm:ss", e); + } + } + + // 获取日期区间 + private static String getDayRange(int day) { + if (day < 1 || day > 31) { + throw new IllegalArgumentException("日期必须在1-31之间"); + } + + if (day <= 10) return "1_10"; + if (day <= 20) return "11_20"; + return "21_31"; + } + + // 验证时间范围 + private static void validateTimeRange(LocalDateTime start, LocalDateTime end) { + if (start == null || end == null) { + throw new IllegalArgumentException("时间范围参数不能为null"); + } + if (start.isAfter(end)) { + throw new IllegalArgumentException("开始时间不能晚于结束时间"); + } + } +} diff --git a/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/utils/WeChatWorkBot.java b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/utils/WeChatWorkBot.java new file mode 100644 index 0000000..8a7a187 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/java/com/tongran/frp/utils/WeChatWorkBot.java @@ -0,0 +1,285 @@ +package com.tongran.frp.utils; + +import com.fasterxml.jackson.databind.ObjectMapper; + +import java.io.OutputStream; +import java.net.HttpURLConnection; +import java.net.URL; +import java.util.HashMap; +import java.util.Map; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +public class WeChatWorkBot { + private static final ObjectMapper mapper = new ObjectMapper(); + + /** + * 发送基于模板的文本消息 + * @param webhookUrl webhook地址 + * @param template 消息模板，例如："项目[项目名称]在[时间]发生[事件类型]" + * @param fieldValues 字段值的映射，key为中文字段名，value为实际值（支持String、Number、Boolean等） + * @return 是否发送成功 + */ + public static boolean sendTemplateMessage(String webhookUrl, String template, + Map fieldValues) { + return sendTemplateMessage(webhookUrl, template, fieldValues, null, null, false); + } + + /** + * 发送基于模板的文本消息（支持@功能） + * @param webhookUrl webhook地址 + * @param template 消息模板 + * @param fieldValues 字段值的映射 + * @param mentionedMobiles 被@的用户列表（手机号） + * @param mentionedAll 是否@所有人 + * @return 是否发送成功 + */ + public static boolean sendTemplateMessage(String webhookUrl, String template, + Map fieldValues, + String[] mentionedMobiles, boolean mentionedAll) { + return sendTemplateMessage(webhookUrl, template, fieldValues, null, mentionedMobiles, mentionedAll); + } + + /** + * 发送基于模板的文本消息（完整参数） + * @param webhookUrl webhook地址 + * @param template 消息模板 + * @param fieldValues 字段值的映射 + * @param defaultValue 未找到字段时的默认值 + * @param mentionedMobiles 被@的用户列表（手机号） + * @param mentionedAll 是否@所有人 + * @return 是否发送成功 + */ + public static boolean sendTemplateMessage(String webhookUrl, String template, + Map fieldValues, Object defaultValue, + String[] mentionedMobiles, boolean mentionedAll) { + try { + String actualContent = processTemplate(template, fieldValues, defaultValue); + return sendTextMessage(webhookUrl, actualContent, mentionedMobiles, mentionedAll); + } catch (Exception e) { + e.printStackTrace(); + return false; + } + } + + /** + * 处理模板，替换字段占位符 + * @param template 消息模板 + * @param fieldValues 字段值映射 + * @param defaultValue 默认值 + * @return 处理后的消息内容 + */ + public static String processTemplate(String template, Map fieldValues, + Object defaultValue) { + if (template == null) return ""; + if (fieldValues == null || fieldValues.isEmpty()) { + return template; + } + + Pattern pattern = Pattern.compile("\\[(.*?)\\]"); + Matcher matcher = pattern.matcher(template); + StringBuffer result = new StringBuffer(); + + while (matcher.find()) { + String fieldName = matcher.group(1); + Object fieldValueObj = fieldValues.get(fieldName); + String fieldValue = convertToString(fieldValueObj); + + // 如果字段值为空，使用默认值或保留原占位符 + if (fieldValue == null || fieldValue.trim().isEmpty()) { + fieldValue = convertToString(defaultValue); + if (fieldValue == null) { + fieldValue = "[" + fieldName + "]"; + } + } + + // 对替换值进行转义，防止正则表达式特殊字符问题 + matcher.appendReplacement(result, Matcher.quoteReplacement(fieldValue)); + } + matcher.appendTail(result); + + return result.toString(); + } + + /** + * 将对象转换为字符串 + * @param obj 要转换的对象 + * @return 字符串表示 + */ + private static String convertToString(Object obj) { + if (obj == null) { + return null; + } + + if (obj instanceof String) { + return (String) obj; + } else if (obj instanceof Number || obj instanceof Boolean) { + return String.valueOf(obj); + } else if (obj instanceof java.util.Date) { + return new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format((java.util.Date) obj); + } else { + return obj.toString(); + } + } + + /** + * 验证模板中的字段是否都有对应的值 + * @param template 消息模板 + * @param fieldValues 字段值映射 + * @return 是否所有字段都有值 + */ + public static boolean validateTemplate(String template, Map fieldValues) { + if (template == null || fieldValues == null) { + return false; + } + + Pattern pattern = Pattern.compile("\\[(.*?)\\]"); + Matcher matcher = pattern.matcher(template); + + while (matcher.find()) { + String fieldName = matcher.group(1); + Object fieldValue = fieldValues.get(fieldName); + String strValue = convertToString(fieldValue); + + if (!fieldValues.containsKey(fieldName) || + strValue == null || + strValue.trim().isEmpty()) { + return false; + } + } + return true; + } + + /** + * 获取模板中的所有字段名 + * @param template 消息模板 + * @return 字段名列表 + */ + public static java.util.List getTemplateFields(String template) { + java.util.List fields = new java.util.ArrayList<>(); + if (template == null) { + return fields; + } + + Pattern pattern = Pattern.compile("\\[(.*?)\\]"); + Matcher matcher = pattern.matcher(template); + + while (matcher.find()) { + fields.add(matcher.group(1)); + } + return fields; + } + + /** + * 发送Markdown模板消息 + * @param webhookUrl webhook地址 + * @param template Markdown模板 + * @param fieldValues 字段值映射 + * @return 是否发送成功 + */ + public static boolean sendMarkdownTemplateMessage(String webhookUrl, String template, + Map fieldValues) { + return sendMarkdownTemplateMessage(webhookUrl, template, fieldValues, null); + } + + /** + * 发送Markdown模板消息 + * @param webhookUrl webhook地址 + * @param template Markdown模板 + * @param fieldValues 字段值映射 + * @param defaultValue 默认值 + * @return 是否发送成功 + */ + public static boolean sendMarkdownTemplateMessage(String webhookUrl, String template, + Map fieldValues, + Object defaultValue) { + try { + String actualContent = processTemplate(template, fieldValues, defaultValue); + return sendMarkdownMessage(webhookUrl, actualContent); + } catch (Exception e) { + e.printStackTrace(); + return false; + } + } + + /** + * 发送文本消息 + */ + public static boolean sendTextMessage(String webhookUrl, String content) { + return sendTextMessage(webhookUrl, content, null, false); + } + + /** + * 发送文本消息（支持@功能） + */ + public static boolean sendTextMessage(String webhookUrl, String content, + String[] mentionedMobiles, boolean mentionedAll) { + try { + Map message = new HashMap<>(); + message.put("msgtype", "text"); + + Map textContent = new HashMap<>(); + textContent.put("content", content); + + if (mentionedAll) { + textContent.put("mentioned_mobile_list", new String[]{"@all"}); + } else if (mentionedMobiles != null && mentionedMobiles.length > 0) { + textContent.put("mentioned_mobile_list", mentionedMobiles); + } + + message.put("text", textContent); + + return sendMessage(webhookUrl, mapper.writeValueAsString(message)); + + } catch (Exception e) { + e.printStackTrace(); + return false; + } + } + + /** + * 发送Markdown消息 + */ + public static boolean sendMarkdownMessage(String webhookUrl, String content) { + try { + Map message = new HashMap<>(); + message.put("msgtype", "markdown"); + + Map markdownContent = new HashMap<>(); + markdownContent.put("content", content); + + message.put("markdown", markdownContent); + + return sendMessage(webhookUrl, mapper.writeValueAsString(message)); + + } catch (Exception e) { + e.printStackTrace(); + return false; + } + } + + private static boolean sendMessage(String webhookUrl, String jsonBody) { + try { + URL url = new URL(webhookUrl); + HttpURLConnection connection = (HttpURLConnection) url.openConnection(); + + connection.setRequestMethod("POST"); + connection.setRequestProperty("Content-Type", "application/json"); + connection.setDoOutput(true); + connection.setConnectTimeout(5000); + connection.setReadTimeout(10000); + + try (OutputStream os = connection.getOutputStream()) { + byte[] input = jsonBody.getBytes("UTF-8"); + os.write(input, 0, input.length); + } + + int responseCode = connection.getResponseCode(); + return responseCode == 200; + + } catch (Exception e) { + e.printStackTrace(); + return false; + } + } +} \ No newline at end of file diff --git a/tongran-modules/tongran-frp/src/main/resources/bootstrap.yml b/tongran-modules/tongran-frp/src/main/resources/bootstrap.yml new file mode 100644 index 0000000..da65df9 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/resources/bootstrap.yml @@ -0,0 +1,43 @@ +# Tomcat +server: + port: 9209 + +# Spring +spring: + application: + # 应用名称 + name: tongran-frp + profiles: + # 环境配置 + active: dev + cloud: + nacos: + discovery: + # 服务注册地址 + server-addr: ${spring.cloud.nacos.config.server-addr} + namespace: ${spring.cloud.nacos.config.namespace} + username: ${spring.cloud.nacos.config.username} + password: ${spring.cloud.nacos.config.password} + config: + # 配置中心地址 + server-addr: 172.16.15.52:8848 +# server-addr: 172.16.15.103:8848 +# namespace: public + namespace: saas-prod + username: nacos + password: nacos + # 配置文件格式 + file-extension: yml + # 共享配置 + shared-configs: + - application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension} + + redisson: + singleServerConfig: + address: redis://localhost:6379 +logging: + level: + com.tongran.app.mapper: DEBUG + + + diff --git a/tongran-modules/tongran-frp/src/main/resources/logback.xml b/tongran-modules/tongran-frp/src/main/resources/logback.xml new file mode 100644 index 0000000..de115b4 --- /dev/null +++ b/tongran-modules/tongran-frp/src/main/resources/logback.xml @@ -0,0 +1,72 @@ + + + + + + + + + + + ${log.pattern} + + + + + + ${log.path}/info.log + + + + ${log.path}/info.%d{yyyy-MM-dd}.log + + 60 + + + ${log.pattern} + + + + INFO + + ACCEPT + + DENY + + + + + + ${log.path}/error.log + + + + ${log.path}/error.%d{yyyy-MM-dd}.log + + 60 + + + ${log.pattern} + + + + ERROR + + ACCEPT + + DENY + + + + + + + + + + + + + + + \ No newline at end of file