update

Author: rockit-ba

src/main/java/cn/pan/connor/core/handle/resp/HeartbeatTimeoutRespHandle.java

@@ -2,6 +2,7 @@
 
 import cn.pan.connor.common.utils.JsonUtil;
 import cn.pan.connor.core.model.response.HeartbeatTimeoutResponse;
+import cn.pan.connor.core.transport.ClientCache;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
 import lombok.extern.slf4j.Slf4j;
@@ -20,6 +21,6 @@ public class HeartbeatTimeoutRespHandle extends SimpleChannelInboundHandler<Hear
     @Override
     protected void channelRead0(ChannelHandlerContext context, HeartbeatTimeoutResponse response) {
         log.info("accept connor service heartbeatTimeoutRespHandle : {}", JsonUtil.toStr(response));
-
+        ClientCache.ServiceCache.timeoutUpdate(response.getTimeoutServiceIds());
     }
 }

src/main/java/cn/pan/connor/core/model/response/HeartbeatTimeoutResponse.java

@@ -15,6 +15,6 @@
  */
 @Data
 public class HeartbeatTimeoutResponse {
-    @SerializedName("service_ids")
-    private List<String> serviceIds;
+    @SerializedName("timeout_service_ids")
+    private List<String> timeoutServiceIds;
 }

src/main/java/cn/pan/connor/core/transport/ClientCache.java

@@ -12,6 +12,7 @@
 import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 /**
  * 客户端缓存
@@ -29,6 +30,19 @@ public static class ServiceCache {
         private static final SynchronousQueue<String> SERVICES_BLOCK = new SynchronousQueue<>();
         private static final ConcurrentSkipListMap<String, List<NewService>> SERVICES = new ConcurrentSkipListMap<>();
 
+        /**
+         * 根据server的推送，移除超时的实例
+         * @param timeoutServiceIds 超时的实例id
+         */
+        public static void timeoutUpdate(List<String> timeoutServiceIds) {
+            SERVICES.forEach((k,v) -> {
+                List<NewService> services = v.stream()
+                        .filter(service -> !timeoutServiceIds.contains(service.getId()))
+                        .collect(Collectors.toList());
+                SERVICES.put(k,services);
+            });
+        }
+
         /**
          * 根据server的推送更新本地缓存
          */

src/test/java/cn/pan/connor/ConnorStarterApplicationTests.java

@@ -1,13 +1,17 @@
 package cn.pan.connor;
 
 import cn.hutool.core.date.DateUtil;
+import cn.pan.connor.core.model.NewService;
 import lombok.extern.slf4j.Slf4j;
 import org.junit.jupiter.api.Test;
 
+import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 @Slf4j
 //@SpringBootTest
@@ -43,10 +47,40 @@ void test_01() {
 			e.printStackTrace();
 		}
 	}
+	public static  ConcurrentSkipListMap<String, List<NewService>> SERVICES = new ConcurrentSkipListMap<>();
 
+	public static void timeoutUpdate(List<String> serviceIds) {
+		SERVICES.forEach((k,v) -> {
+			List<NewService> services = v.stream()
+					.filter(service -> !serviceIds.contains(service.getId()))
+					.collect(Collectors.toList());
+			SERVICES.put(k,services);
+		});
+	}
 
+	@Test
+	void test_02() {
+		// 为SERVICES初始化数据
+		SERVICES.put("1",
+				List.of(
+						NewService.builder().id("1").build(),
+						NewService.builder().id("2").build(),
+						NewService.builder().id("3").build()));
+		SERVICES.put("2",
+				List.of(
+						NewService.builder().id("1").build(),
+						NewService.builder().id("2").build(),
+						NewService.builder().id("3").build()));
+		// 创建一个ID list
+		List<String> serviceIds = List.of("2");
+		// 调用timeoutUpdate方法
+		timeoutUpdate(serviceIds);
+		// 打印结果
+		SERVICES.forEach((k,v) -> {
+			System.out.println(k + ":" + v);
+		});
 
-
+	}
 
 
 }