From b15eeaf45e2d6e295b7a9bb3188b1fb071d0b8a0 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 8 Sep 2026 09:25:45 +0800 Subject: [PATCH] =?UTF-8?q?=E7=BB=9F=E4=B8=80=E8=8A=82=E7=82=B9=E8=AF=B7?= =?UTF-8?q?=E6=B1=82=E7=94=9F=E5=91=BD=E5=91=A8=E6=9C=9F=E5=B9=B6=E9=87=8A?= =?UTF-8?q?=E6=94=BE=E9=87=8D=E8=BF=9E=E8=B5=84=E6=BA=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../lionwebsite/Service/RemoteService.java | 243 ++++++++++-------- .../Service/RemoteServiceTest.java | 55 ++++ 2 files changed, 185 insertions(+), 113 deletions(-) create mode 100644 src/test/java/com/lion/lionwebsite/Service/RemoteServiceTest.java diff --git a/src/main/java/com/lion/lionwebsite/Service/RemoteService.java b/src/main/java/com/lion/lionwebsite/Service/RemoteService.java index f7e9cd2..0da5184 100644 --- a/src/main/java/com/lion/lionwebsite/Service/RemoteService.java +++ b/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>> 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() { @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 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 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 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 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 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(); + } } } } diff --git a/src/test/java/com/lion/lionwebsite/Service/RemoteServiceTest.java b/src/test/java/com/lion/lionwebsite/Service/RemoteServiceTest.java new file mode 100644 index 0000000..045460e --- /dev/null +++ b/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(); } + } +}