From f39aa887c8d5e48f21dbd430d83fd35b25c8aa2b Mon Sep 17 00:00:00 2001 From: gaoyutao Date: Thu, 15 Jan 2026 18:14:17 +0800 Subject: [PATCH] =?UTF-8?q?=E7=A3=81=E7=9B=98=E6=95=B0=E6=8D=AE=E8=A1=A8?= =?UTF-8?q?=E5=88=86=E8=A1=A8=EF=BC=8C=E9=80=82=E9=85=8D=E7=A3=81=E7=9B=98?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=90=8D=E7=A7=B0=E5=AD=98=E5=82=A8=E8=A1=A8?= =?UTF-8?q?=E3=80=82=20=E7=A3=81=E7=9B=98=E7=9B=B8=E5=85=B3=E5=8A=9F?= =?UTF-8?q?=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../tongran/common/core/enums/MsgEnum.java | 2 + .../controller/AllDiskNameController.java | 41 ++----- .../tongran/rocketmq/domain/AllDiskName.java | 81 +------------- .../rocketmq/domain/vo/DiskIopsResultVo.java | 16 +++ .../rocketmq/domain/vo/PolicyTypeVo.java | 4 + .../rocketmq/handler/MessageHandler.java | 103 ++++++++++++++++-- .../rocketmq/service/IAllDiskNameService.java | 2 + .../service/impl/AllDiskNameServiceImpl.java | 38 +++++++ .../mapper/rocketmq/AllDiskNameMapper.xml | 16 ++- 9 files changed, 181 insertions(+), 122 deletions(-) create mode 100644 tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/DiskIopsResultVo.java diff --git a/tongran-common/tongran-common-core/src/main/java/com/tongran/common/core/enums/MsgEnum.java b/tongran-common/tongran-common-core/src/main/java/com/tongran/common/core/enums/MsgEnum.java index 676d2d4..71e2d14 100644 --- a/tongran-common/tongran-common-core/src/main/java/com/tongran/common/core/enums/MsgEnum.java +++ b/tongran-common/tongran-common-core/src/main/java/com/tongran/common/core/enums/MsgEnum.java @@ -19,6 +19,8 @@ public enum MsgEnum { macvlan状态上报("MACVLAN_STATUS_RSP"), + iops结果上报("IOPS_RESULT"), + 获取最新策略("GET_POLICY"), 获取最新策略应答("GET_POLICY_RSP"), diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/controller/AllDiskNameController.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/controller/AllDiskNameController.java index 754c460..71f5be3 100644 --- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/controller/AllDiskNameController.java +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/controller/AllDiskNameController.java @@ -1,9 +1,7 @@ package com.tongran.rocketmq.controller; -import com.tongran.common.core.utils.poi.ExcelUtil; import com.tongran.common.core.web.controller.BaseController; import com.tongran.common.core.web.domain.AjaxResult; -import com.tongran.common.core.web.page.TableDataInfo; import com.tongran.common.log.annotation.Log; import com.tongran.common.log.enums.BusinessType; import com.tongran.common.security.annotation.RequiresPermissions; @@ -12,7 +10,6 @@ import com.tongran.rocketmq.service.IAllDiskNameService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.*; -import javax.servlet.http.HttpServletResponse; import java.util.List; /** @@ -23,6 +20,7 @@ import java.util.List; */ @RestController @RequestMapping("/allDiskName") +@RequiresPermissions("rocketmq:traffic") public class AllDiskNameController extends BaseController { @Autowired @@ -31,32 +29,17 @@ public class AllDiskNameController extends BaseController /** * 查询磁盘名称存储列表 */ - @RequiresPermissions("rocketmq:allDiskName:list") - @GetMapping("/list") - public TableDataInfo list(AllDiskName allDiskName) + @PostMapping("/list") + public AjaxResult list(@RequestBody AllDiskName allDiskName) { - startPage(); List list = allDiskNameService.selectAllDiskNameList(allDiskName); - return getDataTable(list); + return success(list); } - /** - * 导出磁盘名称存储列表 - */ - @RequiresPermissions("rocketmq:allDiskName:export") - @Log(title = "磁盘名称存储", businessType = BusinessType.EXPORT) - @PostMapping("/export") - public void export(HttpServletResponse response, AllDiskName allDiskName) - { - List list = allDiskNameService.selectAllDiskNameList(allDiskName); - ExcelUtil util = new ExcelUtil(AllDiskName.class); - util.exportExcel(response, list, "磁盘名称存储数据"); - } /** * 获取磁盘名称存储详细信息 */ - @RequiresPermissions("rocketmq:allDiskName:query") @GetMapping(value = "/{id}") public AjaxResult getInfo(@PathVariable("id") Long id) { @@ -64,20 +47,19 @@ public class AllDiskNameController extends BaseController } /** - * 新增磁盘名称存储 + * 测试磁盘IOPS/卸载分区 + * @param allDiskName + * @return */ - @RequiresPermissions("rocketmq:allDiskName:add") - @Log(title = "磁盘名称存储", businessType = BusinessType.INSERT) - @PostMapping - public AjaxResult add(@RequestBody AllDiskName allDiskName) - { - return toAjax(allDiskNameService.insertAllDiskName(allDiskName)); + @PostMapping("/issuanceOperation") + public AjaxResult testIOPS(@RequestBody AllDiskName allDiskName){ + return toAjax(allDiskNameService.issuanceOperation(allDiskName)); } + /** * 修改磁盘名称存储 */ - @RequiresPermissions("rocketmq:allDiskName:edit") @Log(title = "磁盘名称存储", businessType = BusinessType.UPDATE) @PutMapping public AjaxResult edit(@RequestBody AllDiskName allDiskName) @@ -88,7 +70,6 @@ public class AllDiskNameController extends BaseController /** * 删除磁盘名称存储 */ - @RequiresPermissions("rocketmq:allDiskName:remove") @Log(title = "磁盘名称存储", businessType = BusinessType.DELETE) @DeleteMapping("/{ids}") public AjaxResult remove(@PathVariable Long[] ids) diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/AllDiskName.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/AllDiskName.java index 03afaf9..d1f100a 100644 --- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/AllDiskName.java +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/AllDiskName.java @@ -1,9 +1,8 @@ package com.tongran.rocketmq.domain; -import org.apache.commons.lang3.builder.ToStringBuilder; -import org.apache.commons.lang3.builder.ToStringStyle; import com.tongran.common.core.annotation.Excel; import com.tongran.common.core.web.domain.BaseEntity; +import lombok.Data; /** * 磁盘名称存储对象 all_disk_name @@ -11,6 +10,7 @@ import com.tongran.common.core.web.domain.BaseEntity; * @author tongran * @date 2026-01-14 */ +@Data public class AllDiskName extends BaseEntity { private static final long serialVersionUID = 1L; @@ -37,80 +37,7 @@ public class AllDiskName extends BaseEntity /** 写入IOPS */ @Excel(name = "写入IOPS") private String writeIops; + /** 操纵类型 1测试磁盘IOPS，2卸载分区 */ + private Integer processType; - public void setId(Long id) - { - this.id = id; - } - - public Long getId() - { - return id; - } - - public void setClientId(String clientId) - { - this.clientId = clientId; - } - - public String getClientId() - { - return clientId; - } - - public void setName(String name) - { - this.name = name; - } - - public String getName() - { - return name; - } - - public void setStatus(Integer status) - { - this.status = status; - } - - public Integer getStatus() - { - return status; - } - - public void setReadIops(String readIops) - { - this.readIops = readIops; - } - - public String getReadIops() - { - return readIops; - } - - public void setWriteIops(String writeIops) - { - this.writeIops = writeIops; - } - - public String getWriteIops() - { - return writeIops; - } - - @Override - public String toString() { - return new ToStringBuilder(this,ToStringStyle.MULTI_LINE_STYLE) - .append("id", getId()) - .append("clientId", getClientId()) - .append("name", getName()) - .append("status", getStatus()) - .append("readIops", getReadIops()) - .append("writeIops", getWriteIops()) - .append("createTime", getCreateTime()) - .append("updateTime", getUpdateTime()) - .append("createBy", getCreateBy()) - .append("updateBy", getUpdateBy()) - .toString(); - } } diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/DiskIopsResultVo.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/DiskIopsResultVo.java new file mode 100644 index 0000000..41b58cb --- /dev/null +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/DiskIopsResultVo.java @@ -0,0 +1,16 @@ +package com.tongran.rocketmq.domain.vo; + +import lombok.Data; + +@Data +public class DiskIopsResultVo { + + /** 磁盘名称 */ + private String name; + /** 读iops */ + private String readIops; + /** 写iops */ + private String writeIops; + /** 时间戳 */ + private long timestamp; +} diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/PolicyTypeVo.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/PolicyTypeVo.java index dfe5047..6389281 100644 --- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/PolicyTypeVo.java +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/PolicyTypeVo.java @@ -21,6 +21,10 @@ public class PolicyTypeVo { private String frpMsg; /** pppoe配置信息 */ private String pppoeConfig; + /** 开启测试磁盘IOPS */ + private String diskIopsTest; + /** 卸载磁盘分区 */ + private String umountDisk; /** 时间戳 */ private Long timestamp = Instant.now().getEpochSecond(); } diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java index a6d8461..c9ae3da 100644 --- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java @@ -45,6 +45,8 @@ import java.util.stream.Collectors; public class MessageHandler { private final Map> messageHandlers = new HashMap<>(); + // 硬盘上报状态 + String DISK_COUNT_PREFIX = "disk:count:"; // 心跳状态 private static final String HEARTBEAT_STATUS_PREFIX = "heartbeat:status:"; // 心跳时间 @@ -104,6 +106,8 @@ public class MessageHandler { private IRmFrpcConfigManageService rmFrpcConfigManageService; @Autowired private IRmPppoeConfigSubService rmPppoeConfigSubService; + @Autowired + private IAllDiskNameService allDiskNameService; /** @@ -128,6 +132,24 @@ public class MessageHandler { registerHandler(MsgEnum.多公网IP探测.getValue(), this::handleNetWorkDelectMessage); registerHandler(MsgEnum.修改frp配置文件应答.getValue(), this::handleUpdateFrpMessage); registerHandler(MsgEnum.macvlan状态上报.getValue(), this::handleMacvlanMessage); + registerHandler(MsgEnum.iops结果上报.getValue(), this::handleIopsResultMessage); + } + + private void handleIopsResultMessage(DeviceMessage message) { + List diskIopsResultVoList = JsonDataParser.parseJsonData(message.getData(), DiskIopsResultVo.class); + if(diskIopsResultVoList != null && !diskIopsResultVoList.isEmpty()){ + DiskIopsResultVo diskIopsResultVo = diskIopsResultVoList.get(0); + // 时间戳转换 + long timestamp = diskIopsResultVo.getTimestamp(); + long millis = timestamp * 1000; + Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 + AllDiskName allDiskName = new AllDiskName(); + allDiskName.setClientId(message.getClientId()); + allDiskName.setName(diskIopsResultVo.getName()); + allDiskName.setReadIops(diskIopsResultVo.getReadIops()); + allDiskName.setWriteIops(diskIopsResultVo.getWriteIops()); + allDiskNameService.updateAllDiskName(allDiskName); + } } private void handleMacvlanMessage(DeviceMessage message) { @@ -540,20 +562,77 @@ public class MessageHandler { */ private void handleDiskMessage(DeviceMessage message) { List disks = JsonDataParser.parseJsonData(message.getData(), InitialDiskInfo.class); - // 时间戳转换 - long timestamp = disks.get(0).getTimestamp(); - long millis = timestamp * 1000; - Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 - if(!disks.isEmpty()){ - disks.forEach(iface -> { - iface.setClientId(message.getClientId()); - iface.setCreateTime(createTime); - }); - // 初始磁盘数据入库 - initialDiskInfoService.batchInsertInitialDiskInfo(disks, createTime); - }else{ + String clientId = message.getClientId(); + + if (disks == null || disks.isEmpty()) { throw new RuntimeException("磁盘data数据为空"); } + + long timestamp = disks.get(0).getTimestamp(); + long millis = timestamp * 1000; + Date createTime = new Date(millis / 1000 * 1000); + + // 关键：每个clientId有自己独立的key + String diskCountKey = DISK_COUNT_PREFIX + clientId; + + // 1. 给这个客户端的所有磁盘次数+1 + Map diskCountMap = redisTemplate.opsForHash().entries(diskCountKey); + for (Map.Entry entry : diskCountMap.entrySet()) { + String diskName = (String) entry.getKey(); + String countStr = (String) entry.getValue(); + + try { + int count = Integer.parseInt(countStr) + 1; + redisTemplate.opsForHash().put(diskCountKey, diskName, String.valueOf(count)); + } catch (NumberFormatException e) { + redisTemplate.opsForHash().put(diskCountKey, diskName, "1"); + } + } + + // 2. 处理本次上报的磁盘 + Set reportedDisks = new HashSet<>(); + disks.forEach(disk -> { + String diskName = disk.getName(); + reportedDisks.add(diskName); + + // 本次上报的磁盘，次数重置为0 + redisTemplate.opsForHash().put(diskCountKey, diskName, "0"); + + disk.setClientId(clientId); + disk.setCreateTime(createTime); + }); + + // 3. 检查次数≥3的磁盘 + diskCountMap = redisTemplate.opsForHash().entries(diskCountKey); + List disksToRemove = new ArrayList<>(); + + for (Map.Entry entry : diskCountMap.entrySet()) { + String diskName = (String) entry.getKey(); + String countStr = (String) entry.getValue(); + + try { + int count = Integer.parseInt(countStr); + // 如果次数≥3且本次没上报 + if (count >= 3 && !reportedDisks.contains(diskName)) { + AllDiskName allDiskName = new AllDiskName(); + allDiskName.setStatus(0); + allDiskName.setClientId(clientId); + allDiskName.setName(diskName); + allDiskNameService.updateAllDiskName(allDiskName); + disksToRemove.add(diskName); + } + } catch (NumberFormatException e) { + disksToRemove.add(diskName); + } + } + + // 4. 删除已处理的磁盘记录 + if (!disksToRemove.isEmpty()) { + redisTemplate.opsForHash().delete(diskCountKey, disksToRemove.toArray()); + } + + // 5. 数据入库 + initialDiskInfoService.batchInsertInitialDiskInfo(disks, createTime); } /** * 内存数据入库 diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IAllDiskNameService.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IAllDiskNameService.java index 24c9fe5..9e8e067 100644 --- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IAllDiskNameService.java +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IAllDiskNameService.java @@ -62,4 +62,6 @@ public interface IAllDiskNameService public int deleteAllDiskNameById(Long id); int batchInsertAllDistName(List dataList); + + int issuanceOperation(AllDiskName allDiskName); } diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/AllDiskNameServiceImpl.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/AllDiskNameServiceImpl.java index e100724..1a8f6cf 100644 --- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/AllDiskNameServiceImpl.java +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/AllDiskNameServiceImpl.java @@ -1,9 +1,15 @@ package com.tongran.rocketmq.service.impl; +import com.alibaba.fastjson.JSONObject; +import com.tongran.common.core.enums.MsgEnum; import com.tongran.common.core.utils.DateUtils; import com.tongran.rocketmq.domain.AllDiskName; +import com.tongran.rocketmq.domain.DeviceMessage; import com.tongran.rocketmq.domain.InitialDiskInfo; +import com.tongran.rocketmq.domain.vo.PolicyTypeVo; import com.tongran.rocketmq.mapper.AllDiskNameMapper; +import com.tongran.rocketmq.model.ProducerMode; +import com.tongran.rocketmq.producer.MessageProducer; import com.tongran.rocketmq.service.IAllDiskNameService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; @@ -22,6 +28,8 @@ public class AllDiskNameServiceImpl implements IAllDiskNameService { @Autowired private AllDiskNameMapper allDiskNameMapper; + @Autowired + private ProducerMode producerMode; /** * 查询磁盘名称存储 @@ -115,4 +123,34 @@ public class AllDiskNameServiceImpl implements IAllDiskNameService allDiskNameMapper.batchInsertAllDistName(nameList); return 1; } + + @Override + public int issuanceOperation(AllDiskName query) { + Long id = query.getId(); + Integer processType = query.getProcessType(); + AllDiskName allDiskName = new AllDiskName(); + allDiskName = allDiskNameMapper.selectAllDiskNameById(id); + String clientId = allDiskName.getClientId(); + // 构建磁盘下发信息 + PolicyTypeVo policyTypeVo = new PolicyTypeVo(); + if(processType == 1){ + policyTypeVo.setDiskIopsTest(allDiskName.getName()); + }else if(processType ==2){ + policyTypeVo.setUmountDisk(allDiskName.getName()); + } + MessageProducer messageProducer = new MessageProducer(); + String configJson = JSONObject.toJSONString(policyTypeVo); + DeviceMessage message = new DeviceMessage(); + message.setClientId(clientId); + message.setData(configJson); + message.setDataType(MsgEnum.获取最新策略应答.getValue()); + + messageProducer.sendAsyncProducerMessage( + producerMode.getAgentTopic(), + "", + "", + JSONObject.toJSONString(message) + ); + return 1; + } } diff --git a/tongran-rocketmq/src/main/resources/mapper/rocketmq/AllDiskNameMapper.xml b/tongran-rocketmq/src/main/resources/mapper/rocketmq/AllDiskNameMapper.xml index 709d35f..54d6be6 100644 --- a/tongran-rocketmq/src/main/resources/mapper/rocketmq/AllDiskNameMapper.xml +++ b/tongran-rocketmq/src/main/resources/mapper/rocketmq/AllDiskNameMapper.xml @@ -75,8 +75,6 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" update all_disk_name - client_id = #{clientId}, - name = #{name}, status = #{status}, read_iops = #{readIops}, write_iops = #{writeIops}, @@ -85,7 +83,19 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" create_by = #{createBy}, update_by = #{updateBy}, - where id = #{id} + + + + and id=#{id} + + + and client_id = #{clientId} and name =#{name} + + + and 1=0 + + +