优化服务监控

jetlinks · Jun 5, 2019 · e45503a · e45503a
1 parent 56873dc
commit e45503a
Show file tree

Hide file tree

Showing 3 changed files with 40 additions and 5 deletions.
diff --git a/src/main/java/org/jetlinks/gateway/monitor/GatewayServerMonitor.java b/src/main/java/org/jetlinks/gateway/monitor/GatewayServerMonitor.java
@@ -4,6 +4,7 @@
 
 import java.util.List;
 import java.util.Optional;
+import java.util.function.Consumer;
 
 
 /**
@@ -26,5 +27,7 @@ public interface GatewayServerMonitor {
 
     long getDeviceCount(String serverId);
 
+    void onServerDown(Consumer<String> listener);
+
     long getDeviceCount();
 }
diff --git a/src/main/java/org/jetlinks/gateway/monitor/RedissonGatewayServerMonitor.java b/src/main/java/org/jetlinks/gateway/monitor/RedissonGatewayServerMonitor.java
@@ -5,13 +5,16 @@
 import lombok.extern.slf4j.Slf4j;
 import org.jetlinks.core.message.codec.Transport;
 import org.redisson.api.RMap;
+import org.redisson.api.RQueue;
 import org.redisson.api.RedissonClient;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
 import java.util.*;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -34,6 +37,8 @@ public class RedissonGatewayServerMonitor implements GatewayServerMonitor {
 
     private ScheduledExecutorService executorService;
 
+    private List<Consumer<String>> downListener = new CopyOnWriteArrayList<>();
+
     private static final String transport_hosts = "transport_hosts";
     private static final String transport_all_support = "transport_supports";
     private static final String transport_connection_total = "transport_conn_total";
@@ -46,12 +51,15 @@ public String getRedisKey(String... key) {
 
     private GatewayServerInfo current;
 
+    private RQueue<String> serverDownQueue;
+
     public RedissonGatewayServerMonitor(String currentServerId, RedissonClient redissonClient, ScheduledExecutorService executorService) {
         this.allServerId = redissonClient.getMap(getRedisKey("server:all"));
         this.client = redissonClient;
         this.currentServerId = currentServerId;
         this.executorService = executorService;
-        current=newGatewayServerInfo(currentServerId);
+        current = newGatewayServerInfo(currentServerId);
+        serverDownQueue = client.getQueue("device-gateway-server-down");
     }
 
     private GatewayServerInfo newGatewayServerInfo(String serverId) {
@@ -122,7 +130,12 @@ public List<GatewayServerInfo> getAllServerInfo() {
     @Override
     public void serverOffline(String serverId) {
         log.debug("device gateway server [{}] offline ", serverId);
-        allServerId.fastRemove(serverId);
+        long number = allServerId.fastRemove(serverId);
+        if (number > 0) {
+            for (Consumer<String> consumer : downListener) {
+                consumer.accept(serverId);
+            }
+        }
     }
 
     @Override
@@ -140,6 +153,11 @@ public void reportDeviceCount(Transport transport, long count) {
                 .set(count);
     }
 
+    @Override
+    public void onServerDown(Consumer<String> listener) {
+        downListener.add(listener);
+    }
+
     @Override
     public long getDeviceCount(String serverId) {
         return client.<Transport>getSet(getRedisKey(transport_all_support, serverId))
@@ -169,7 +187,8 @@ protected void clean() {
 
     @PreDestroy
     public void shutdown() {
-        allServerId.remove(currentServerId);
+        allServerId.fastRemove(currentServerId);
+        serverDownQueue.add(currentServerId);
         clean();
     }
 
@@ -179,7 +198,7 @@ protected synchronized void doStartup() {
         }
         startup = true;
         allServerId.put(currentServerId, System.currentTimeMillis());
-
+        serverDownQueue.remove(currentServerId);
         executorService.scheduleAtFixedRate(() -> {
             log.debug("device gateway server [{}] keepalive", currentServerId);
             allServerId.put(currentServerId, System.currentTimeMillis());
@@ -190,7 +209,14 @@ protected synchronized void doStartup() {
                     .map(Map.Entry::getKey)
                     .filter(Objects::nonNull)
                     .forEach(this::serverOffline);
-
+            //触发服务下线事件
+            for (String offlineServer = serverDownQueue.poll()
+                 ; offlineServer != null && (!currentServerId.equals(offlineServer))
+                    ; offlineServer = serverDownQueue.poll()) {
+                for (Consumer<String> listener : downListener) {
+                    listener.accept(offlineServer);
+                }
+            }
         }, 1, Math.max(1, timeToLive - 1), TimeUnit.SECONDS);
     }
 

diff --git a/src/test/java/org/jetlinks/gateway/monitor/RedissonGatewayServerMonitorTest.java b/src/test/java/org/jetlinks/gateway/monitor/RedissonGatewayServerMonitorTest.java
@@ -9,6 +9,7 @@
 import org.junit.Test;
 
 import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicReference;
 
 public class RedissonGatewayServerMonitorTest {
 
@@ -29,11 +30,16 @@ public void after() {
     @Test
     public void testOnlineOffline() {
         monitor.registerTransport(Transport.MQTT);
+        AtomicReference<String> reference=new AtomicReference<>();
+
+        monitor.onServerDown(reference::set);
+
         GatewayServerInfo info = monitor.getServerInfo("test").orElse(null);
         Assert.assertNotNull(info);
 
         Assert.assertEquals(info.getId(), "test");
         monitor.serverOffline("test");
+        Assert.assertNotNull(reference.get());
         info = monitor.getServerInfo("test").orElse(null);
         Assert.assertNull(info);
     }