减少缓存时间

Author: master-coder-ll

proxy/src/main/java/com/jhl/cache/ConnectionStatsCache.java

@@ -3,7 +3,6 @@
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
 import com.jhl.pojo.AccountConnectionStat;
-import com.jhl.service.ProxyAccountService;
 import com.jhl.task.GlobalConnectionStatTask;
 import com.jhl.task.service.TaskService;
 import com.jhl.utils.SynchronizedInternerUtils;
@@ -22,7 +21,7 @@
 public class ConnectionStatsCache {
 
     private static final Cache<Object, AccountConnectionStat> ACCOUNT_CONNECTION_COUNT_STATS = CacheBuilder.newBuilder()
-            .expireAfterAccess(ProxyAccountService.ACCOUNT_EXPIRE_TIME, TimeUnit.MINUTES).build();
+            .expireAfterAccess(5, TimeUnit.MINUTES).build();
 
     public final static long _1HOUR_MS = 3600000;
 
@@ -107,20 +106,20 @@ public static boolean isFull(String accountId, int maxConnections) {
         return false;
     }
 
-    public static  boolean canReport(String accountId) {
-        synchronized (SynchronizedInternerUtils.getInterner().intern(accountId)){
-        AccountConnectionStat connectionCounter = ACCOUNT_CONNECTION_COUNT_STATS.getIfPresent(accountId);
-        if (connectionCounter != null) {
-            long interruptionTime = connectionCounter.getInterruptionTime();
-            //操作一个小时后，可以继续执行上报
-            //interruptionTime =0 ok
-            // interruptionTime !=0 ok
-            if ((System.currentTimeMillis() - interruptionTime) > _1HOUR_MS) {
-                connectionCounter.setInterruptionTime(System.currentTimeMillis());
-                return true;
+    public static boolean canReport(String accountId) {
+        synchronized (SynchronizedInternerUtils.getInterner().intern(accountId)) {
+            AccountConnectionStat connectionCounter = ACCOUNT_CONNECTION_COUNT_STATS.getIfPresent(accountId);
+            if (connectionCounter != null) {
+                long interruptionTime = connectionCounter.getInterruptionTime();
+                //操作一个小时后，可以继续执行上报
+                //interruptionTime =0 ok
+                // interruptionTime !=0 ok
+                if ((System.currentTimeMillis() - interruptionTime) > _1HOUR_MS) {
+                    connectionCounter.setInterruptionTime(System.currentTimeMillis());
+                    return true;
+                }
             }
-        }
-        return false;
+            return false;
         }
     }
 
@@ -136,7 +135,7 @@ public static void updateGlobalConnectionStat(String accountNo, int count, long
             connectionCounter.updateRemoteConnectionNum(count);
         }
         //全局控制
-        if (interruptionTime>0)  connectionCounter.setInterruptionTime(interruptionTime);
+        if (interruptionTime > 0) connectionCounter.setInterruptionTime(interruptionTime);
     }
 
     /**
@@ -152,9 +151,9 @@ public static void updateGlobalConnectionStat(String accountNo, int count, long
      */
     public static void reportConnectionNum(String accountNo, String proxyIp) {
         AccountConnectionStat connectionCounter = ACCOUNT_CONNECTION_COUNT_STATS.getIfPresent(accountNo);
-        int internalConnectionCount = connectionCounter.getByServer();
 
         if (System.currentTimeMillis() - connectionCounter.getLastReportTime() > _30S) {
+            int internalConnectionCount = connectionCounter.getByServer();
             GlobalConnectionStatTask globalConnectionStatTask =
                     new GlobalConnectionStatTask(accountNo, proxyIp, internalConnectionCount);
             TaskService.addTask(globalConnectionStatTask);

proxy/src/main/java/com/jhl/proxy/Dispatcher.java

@@ -36,7 +36,7 @@ public class Dispatcher extends ChannelInboundHandlerAdapter {
 
     private static final String HOST = "HOST";
     private static final Long MAX_INTERVAL_REPORT_TIME_MS = 1000 * 60 * 5L;
-
+    private static final Long _1MINUTES=60_000L;
     /**
      * proxy端配置数据
      */
@@ -71,7 +71,13 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
         } else {
 
             try {
+
+                if (proxyAccountService.interrupted(accountNo, host, version))
+                    throw new ReleaseDirectMemoryException("【当前版本已经更新】抛出异常。统一内存释放");
+
                 writeToOutBoundChannel(msg, ctx);
+
+                ConnectionStatsCache.reportConnectionNum(accountNo,proxyIp);
                 segmentReportingFlowStat();
             } catch (Exception e) {
                 if (!(e instanceof ReleaseDirectMemoryException)) {
@@ -379,9 +385,7 @@ private String getAccountId() {
 
 
     private void writeToOutBoundChannel(Object msg, final ChannelHandlerContext ctx) throws ReleaseDirectMemoryException {
-        if (proxyAccountService.interrupted(accountNo, host, version)) {
-            throw new ReleaseDirectMemoryException("【当前版本已经更新】抛出异常。统一内存释放");
-        }
+
         outboundChannel.writeAndFlush(msg).addListener((ChannelFutureListener) future -> {
             release((ByteBuf) msg);
             if (future.isSuccess()) {
@@ -392,6 +396,7 @@ private void writeToOutBoundChannel(Object msg, final ChannelHandlerContext ctx)
         });
     }
 
+
     private void reportConnectionLimit() {
 
         if (ConnectionStatsCache.canReport(accountNo)) {
@@ -412,8 +417,8 @@ private void segmentReportingFlowStat() {
       if (ConnectionStatsCache.getByHost(accountNo,host) < 1) return;
         TrafficCounter trafficCounter = TrafficControllerCache.getGlobalTrafficShapingHandler(getAccountId()).trafficCounter();
         if (System.currentTimeMillis() - trafficCounter.lastCumulativeTime() >= MAX_INTERVAL_REPORT_TIME_MS) {
-            synchronized (SynchronizedInternerUtils.getInterner().intern(accountNo + ":reportStat")) {
 
+            synchronized (SynchronizedInternerUtils.getInterner().intern(accountNo + ":reportStat")) {
                 if (System.currentTimeMillis() - trafficCounter.lastCumulativeTime() >= MAX_INTERVAL_REPORT_TIME_MS) {
                     long writtenBytes = trafficCounter.cumulativeWrittenBytes();
                     long readBytes = trafficCounter.cumulativeReadBytes();