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

2026-09-08 09:25:45 +08:00
parent 1e6e3a1557
commit b15eeaf45e
2 changed files with 185 additions and 113 deletions
@@ -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()){     //如果远程服务器连接失败，则开启本地监听
-            monitor = new Thread(this::monitorFunc);
-            monitor.start();
-        }
+        initChannel();
    }

    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();
-        acm.setMessageId(atomicInteger.getAndIncrement());
+        return sendRequest(new AvailableCheckMessage(), 10, TimeUnit.SECONDS);
+    }

-        channel.writeAndFlush(acm);
+    byte sendRequest(AbstractMessage message, long timeout, TimeUnit unit) {
+        Channel target = channel;
+        if (stopping || target == null || !target.isActive())
+            return -1;
+        message.setMessageId(atomicInteger.getAndIncrement());
        DefaultPromise<AbstractMessage> promise = new DefaultPromise<>(eventLoopGroup);
-        promiseHashMap.put(acm.messageId, promise);
+        promiseHashMap.put(message.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){
-            log.warn("checkAvailability interrupted", e);
+            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;
-        subscriptionSyncExecutor.execute(this::drainSubscriptionSyncQueue);
+        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();
-            message.setMessageId(atomicInteger.getAndIncrement());
-            promise = new DefaultPromise<>(eventLoopGroup);
-            promiseHashMap.put(message.messageId, promise);
-            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()),
-                            reply instanceof ResponseMessage response ? response.getResult() : "invalid-response");
-            } else {
-                log.warn("订阅快照同步超时 revision={}", shortRevision(message.getRevision()));
-            }
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-            log.warn("订阅快照同步线程被中断");
+            SubscriptionSnapshotMessage message = subscriptionStandbySnapshotService.build();
+            byte result = sendRequest(message, 30, TimeUnit.SECONDS);
+            if (result == 0 || result == 3)
+                log.info("订阅快照同步完成 revision={} result={}", shortRevision(message.getRevision()), result);
+            else
+                log.warn("订阅快照同步失败或超时 revision={} result={}", shortRevision(message.getRevision()), result);
        } 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 dpm = new DownloadPostMessage();
-        dpm.messageId = atomicInteger.getAndIncrement();
-        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);
-        }
+        DownloadPostMessage message = new DownloadPostMessage();
+        message.setGalleryTask(galleryTask);
+        return sendRequest(message, 10, TimeUnit.SECONDS);
    }

    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();
-        dgm.setGalleryName(gallery.getName());
-        dgm.messageId = atomicInteger.getAndIncrement();
+        DeleteGalleryMessage message = new DeleteGalleryMessage();
+        message.setGalleryName(gallery.getName());
+        return sendRequest(message, 10, TimeUnit.SECONDS);
+    }

-        channel.writeAndFlush(dgm);
-        DefaultPromise<AbstractMessage> promise = new DefaultPromise<>(eventLoopGroup);
-        promiseHashMap.put(dgm.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){
-            log.warn("deleteGallery interrupted", e);
-            Thread.currentThread().interrupt();
-            return -1;
+    private void startMonitor() {
+        if (stopping || !monitoring.compareAndSet(false, true))
+            return;
+        monitor = new Thread(this::monitorFunc, "storage-node-monitor");
+        monitor.setDaemon(true);
+        monitor.start();
+    }
+
+    private void closeMonitorSocket() {
+        ServerSocket socket = monitorSocket;
+        if (socket != null) {
+            try { socket.close(); }
+            catch (IOException e) { log.debug("关闭节点监听失败", e); }
        }
    }

    public void monitorFunc(){
-        int real_port = CustomUtil._findIdlePort(port + 1);
-        log.info("监听端口： {}等待节点上线", real_port);
-        try(ServerSocket socket = new ServerSocket(real_port)) {
-            Socket client;
-            while(true){
-                client = socket.accept();
-
-                if(client.getInetAddress().getHostAddress().equals(ip)){
-                    //连接之后发送lionwebsite，否则存储节点不能确认这个端口是否有效
-                    OutputStream outputStream = client.getOutputStream();
-                    outputStream.write("lionwebsite".getBytes());
-                    outputStream.flush();
-                    outputStream.close();
-                    log.info("尝试连接");
-                    initChannel();
-                    client.close();
-                    socket.close();
-                    break;
+        try (ServerSocket socket = new ServerSocket(CustomUtil._findIdlePort(port + 1))) {
+            monitorSocket = socket;
+            if (stopping || !isDead())
+                return;
+            log.info("监听端口： {}等待节点上线", socket.getLocalPort());
+            while (!stopping) {
+                try (Socket client = socket.accept()) {
+                    if (!client.getInetAddress().getHostAddress().equals(ip))
+                        continue;
+                    OutputStream output = client.getOutputStream();
+                    output.write("lionwebsite".getBytes(java.nio.charset.StandardCharsets.UTF_8));
+                    output.flush();
+                    client.shutdownOutput();
+                    if (initChannel())
+                        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());
-            log.info("{}", channel);
-            if(ctx.channel() != null && ctx.channel().remoteAddress().toString().equals(channel.remoteAddress().toString())){
-                log.info("activate monitor thread, waiting for node back online");
-                pushService.storageNodeOffline();
-                monitor = new Thread(RemoteService.this::monitorFunc);
-                monitor.start();
+            if (ctx.channel() == channel) {
+                failPendingRequests();
+                if (!stopping) {
+                    pushService.storageNodeOffline();
+                    startMonitor();
+                }
            }
        }
    }
@@ -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(); }
+    }
+}