完善【流量统计报表 - 分钟级别】处理逻辑

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/dal/FlowReportMinuteMapper.java

@@ -22,11 +22,17 @@
 package fun.asgc.neutrino.proxy.server.dal;
 
 import fun.asgc.neutrino.core.annotation.Component;
+import fun.asgc.neutrino.core.annotation.Param;
 import fun.asgc.neutrino.core.aop.Intercept;
 import fun.asgc.neutrino.core.db.annotation.Insert;
+import fun.asgc.neutrino.core.db.annotation.ResultType;
+import fun.asgc.neutrino.core.db.annotation.Select;
 import fun.asgc.neutrino.core.db.mapper.SqlMapper;
 import fun.asgc.neutrino.proxy.server.dal.entity.FlowReportMinuteDO;
 
+import java.util.List;
+import java.util.Set;
+
 /**
  * @author: aoshiguchen
  * @date: 2022/10/24
@@ -34,5 +40,13 @@ import fun.asgc.neutrino.proxy.server.dal.entity.FlowReportMinuteDO;
 @Intercept(ignoreGlobal = true)
 @Component
 public interface FlowReportMinuteMapper extends SqlMapper {
+    @Select("select * from flow_report_minute where license_id = :licenseId and date = :date")
+    FlowReportMinuteDO findOne(@Param("licenseId") Integer licenseId, @Param("date") String date);
 
+    @ResultType(FlowReportMinuteDO.class)
+    @Select("select * from flow_report_minute where license_id in (:licenseIds) and date = :date")
+    List<FlowReportMinuteDO> findList(@Param("licenseIds") Set<Integer> licenseIds, @Param("date") String date);
+
+    @Insert("insert into flow_report_minute(`user_id`,`license_id`,`write_bytes`,`read_bytes`,`date`,`create_time`) values(:userId,:licenseId,:writeBytes,:readBytes,:date,:createTime)")
+    void add(FlowReportMinuteDO flowReportMinuteDO);
 }

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/job/FlowReportForMinuteJob.java

@@ -35,8 +35,9 @@ import fun.asgc.neutrino.proxy.server.dal.entity.LicenseDO;
 import fun.asgc.neutrino.proxy.server.service.FlowReportService;
 import lombok.extern.slf4j.Slf4j;
 
-import java.util.Date;
-import java.util.List;
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 /**
  * 流量统计报表 - 分钟级别
@@ -62,8 +63,18 @@ public class FlowReportForMinuteJob implements IJobHandler {
         if (CollectionUtil.isEmpty(list)) {
             return;
         }
+        Set<Integer> licenseIds = list.stream().map(LicenseDO::getId).collect(Collectors.toSet());
         Date now = new Date();
+        String date = DateUtil.format(DateUtil.addDate(now, Calendar.MINUTE, -1), "yyyy-MM-dd HH:mm");
+        List<FlowReportMinuteDO> oldList = flowReportMinuteMapper.findList(licenseIds, date);
+        Map<Integer, FlowReportMinuteDO> oldMap = CollectionUtil.isEmpty(oldList) ? new HashMap<>() :
+                oldList.stream().collect(Collectors.toMap(FlowReportMinuteDO::getLicenseId, Function.identity(), (a,b) -> a));
+
         for (LicenseDO item : list) {
+            // 避免job重复执行导致数据重复
+            if (oldMap.containsKey(item.getId())) {
+                continue;
+            }
             Integer writeBytes = flowReportService.getAndResetWriteByte(item.getId());
             Integer readBytes = flowReportService.getAndResetReadByte(item.getId());
             if (writeBytes == 0 && readBytes == 0) {
@@ -74,9 +85,9 @@ public class FlowReportForMinuteJob implements IJobHandler {
             flowReportMinuteDO.setLicenseId(item.getId());
             flowReportMinuteDO.setWriteBytes(writeBytes);
             flowReportMinuteDO.setReadBytes(readBytes);
-            flowReportMinuteDO.setDate(DateUtil.format(now, "yyyy-MM-dd HH:mm"));
+            flowReportMinuteDO.setDate(date);
             flowReportMinuteDO.setCreateTime(now);
-            // TODO insert
+            flowReportMinuteMapper.add(flowReportMinuteDO);
         }
     }
 }

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/core/BytesMetricsHandler.java

@@ -43,7 +43,7 @@ public class BytesMetricsHandler extends ChannelDuplexHandler {
         MetricsCollector metricsCollector = MetricsCollector.getCollector(sa.getPort());
         metricsCollector.incrementReadBytes(((ByteBuf) msg).readableBytes());
         metricsCollector.incrementReadMsgs(1);
-        System.out.println("字节数:" + metricsCollector.getMetrics().getReadBytes());
+//        System.out.println("字节数:" + metricsCollector.getMetrics().getReadBytes());
         ctx.fireChannelRead(msg);
     }
 

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/core/VisitorChannelHandler.java

@@ -71,7 +71,7 @@ public class VisitorChannelHandler extends SimpleChannelInboundHandler<ByteBuf>
 
             // 增加流量计数
             VisitorChannelAttachInfo visitorChannelAttachInfo = ProxyUtil.getAttachInfo(visitorChannel);
-            BeanManager.getBean(FlowReportService.class).addWriteByte(visitorChannelAttachInfo.getLicenseId(), buf.readableBytes());
+            BeanManager.getBean(FlowReportService.class).addWriteByte(visitorChannelAttachInfo.getLicenseId(), bytes.length);
         }
     }
 

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageTransferHandler.java

@@ -57,7 +57,7 @@ public class ProxyMessageTransferHandler implements ProxyMessageHandler {
 
 			// 增加流量计数
 			VisitorChannelAttachInfo visitorChannelAttachInfo = ProxyUtil.getAttachInfo(visitorChannel);
-			BeanManager.getBean(FlowReportService.class).addReadByte(visitorChannelAttachInfo.getLicenseId(), buf.readableBytes());
+			BeanManager.getBean(FlowReportService.class).addReadByte(visitorChannelAttachInfo.getLicenseId(), proxyMessage.getData().length);
 		}
 	}