统一节点请求生命周期并释放重连资源

src/main/java/com/lion/lionwebsite/Service/RemoteService.java

@@ -44,9 +44,9 @@ import java.util.concurrent.atomic.AtomicBoolean;
 @Slf4j
 public class RemoteService {
 
-    ChannelFuture channelFuture;
+    volatile ChannelFuture channelFuture;
 
-    Channel channel;
+    volatile Channel channel;
 
     @Value("${remote.ip:5.255.110.45}")
     String ip;
@@ -62,7 +62,12 @@ public class RemoteService {
     ConcurrentHashMap<Integer, CopyOnWriteArrayList<CompletableFuture<String>>> retryStatusWaiters =
             new ConcurrentHashMap<>();
 
-    EventLoop eventLoopGroup = new DefaultEventLoop();
+    final EventLoop eventLoopGroup = new DefaultEventLoop();
+    final EventLoopGroup networkGroup = new NioEventLoopGroup(2);
+    final AtomicBoolean connecting = new AtomicBoolean();
+    final AtomicBoolean monitoring = new AtomicBoolean();
+    volatile boolean stopping;
+    volatile ServerSocket monitorSocket;
 
     ExecutorService downloadThread = Executors.newCachedThreadPool();
 
@@ -89,20 +94,22 @@ public class RemoteService {
 
     @PostConstruct
     void init() {
-        if(!initChannel()){     //如果远程服务器连接失败，则开启本地监听
+        initChannel();
-            monitor = new Thread(this::monitorFunc);
-            monitor.start();
-        }
     }
 
     public boolean initChannel(){
+        if (stopping || !connecting.compareAndSet(false, true))
+            return !isDead();
         try {
+            if (!isDead())
+                return true;
             int i;
             for(i=0; i<20; i++) {
                 try {
                     channelFuture = new Bootstrap()
                             .channel(NioSocketChannel.class)
-                            .group(new NioEventLoopGroup())
+                            .group(networkGroup)
+                            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3_000)
                             .handler(new ChannelInitializer<NioSocketChannel>() {
                                 @Override
                                 protected void initChannel(NioSocketChannel channel) {
@@ -114,9 +121,14 @@ public class RemoteService {
                                 }
                             }).connect(new InetSocketAddress(ip, port + i)).sync();
                     break;
-                }catch (Exception e){
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    return false;
+                } catch (Exception e) {
                     log.error("连接storageNode失败，端口偏移量(重试次数):{}", i);
                 }
+                if (stopping)
+                    return false;
             }
 
             //超过二十次连不上，主动抛出错误，由下方catch
@@ -124,11 +136,16 @@ public class RemoteService {
                 throw new Exception();
             }
 
+            if (stopping) {
+                channelFuture.channel().close();
+                return false;
+            }
             log.info("connect success");
             if(pushService != null)
                 pushService.storageNodeOnline();
 
             channel = channelFuture.channel();
+            closeMonitorSocket();
             channel.writeAndFlush(new IdentityMessage("lionwebsite"));
 
             //子节点上线时，发送未完成的任务
@@ -138,6 +155,10 @@ public class RemoteService {
         }catch (Exception e){
             log.error("connect node failed, wait for node back online", e);
             return false;
+        } finally {
+            connecting.set(false);
+            if (isDead())
+                startMonitor();
         }
     }
 
@@ -147,7 +168,7 @@ public class RemoteService {
             return -2;
         }
 
-        channelFuture.channel().close();
+        channelFuture.channel().close().awaitUninterruptibly();
 
         if(initChannel()){
             return 0;
@@ -157,34 +178,49 @@ public class RemoteService {
     }
 
     public byte checkAvailability(){
-        AvailableCheckMessage acm = new AvailableCheckMessage();
+        return sendRequest(new AvailableCheckMessage(), 10, TimeUnit.SECONDS);
-        acm.setMessageId(atomicInteger.getAndIncrement());
-
-        channel.writeAndFlush(acm);
-        DefaultPromise<AbstractMessage> promise = new DefaultPromise<>(eventLoopGroup);
-        promiseHashMap.put(acm.messageId, promise);
-        try {
-            boolean result = promise.await(10, TimeUnit.SECONDS);
-            if(result){
-                ResponseMessage rsm = (ResponseMessage)promise.getNow();
-                return rsm.getResult();
     }
-            else return -1;
+
-        }catch (InterruptedException e){
+    byte sendRequest(AbstractMessage message, long timeout, TimeUnit unit) {
-            log.warn("checkAvailability interrupted", e);
+        Channel target = channel;
+        if (stopping || target == null || !target.isActive())
+            return -1;
+        message.setMessageId(atomicInteger.getAndIncrement());
+        DefaultPromise<AbstractMessage> promise = new DefaultPromise<>(eventLoopGroup);
+        promiseHashMap.put(message.messageId, promise);
+        try {
+            target.writeAndFlush(message).addListener(future -> {
+                if (!future.isSuccess())
+                    promise.tryFailure(future.cause() == null ? new IOException("节点发送失败") : future.cause());
+            });
+            if (promise.await(timeout, unit) && promise.isSuccess()
+                    && promise.getNow() instanceof ResponseMessage response)
+                return response.getResult();
+            return -1;
+        } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
             return -1;
+        } catch (Exception e) {
+            log.warn("节点请求失败: messageId={}", message.messageId, e);
+            return -1;
+        } finally {
+            promiseHashMap.remove(message.messageId, promise);
         }
     }
 
     /** 请求将当前全部订阅状态异步同步到存储节点，短时间内的多次请求会合并。 */
     public void requestSubscriptionSync() {
-        if (!subscriptionSyncEnabled)
+        if (stopping || !subscriptionSyncEnabled)
             return;
         subscriptionSyncQueued.set(true);
         if (!subscriptionSyncRunning.compareAndSet(false, true))
             return;
+        try {
             subscriptionSyncExecutor.execute(this::drainSubscriptionSyncQueue);
+        } catch (java.util.concurrent.RejectedExecutionException e) {
+            subscriptionSyncRunning.set(false);
+            if (!stopping) throw e;
+        }
     }
 
     @Scheduled(fixedDelayString = "${subscription.standby.retry-interval-ms:60000}")
@@ -207,32 +243,15 @@ public class RemoteService {
     }
 
     private void syncSubscriptionSnapshotOnce() {
-        SubscriptionSnapshotMessage message = null;
-        DefaultPromise<AbstractMessage> promise = null;
         try {
-            message = subscriptionStandbySnapshotService.build();
+            SubscriptionSnapshotMessage message = subscriptionStandbySnapshotService.build();
-            message.setMessageId(atomicInteger.getAndIncrement());
+            byte result = sendRequest(message, 30, TimeUnit.SECONDS);
-            promise = new DefaultPromise<>(eventLoopGroup);
+            if (result == 0 || result == 3)
-            promiseHashMap.put(message.messageId, promise);
+                log.info("订阅快照同步完成 revision={} result={}", shortRevision(message.getRevision()), result);
-            channel.writeAndFlush(message);
-            if (promise.await(30, TimeUnit.SECONDS)) {
-                AbstractMessage reply = promise.getNow();
-                if (reply instanceof ResponseMessage response && (response.getResult() == 0 || response.getResult() == 3))
-                    log.info("订阅快照同步完成 revision={} result={}", shortRevision(message.getRevision()), response.getResult());
             else
-                    log.warn("订阅快照同步失败 revision={} result={}", shortRevision(message.getRevision()),
+                log.warn("订阅快照同步失败或超时 revision={} result={}", shortRevision(message.getRevision()), result);
-                            reply instanceof ResponseMessage response ? response.getResult() : "invalid-response");
-            } else {
-                log.warn("订阅快照同步超时 revision={}", shortRevision(message.getRevision()));
-            }
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-            log.warn("订阅快照同步线程被中断");
         } catch (Exception e) {
             log.warn("生成或发送订阅快照失败: {}", e.getMessage());
-        } finally {
-            if (message != null && promise != null)
-                promiseHashMap.remove(message.messageId, promise);
         }
     }
 
@@ -241,8 +260,24 @@ public class RemoteService {
     }
 
     @PreDestroy
-    void shutdownSubscriptionSync() {
+    void shutdownResources() {
+        stopping = true;
+        closeMonitorSocket();
+        if (monitor != null) monitor.interrupt();
+        failPendingRequests();
+        if (channel != null) channel.close();
         subscriptionSyncExecutor.shutdownNow();
+        downloadThread.shutdownNow();
+        networkGroup.shutdownGracefully();
+        eventLoopGroup.shutdownGracefully();
+    }
+
+    private void failPendingRequests() {
+        promiseHashMap.forEach((id, promise) -> promise.tryFailure(new IOException("节点连接已关闭")));
+        promiseHashMap.clear();
+        retryStatusWaiters.forEach((gid, waiters) ->
+                waiters.forEach(waiter -> waiter.completeExceptionally(new IOException("节点连接已关闭"))));
+        retryStatusWaiters.clear();
     }
 
     public boolean isDead(){
@@ -266,28 +301,9 @@ public class RemoteService {
         GalleryTask galleryTask = new GalleryTask();
         galleryTask.setGid(gallery.getGid());
         galleryTask.setName(gallery.getName());
-
+        DownloadPostMessage message = new DownloadPostMessage();
-        DownloadPostMessage dpm = new DownloadPostMessage();
+        message.setGalleryTask(galleryTask);
-        dpm.messageId = atomicInteger.getAndIncrement();
+        return sendRequest(message, 10, TimeUnit.SECONDS);
-        dpm.setGalleryTask(galleryTask);
-
-        DefaultPromise<AbstractMessage> promise = new DefaultPromise<>(eventLoopGroup);
-        promiseHashMap.put(dpm.messageId, promise);
-        channel.writeAndFlush(dpm);
-        try {
-            boolean result = promise.await(10, TimeUnit.SECONDS);
-            if(result){
-                ResponseMessage rsm = (ResponseMessage)promise.getNow();
-                return rsm.getResult();
-            }
-            else return -1;
-        }catch (InterruptedException e){
-            log.warn("addGalleryToQueue interrupted", e);
-            Thread.currentThread().interrupt();
-            return -1;
-        }finally {
-            promiseHashMap.remove(dpm.messageId, promise);
-        }
     }
 
     public RetryResult retryGallery(Gallery gallery){
@@ -325,49 +341,51 @@ public class RemoteService {
     public record RetryResult(boolean success, String message) {}
 
     public byte deleteGallery(Gallery gallery){
-        DeleteGalleryMessage dgm = new DeleteGalleryMessage();
+        DeleteGalleryMessage message = new DeleteGalleryMessage();
-        dgm.setGalleryName(gallery.getName());
+        message.setGalleryName(gallery.getName());
-        dgm.messageId = atomicInteger.getAndIncrement();
+        return sendRequest(message, 10, TimeUnit.SECONDS);
+    }
 
-        channel.writeAndFlush(dgm);
+    private void startMonitor() {
-        DefaultPromise<AbstractMessage> promise = new DefaultPromise<>(eventLoopGroup);
+        if (stopping || !monitoring.compareAndSet(false, true))
-        promiseHashMap.put(dgm.messageId, promise);
+            return;
-        try{
+        monitor = new Thread(this::monitorFunc, "storage-node-monitor");
-            boolean result = promise.await(10, TimeUnit.SECONDS);
+        monitor.setDaemon(true);
-            if(result){
+        monitor.start();
-                ResponseMessage rsm = (ResponseMessage) promise.getNow();
+    }
-                return rsm.getResult();
+
-            }else return -1;
+    private void closeMonitorSocket() {
-        }catch (InterruptedException e){
+        ServerSocket socket = monitorSocket;
-            log.warn("deleteGallery interrupted", e);
+        if (socket != null) {
-            Thread.currentThread().interrupt();
+            try { socket.close(); }
-            return -1;
+            catch (IOException e) { log.debug("关闭节点监听失败", e); }
         }
     }
 
     public void monitorFunc(){
-        int real_port = CustomUtil._findIdlePort(port + 1);
+        try (ServerSocket socket = new ServerSocket(CustomUtil._findIdlePort(port + 1))) {
-        log.info("监听端口： {}等待节点上线", real_port);
+            monitorSocket = socket;
-        try(ServerSocket socket = new ServerSocket(real_port)) {
+            if (stopping || !isDead())
-            Socket client;
+                return;
-            while(true){
+            log.info("监听端口： {}等待节点上线", socket.getLocalPort());
-                client = socket.accept();
+            while (!stopping) {
-
+                try (Socket client = socket.accept()) {
-                if(client.getInetAddress().getHostAddress().equals(ip)){
+                    if (!client.getInetAddress().getHostAddress().equals(ip))
-                    //连接之后发送lionwebsite，否则存储节点不能确认这个端口是否有效
+                        continue;
-                    OutputStream outputStream = client.getOutputStream();
+                    OutputStream output = client.getOutputStream();
-                    outputStream.write("lionwebsite".getBytes());
+                    output.write("lionwebsite".getBytes(java.nio.charset.StandardCharsets.UTF_8));
-                    outputStream.flush();
+                    output.flush();
-                    outputStream.close();
+                    client.shutdownOutput();
-                    log.info("尝试连接");
+                    if (initChannel())
-                    initChannel();
-                    client.close();
-                    socket.close();
                         break;
                 }
             }
         } catch (IOException e) {
-            throw new RuntimeException(e);
+            if (!stopping && isDead())
+                log.warn("等待节点上线失败", e);
+        } finally {
+            monitorSocket = null;
+            monitoring.set(false);
         }
     }
 
@@ -416,7 +434,7 @@ public class RemoteService {
             else if(msg instanceof ResponseMessage rsm) {
                 Promise<AbstractMessage> promise = promiseHashMap.remove(rsm.messageId);
                 if(promise != null)
-                    promise.setSuccess(rsm);
+                    promise.trySuccess(rsm);
                 else
                     log.warn("收到无等待者的响应消息: messageId={}", rsm.messageId);
             }
@@ -424,13 +442,12 @@ public class RemoteService {
 
         @Override
         public void channelUnregistered(ChannelHandlerContext ctx) {
-            log.info("{}", ctx.channel());
+            if (ctx.channel() == channel) {
-            log.info("{}", channel);
+                failPendingRequests();
-            if(ctx.channel() != null && ctx.channel().remoteAddress().toString().equals(channel.remoteAddress().toString())){
+                if (!stopping) {
-                log.info("activate monitor thread, waiting for node back online");
                     pushService.storageNodeOffline();
-                monitor = new Thread(RemoteService.this::monitorFunc);
+                    startMonitor();
-                monitor.start();
+                }
             }
         }
     }

src/test/java/com/lion/lionwebsite/Service/RemoteServiceTest.java

@@ -0,0 +1,55 @@
+package com.lion.lionwebsite.Service;
+
+import com.lion.lionwebsite.Dao.normal.GalleryMapper;
+import com.lion.lionwebsite.Message.*;
+import io.netty.channel.*;
+import io.netty.channel.embedded.EmbeddedChannel;
+import org.junit.jupiter.api.Test;
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class RemoteServiceTest {
+    private RemoteService service() {
+        return new RemoteService(mock(GalleryMapper.class), mock(PushService.class),
+                mock(WebSocketService.class), mock(SubscriptionStandbySnapshotService.class));
+    }
+
+    @Test void immediateResponseHasRegisteredWaiter() {
+        RemoteService service = service();
+        EmbeddedChannel channel = new EmbeddedChannel(new ChannelOutboundHandlerAdapter() {
+            @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
+                AbstractMessage request = (AbstractMessage) msg;
+                ctx.fireChannelRead(new ResponseMessage(request.messageId, (byte) 0));
+                promise.setSuccess();
+            }
+        }, service.new MyChannelInboundHandlerAdapter());
+        service.channel = channel;
+        try {
+            assertEquals(0, service.checkAvailability());
+            assertTrue(service.promiseHashMap.isEmpty());
+        } finally { service.shutdownResources(); channel.finishAndReleaseAll(); }
+    }
+
+    @Test void timeoutAndWriteFailureRemoveWaiters() {
+        RemoteService service = service();
+        EmbeddedChannel channel = new EmbeddedChannel(new ChannelOutboundHandlerAdapter() {
+            @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
+                promise.setSuccess(); // no reply
+            }
+        });
+        service.channel = channel;
+        try {
+            assertEquals(-1, service.sendRequest(new AvailableCheckMessage(), 1, TimeUnit.MILLISECONDS));
+            assertTrue(service.promiseHashMap.isEmpty());
+            channel.pipeline().addLast(new ChannelOutboundHandlerAdapter() {
+                @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
+                    promise.setFailure(new IOException("test failure"));
+                }
+            });
+            assertEquals(-1, service.sendRequest(new AvailableCheckMessage(), 1, TimeUnit.SECONDS));
+            assertTrue(service.promiseHashMap.isEmpty());
+        } finally { service.shutdownResources(); channel.finishAndReleaseAll(); }
+    }
+}