From 6e9bd1345187f33c00dc814feb36b71790afc1c5 Mon Sep 17 00:00:00 2001 From: Xiaoyumuxi <3075514079@qq.com> Date: Mon, 14 Sep 2026 15:54:04 +0800 Subject: [PATCH] fix: harden service startup and discovery lifecycle --- .github/workflows/ci.yml | 7 +- .../com/xiaoyu/rpc/core/client/RpcClient.java | 23 ++- .../rpc/core/registry/ServiceDiscovery.java | 11 +- .../registry/nacos/NacosServiceDiscovery.java | 136 ++++++++------ .../com/xiaoyu/rpc/core/server/RpcServer.java | 38 +++- .../rpc/core/transport/TransportServer.java | 19 +- .../nacos/NacosServiceDiscoveryTest.java | 173 ++++++++++-------- .../core/server/RpcServerLifecycleTest.java | 96 ++++++++++ .../com/xiaoyu/rpc/provider/ProviderApp.java | 7 +- rpc-spring-boot-starter/pom.xml | 11 +- .../rpc/spring/RpcAutoConfiguration.java | 82 +++++---- .../rpc/spring/RpcServerLifecycleTest.java | 50 +++++ .../transport/netty/NettyTransportServer.java | 14 +- 13 files changed, 472 insertions(+), 195 deletions(-) create mode 100644 rpc-core/src/test/java/com/xiaoyu/rpc/core/server/RpcServerLifecycleTest.java create mode 100644 rpc-spring-boot-starter/src/test/java/com/xiaoyu/rpc/spring/RpcServerLifecycleTest.java diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7a39192..603b665 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -49,8 +49,11 @@ jobs: distribution: 'temurin' cache: 'maven' - - name: Run core and transport unit tests - run: mvn -B -ntp test -pl rpc-core,rpc-transport-netty -am -Drpc.registry=local + - name: Run core, transport and starter unit tests + run: >- + mvn -B -ntp test + -pl rpc-core,rpc-transport-netty,rpc-spring-boot-starter -am + -Drpc.registry=local - name: Upload unit test and coverage reports if: always() diff --git a/rpc-core/src/main/java/com/xiaoyu/rpc/core/client/RpcClient.java b/rpc-core/src/main/java/com/xiaoyu/rpc/core/client/RpcClient.java index 686c258..deb0bea 100644 --- a/rpc-core/src/main/java/com/xiaoyu/rpc/core/client/RpcClient.java +++ b/rpc-core/src/main/java/com/xiaoyu/rpc/core/client/RpcClient.java @@ -76,8 +76,29 @@ public CompletableFuture sendRequest(RpcRequest request, Class return @Override public void close() { - if (closed.compareAndSet(false, true)) { + if (!closed.compareAndSet(false, true)) { + return; + } + + RuntimeException failure = null; + try { transportClient.close(); + } catch (RuntimeException e) { + failure = e; + } + + try { + serviceDiscovery.close(); + } catch (Exception e) { + if (failure == null) { + failure = new RuntimeException("关闭 ServiceDiscovery 失败", e); + } else { + failure.addSuppressed(e); + } + } + + if (failure != null) { + throw failure; } } } diff --git a/rpc-core/src/main/java/com/xiaoyu/rpc/core/registry/ServiceDiscovery.java b/rpc-core/src/main/java/com/xiaoyu/rpc/core/registry/ServiceDiscovery.java index 6df4169..2484caf 100644 --- a/rpc-core/src/main/java/com/xiaoyu/rpc/core/registry/ServiceDiscovery.java +++ b/rpc-core/src/main/java/com/xiaoyu/rpc/core/registry/ServiceDiscovery.java @@ -5,15 +5,20 @@ import java.net.InetSocketAddress; /** - * 服务发现接口 + * 服务发现接口。 */ @SPI -public interface ServiceDiscovery { +public interface ServiceDiscovery extends AutoCloseable { /** - * 查找服务地址 + * 查找服务地址。 * * @param serviceName 服务名称 * @return 服务地址 */ InetSocketAddress lookupService(String serviceName); + + @Override + default void close() { + // 默认实现无外部资源需要释放。 + } } diff --git a/rpc-core/src/main/java/com/xiaoyu/rpc/core/registry/nacos/NacosServiceDiscovery.java b/rpc-core/src/main/java/com/xiaoyu/rpc/core/registry/nacos/NacosServiceDiscovery.java index c53ebda..724dec3 100644 --- a/rpc-core/src/main/java/com/xiaoyu/rpc/core/registry/nacos/NacosServiceDiscovery.java +++ b/rpc-core/src/main/java/com/xiaoyu/rpc/core/registry/nacos/NacosServiceDiscovery.java @@ -2,34 +2,31 @@ import com.alibaba.nacos.api.exception.NacosException; import com.alibaba.nacos.api.naming.NamingService; -import com.alibaba.nacos.api.naming.pojo.Instance; -import com.alibaba.nacos.api.naming.listener.EventListener; import com.alibaba.nacos.api.naming.listener.Event; +import com.alibaba.nacos.api.naming.listener.EventListener; import com.alibaba.nacos.api.naming.listener.NamingEvent; +import com.alibaba.nacos.api.naming.pojo.Instance; import com.xiaoyu.rpc.common.extension.ExtensionLoader; import com.xiaoyu.rpc.core.config.RpcConfig; import com.xiaoyu.rpc.core.loadbalancer.LoadBalancer; import com.xiaoyu.rpc.core.registry.ServiceDiscovery; -import lombok.extern.slf4j.Slf4j; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.net.InetSocketAddress; import java.util.List; import java.util.Map; -import java.util.Set; import java.util.concurrent.ConcurrentHashMap; -import java.util.stream.Collectors; +import java.util.concurrent.atomic.AtomicBoolean; public class NacosServiceDiscovery implements ServiceDiscovery { private static final Logger log = LoggerFactory.getLogger(NacosServiceDiscovery.class); private final NamingService namingService; private final LoadBalancer loadBalancer; - // 本地缓存，用于容错和防抖 - private static final java.util.Map> serviceCache = new java.util.concurrent.ConcurrentHashMap<>(); - // 已订阅的服务集合 - private static final java.util.Set subscribedServices = java.util.concurrent.ConcurrentHashMap.newKeySet(); + private final Map> serviceCache = new ConcurrentHashMap<>(); + private final Map subscriptions = new ConcurrentHashMap<>(); + private final AtomicBoolean closed = new AtomicBoolean(false); public NacosServiceDiscovery() { this(NacosUtils.getNacosNamingService(), @@ -44,73 +41,96 @@ public NacosServiceDiscovery() { @Override public InetSocketAddress lookupService(String serviceName) { + if (closed.get()) { + throw new IllegalStateException("NacosServiceDiscovery 已关闭"); + } + try { - // 第一次查找时订阅服务变更 - if (subscribedServices.add(serviceName)) { - // add 返回 true 说明此前未订阅，避免同一个服务被重复订阅 - subscribeService(serviceName); - } + ensureSubscribed(serviceName); - // 优先从 Nacos 拉取最新实例列表 List instances = namingService.getAllInstances(serviceName); - - if (instances.isEmpty()) { - log.warn("Nacos 返回实例列表为空，尝试使用本地缓存: {}", serviceName); - instances = serviceCache.get(serviceName); - } else { - // 更新本地缓存 - serviceCache.put(serviceName, instances); - } - if (instances == null || instances.isEmpty()) { - log.error("未找到服务且本地无缓存: {}", serviceName); + // 注册中心明确返回空实例，代表当前服务已下线；不能继续使用旧缓存。 + serviceCache.remove(serviceName); throw new RuntimeException("未找到服务: " + serviceName); } - // 转换 Instance 列表为 String 列表 (ip:port) - List addressList = instances.stream() - .map(instance -> instance.getIp() + ":" + instance.getPort()) - .collect(java.util.stream.Collectors.toList()); - - // 负载均衡选择 - String targetAddress = loadBalancer.select(addressList); - log.info("负载均衡选择服务地址: {}", targetAddress); - - String[] array = targetAddress.split(":"); - return new InetSocketAddress(array[0], Integer.parseInt(array[1])); - + updateCache(serviceName, instances); + return selectAddress(instances); } catch (NacosException e) { - log.error("获取服务实例时发生网络异常，尝试回滚到本地缓存:", e); - // Nacos 短暂不可用时，优先用最近一次成功拉取到的实例兜底 + // 只有 Nacos 网络/协议异常时，才允许使用最后一次成功结果容错。 + log.error("获取服务实例时发生 Nacos 异常，尝试使用最近一次成功缓存: {}", serviceName, e); List cachedInstances = serviceCache.get(serviceName); if (cachedInstances != null && !cachedInstances.isEmpty()) { - List addressList = cachedInstances.stream() - .map(instance -> instance.getIp() + ":" + instance.getPort()) - .collect(java.util.stream.Collectors.toList()); - String targetAddress = loadBalancer.select(addressList); - String[] array = targetAddress.split(":"); - return new InetSocketAddress(array[0], Integer.parseInt(array[1])); + return selectAddress(cachedInstances); } throw new RuntimeException("服务发现失败且无缓存可用: " + serviceName, e); } } - /** - * 订阅服务变更，实现本地缓存的实时更新 - */ - private void subscribeService(String serviceName) throws NacosException { - namingService.subscribe(serviceName, new EventListener() { - @Override - public void onEvent(Event event) { - if (event instanceof NamingEvent) { - NamingEvent namingEvent = (NamingEvent) event; - List instances = namingEvent.getInstances(); - log.info("监听到服务变更，更新本地缓存: {} -> 实例数 {}", serviceName, instances.size()); - if (instances != null && !instances.isEmpty()) { - serviceCache.put(serviceName, instances); + private void ensureSubscribed(String serviceName) throws NacosException { + if (subscriptions.containsKey(serviceName)) { + return; + } + + synchronized (subscriptions) { + if (subscriptions.containsKey(serviceName)) { + return; + } + + EventListener listener = new EventListener() { + @Override + public void onEvent(Event event) { + if (event instanceof NamingEvent) { + List instances = ((NamingEvent) event).getInstances(); + updateCache(serviceName, instances); + log.info("监听到服务变更，更新本地缓存: {} -> 实例数 {}", + serviceName, instances == null ? 0 : instances.size()); } } + }; + namingService.subscribe(serviceName, listener); + subscriptions.put(serviceName, listener); + } + } + + void updateCache(String serviceName, List instances) { + if (instances == null || instances.isEmpty()) { + serviceCache.remove(serviceName); + } else { + serviceCache.put(serviceName, List.copyOf(instances)); + } + } + + private InetSocketAddress selectAddress(List instances) { + List addressList = instances.stream() + .map(instance -> instance.getIp() + ":" + instance.getPort()) + .toList(); + String targetAddress = loadBalancer.select(addressList); + String[] array = targetAddress.split(":"); + return new InetSocketAddress(array[0], Integer.parseInt(array[1])); + } + + @Override + public void close() { + if (!closed.compareAndSet(false, true)) { + return; + } + + subscriptions.forEach((serviceName, listener) -> { + try { + namingService.unsubscribe(serviceName, listener); + } catch (NacosException e) { + log.warn("取消 Nacos 服务订阅失败: {}", serviceName, e); } }); + subscriptions.clear(); + serviceCache.clear(); + + try { + namingService.shutDown(); + } catch (NacosException e) { + log.warn("关闭 Nacos NamingService 失败", e); + } } } diff --git a/rpc-core/src/main/java/com/xiaoyu/rpc/core/server/RpcServer.java b/rpc-core/src/main/java/com/xiaoyu/rpc/core/server/RpcServer.java index 8329af8..bbf6e0d 100644 --- a/rpc-core/src/main/java/com/xiaoyu/rpc/core/server/RpcServer.java +++ b/rpc-core/src/main/java/com/xiaoyu/rpc/core/server/RpcServer.java @@ -8,6 +8,8 @@ import lombok.extern.slf4j.Slf4j; import java.net.InetSocketAddress; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; @Slf4j @@ -17,6 +19,8 @@ public class RpcServer implements AutoCloseable { private final int serverPort; private final ServiceRegistry serviceRegistry; private final TransportServer transportServer; + private final Set registeredServices = ConcurrentHashMap.newKeySet(); + private final AtomicBoolean started = new AtomicBoolean(false); private final AtomicBoolean stopped = new AtomicBoolean(false); private final Thread shutdownHook; @@ -67,32 +71,49 @@ public void register(Class interfaceClass, T serviceImpl) { String serviceName = interfaceClass.getName(); ServiceRepository.registerService(serviceName, serviceImpl); + registeredServices.add(serviceName); - try { - serviceRegistry.registerService(serviceName, new InetSocketAddress(serverHost, serverPort)); - log.info("Service registered: {}", serviceName); - } catch (Exception e) { - log.error("Failed to register service: {}", serviceName, e); + if (started.get()) { + publishService(serviceName); } } + /** + * 启动传输层，确认端口已 bind 后再将服务发布到注册中心。 + */ public void start() throws InterruptedException { if (stopped.get()) { throw new IllegalStateException("RpcServer 已关闭"); } + if (!started.compareAndSet(false, true)) { + throw new IllegalStateException("RpcServer 已启动"); + } try { transportServer.start(); - } finally { + for (String serviceName : registeredServices) { + publishService(serviceName); + } + log.info("RPC Server 启动完成，已发布 {} 个服务", registeredServices.size()); + } catch (InterruptedException | RuntimeException e) { stop(); + throw e; } } + /** + * 阻塞等待底层 TransportServer 停止。 + */ + public void awaitTermination() throws InterruptedException { + transportServer.awaitTermination(); + } + public void stop() { if (!stopped.compareAndSet(false, true)) { return; } + // 先从注册中心摘除，阻止新流量，再停止网络层。 try { serviceRegistry.clearRegistry(); } catch (Exception e) { @@ -113,6 +134,11 @@ public void close() { stop(); } + private void publishService(String serviceName) { + serviceRegistry.registerService(serviceName, new InetSocketAddress(serverHost, serverPort)); + log.info("Service published after transport ready: {} -> {}:{}", serviceName, serverHost, serverPort); + } + private void removeShutdownHook() { if (shutdownHook == null || Thread.currentThread() == shutdownHook) { return; diff --git a/rpc-core/src/main/java/com/xiaoyu/rpc/core/transport/TransportServer.java b/rpc-core/src/main/java/com/xiaoyu/rpc/core/transport/TransportServer.java index 80b04f2..0d0c1d4 100644 --- a/rpc-core/src/main/java/com/xiaoyu/rpc/core/transport/TransportServer.java +++ b/rpc-core/src/main/java/com/xiaoyu/rpc/core/transport/TransportServer.java @@ -1,19 +1,28 @@ package com.xiaoyu.rpc.core.transport; /** - * 传输层服务端接口 + * 传输层服务端接口。 */ public interface TransportServer { /** - * 启动服务 - * - * @throws InterruptedException 如果启动过程被中断 + * 启动服务并在监听端口成功 bind 后返回。 + * + * @throws InterruptedException 启动过程被中断 */ void start() throws InterruptedException; /** - * 停止服务 + * 阻塞等待服务端停止。默认实现用于不需要阻塞语义的传输层。 + * + * @throws InterruptedException 等待过程被中断 + */ + default void awaitTermination() throws InterruptedException { + // no-op by default + } + + /** + * 停止服务并释放资源。 */ void stop(); } diff --git a/rpc-core/src/test/java/com/xiaoyu/rpc/core/registry/nacos/NacosServiceDiscoveryTest.java b/rpc-core/src/test/java/com/xiaoyu/rpc/core/registry/nacos/NacosServiceDiscoveryTest.java index 7913d92..da4c70b 100644 --- a/rpc-core/src/test/java/com/xiaoyu/rpc/core/registry/nacos/NacosServiceDiscoveryTest.java +++ b/rpc-core/src/test/java/com/xiaoyu/rpc/core/registry/nacos/NacosServiceDiscoveryTest.java @@ -5,36 +5,21 @@ import com.alibaba.nacos.api.naming.listener.EventListener; import com.alibaba.nacos.api.naming.pojo.Instance; import com.xiaoyu.rpc.core.loadbalancer.LoadBalancer; -import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; -import java.lang.reflect.Field; import java.lang.reflect.Proxy; import java.net.InetSocketAddress; -import java.util.ArrayList; import java.util.List; -import java.util.Map; -import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; -import static org.junit.jupiter.api.Assertions.*; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; -@DisplayName("NacosServiceDiscovery 缓存与回退测试") +@DisplayName("NacosServiceDiscovery 缓存、订阅与关闭测试") public class NacosServiceDiscoveryTest { - @BeforeEach - @SuppressWarnings("unchecked") - void setUp() throws Exception { - Field cacheField = NacosServiceDiscovery.class.getDeclaredField("serviceCache"); - cacheField.setAccessible(true); - ((Map>) cacheField.get(null)).clear(); - - Field subscribedField = NacosServiceDiscovery.class.getDeclaredField("subscribedServices"); - subscribedField.setAccessible(true); - ((Set) subscribedField.get(null)).clear(); - } - @Test @DisplayName("正常发现服务并且同服务只订阅一次") void testLookupAndSubscribeOnce() { @@ -52,70 +37,124 @@ void testLookupAndSubscribeOnce() { return null; }); - LoadBalancer loadBalancer = addresses -> addresses.get(0); - NacosServiceDiscovery discovery = new NacosServiceDiscovery(namingService, loadBalancer); - + NacosServiceDiscovery discovery = new NacosServiceDiscovery(namingService, firstAddress()); InetSocketAddress first = discovery.lookupService("svc-a"); InetSocketAddress second = discovery.lookupService("svc-a"); assertEquals("10.0.0.1", first.getHostString()); assertEquals(8080, first.getPort()); assertEquals("10.0.0.1", second.getHostString()); - assertEquals(1, subscribeCount.get(), "Same service should only subscribe once"); + assertEquals(1, subscribeCount.get()); } @Test - @DisplayName("Nacos 异常时回退到本地缓存") - @SuppressWarnings("unchecked") - void testFallbackToCacheOnNacosError() throws Exception { - Field cacheField = NacosServiceDiscovery.class.getDeclaredField("serviceCache"); - cacheField.setAccessible(true); - Map> cache = (Map>) cacheField.get(null); - cache.put("svc-b", new ArrayList<>(List.of(instance("127.0.0.1", 9000)))); - + @DisplayName("Nacos 网络异常时回退到最近一次成功缓存") + void testFallbackToCacheOnNacosError() { + AtomicInteger lookupCount = new AtomicInteger(); NamingService namingService = namingServiceProxy((method, args) -> { if ("getAllInstances".equals(method) && args.length == 1) { + if (lookupCount.getAndIncrement() == 0) { + return List.of(instance("127.0.0.1", 9000)); + } throw new NacosException(500, "network down"); } - if ("subscribe".equals(method)) { - return null; + return null; + }); + + NacosServiceDiscovery discovery = new NacosServiceDiscovery(namingService, firstAddress()); + discovery.lookupService("svc-b"); + InetSocketAddress cached = discovery.lookupService("svc-b"); + + assertEquals("127.0.0.1", cached.getHostString()); + assertEquals(9000, cached.getPort()); + } + + @Test + @DisplayName("Nacos 明确返回空实例时清理旧缓存，不再路由到下线节点") + void testEmptyAuthoritativeResultInvalidatesCache() { + AtomicInteger lookupCount = new AtomicInteger(); + NamingService namingService = namingServiceProxy((method, args) -> { + if ("getAllInstances".equals(method) && args.length == 1) { + int index = lookupCount.getAndIncrement(); + if (index == 0) { + return List.of(instance("10.0.0.9", 8080)); + } + if (index == 1) { + return List.of(); + } + throw new NacosException(500, "network down after empty result"); } return null; }); - LoadBalancer loadBalancer = addresses -> addresses.get(0); - NacosServiceDiscovery discovery = new NacosServiceDiscovery(namingService, loadBalancer); + NacosServiceDiscovery discovery = new NacosServiceDiscovery(namingService, firstAddress()); + discovery.lookupService("svc-stale"); - InetSocketAddress address = discovery.lookupService("svc-b"); - assertEquals("127.0.0.1", address.getHostString()); - assertEquals(9000, address.getPort()); + RuntimeException notFound = assertThrows(RuntimeException.class, + () -> discovery.lookupService("svc-stale")); + assertTrue(notFound.getMessage().contains("未找到服务")); + + RuntimeException noFallback = assertThrows(RuntimeException.class, + () -> discovery.lookupService("svc-stale")); + assertTrue(noFallback.getMessage().contains("无缓存可用")); + } + + @Test + @DisplayName("订阅事件为空时删除缓存") + void testUpdateCacheRemovesEmptyInstances() { + NamingService namingService = namingServiceProxy((method, args) -> null); + NacosServiceDiscovery discovery = new NacosServiceDiscovery(namingService, firstAddress()); + + discovery.updateCache("svc-c", List.of(instance("10.0.0.1", 8080))); + discovery.updateCache("svc-c", List.of()); + + NamingService failingService = namingServiceProxy((method, args) -> { + if ("getAllInstances".equals(method)) { + throw new NacosException(500, "network down"); + } + return null; + }); + NacosServiceDiscovery emptyDiscovery = new NacosServiceDiscovery(failingService, firstAddress()); + assertThrows(RuntimeException.class, () -> emptyDiscovery.lookupService("svc-c")); } @Test - @DisplayName("Nacos 空列表且无缓存时抛异常") - void testNoInstanceAndNoCache() { + @DisplayName("close 幂等取消订阅并关闭 NamingService") + void testCloseUnsubscribesAndShutsDown() { + AtomicInteger unsubscribeCount = new AtomicInteger(); + AtomicInteger shutdownCount = new AtomicInteger(); NamingService namingService = namingServiceProxy((method, args) -> { if ("getAllInstances".equals(method) && args.length == 1) { - return List.of(); + return List.of(instance("127.0.0.1", 8080)); } - if ("subscribe".equals(method)) { - return null; + if ("unsubscribe".equals(method)) { + unsubscribeCount.incrementAndGet(); + } + if ("shutDown".equals(method)) { + shutdownCount.incrementAndGet(); } return null; }); - LoadBalancer loadBalancer = addresses -> addresses.get(0); - NacosServiceDiscovery discovery = new NacosServiceDiscovery(namingService, loadBalancer); + NacosServiceDiscovery discovery = new NacosServiceDiscovery(namingService, firstAddress()); + discovery.lookupService("svc-close"); + discovery.close(); + discovery.close(); + + assertEquals(1, unsubscribeCount.get()); + assertEquals(1, shutdownCount.get()); + assertThrows(IllegalStateException.class, () -> discovery.lookupService("svc-close")); + } - RuntimeException ex = assertThrows(RuntimeException.class, () -> discovery.lookupService("svc-empty")); - assertTrue(ex.getMessage().contains("未找到服务"), "Should throw not found error"); + private static LoadBalancer firstAddress() { + return addresses -> addresses.get(0); } private static Instance instance(String ip, int port) { - Instance i = new Instance(); - i.setIp(ip); - i.setPort(port); - return i; + Instance instance = new Instance(); + instance.setIp(ip); + instance.setPort(port); + return instance; } private static NamingService namingServiceProxy(Invocation invocation) { @@ -133,30 +172,14 @@ private static NamingService namingServiceProxy(Invocation invocation) { } Object result = invocation.invoke(method.getName(), args == null ? new Object[0] : args); if (result == null && method.getReturnType().isPrimitive()) { - if (method.getReturnType() == boolean.class) { - return false; - } - if (method.getReturnType() == byte.class) { - return (byte) 0; - } - if (method.getReturnType() == short.class) { - return (short) 0; - } - if (method.getReturnType() == int.class) { - return 0; - } - if (method.getReturnType() == long.class) { - return 0L; - } - if (method.getReturnType() == float.class) { - return 0F; - } - if (method.getReturnType() == double.class) { - return 0D; - } - if (method.getReturnType() == char.class) { - return '\0'; - } + if (method.getReturnType() == boolean.class) return false; + if (method.getReturnType() == byte.class) return (byte) 0; + if (method.getReturnType() == short.class) return (short) 0; + if (method.getReturnType() == int.class) return 0; + if (method.getReturnType() == long.class) return 0L; + if (method.getReturnType() == float.class) return 0F; + if (method.getReturnType() == double.class) return 0D; + if (method.getReturnType() == char.class) return '\0'; } return result; }); diff --git a/rpc-core/src/test/java/com/xiaoyu/rpc/core/server/RpcServerLifecycleTest.java b/rpc-core/src/test/java/com/xiaoyu/rpc/core/server/RpcServerLifecycleTest.java new file mode 100644 index 0000000..e6f8126 --- /dev/null +++ b/rpc-core/src/test/java/com/xiaoyu/rpc/core/server/RpcServerLifecycleTest.java @@ -0,0 +1,96 @@ +package com.xiaoyu.rpc.core.server; + +import com.xiaoyu.rpc.core.registry.ServiceRegistry; +import com.xiaoyu.rpc.core.transport.TransportServer; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import java.net.InetSocketAddress; +import java.util.ArrayList; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +@DisplayName("RpcServer 启动发布顺序测试") +class RpcServerLifecycleTest { + + @Test + @DisplayName("服务必须在传输层 bind 成功后才发布到注册中心") + void testBindBeforeRegistryPublish() throws Exception { + List events = new ArrayList<>(); + RecordingRegistry registry = new RecordingRegistry(events); + RecordingTransportServer transportServer = new RecordingTransportServer(events); + RpcServer server = new RpcServer("127.0.0.1", 9090, registry, transportServer); + + server.register(DemoService.class, new DemoServiceImpl()); + assertTrue(events.isEmpty(), "register() 只登记本地服务，启动前不应向注册中心发布"); + + server.start(); + + assertEquals(List.of("bind", "publish:" + DemoService.class.getName()), events); + server.close(); + } + + @Test + @DisplayName("服务端启动后新增服务应立即发布") + void testRegisterAfterStartedPublishesImmediately() throws Exception { + List events = new ArrayList<>(); + RecordingRegistry registry = new RecordingRegistry(events); + RecordingTransportServer transportServer = new RecordingTransportServer(events); + RpcServer server = new RpcServer("127.0.0.1", 9090, registry, transportServer); + + server.start(); + server.register(DemoService.class, new DemoServiceImpl()); + + assertEquals(List.of("bind", "publish:" + DemoService.class.getName()), events); + server.close(); + } + + interface DemoService { + String hello(); + } + + static class DemoServiceImpl implements DemoService { + @Override + public String hello() { + return "ok"; + } + } + + private static final class RecordingRegistry implements ServiceRegistry { + private final List events; + + private RecordingRegistry(List events) { + this.events = events; + } + + @Override + public void registerService(String serviceName, InetSocketAddress inetSocketAddress) { + events.add("publish:" + serviceName); + } + + @Override + public void clearRegistry() { + // 生命周期顺序断言只关注 bind/publish。 + } + } + + private static final class RecordingTransportServer implements TransportServer { + private final List events; + + private RecordingTransportServer(List events) { + this.events = events; + } + + @Override + public void start() { + events.add("bind"); + } + + @Override + public void stop() { + // no-op + } + } +} diff --git a/rpc-provider/src/main/java/com/xiaoyu/rpc/provider/ProviderApp.java b/rpc-provider/src/main/java/com/xiaoyu/rpc/provider/ProviderApp.java index 7ffd743..7d123eb 100644 --- a/rpc-provider/src/main/java/com/xiaoyu/rpc/provider/ProviderApp.java +++ b/rpc-provider/src/main/java/com/xiaoyu/rpc/provider/ProviderApp.java @@ -9,13 +9,10 @@ public class ProviderApp { private static final Logger log = LoggerFactory.getLogger(ProviderApp.class); public static void main(String[] args) { - try { - // Use configuration from rpc-config.yaml (default: nacos) - // System.setProperty("rpc.registry", "local"); - - RpcServer server = new RpcServer(); + try (RpcServer server = new RpcServer()) { server.register(HelloService.class, new HelloServiceImpl()); server.start(); + server.awaitTermination(); } catch (Exception e) { log.error("Failed to start RPC provider", e); System.exit(1); diff --git a/rpc-spring-boot-starter/pom.xml b/rpc-spring-boot-starter/pom.xml index 0ba800a..e78b62b 100644 --- a/rpc-spring-boot-starter/pom.xml +++ b/rpc-spring-boot-starter/pom.xml @@ -15,21 +15,17 @@ - com.xiaoyu.rpc rpc-core ${project.version} - - com.xiaoyu.rpc rpc-transport-netty ${project.version} - org.springframework.boot spring-boot-starter @@ -46,8 +42,13 @@ ${spring-boot.version} true + + org.springframework.boot + spring-boot-starter-test + ${spring-boot.version} + test + - org.projectlombok lombok diff --git a/rpc-spring-boot-starter/src/main/java/com/xiaoyu/rpc/spring/RpcAutoConfiguration.java b/rpc-spring-boot-starter/src/main/java/com/xiaoyu/rpc/spring/RpcAutoConfiguration.java index d3dd01f..745e9b8 100644 --- a/rpc-spring-boot-starter/src/main/java/com/xiaoyu/rpc/spring/RpcAutoConfiguration.java +++ b/rpc-spring-boot-starter/src/main/java/com/xiaoyu/rpc/spring/RpcAutoConfiguration.java @@ -8,20 +8,20 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.SmartLifecycle; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import java.util.concurrent.atomic.AtomicBoolean; + /** - * RPC 自动配置类 + * RPC 自动配置类。 */ @Slf4j @Configuration @EnableConfigurationProperties(RpcProperties.class) public class RpcAutoConfiguration { - /** - * 将 Spring Boot 配置同步到 RpcConfig (核心框架配置) - */ @Bean public RpcConfig rpcConfig(RpcProperties properties) { // 核心模块不依赖 Spring，通过 System Properties 作为两层之间的配置桥接。 @@ -45,13 +45,9 @@ public RpcConfig rpcConfig(RpcProperties properties) { log.info("RPC 配置已从 Spring Boot 同步: registry={}, server={}:{}, protocol={}, requestTimeoutMs={}", properties.getRegistry(), properties.getServerHost(), properties.getServerPort(), properties.getProtocol(), properties.getRequestTimeoutMs()); - return RpcConfig.getInstance(); } - /** - * 创建 RpcServer Bean (仅当 serverEnabled=true 时) - */ @Bean @ConditionalOnProperty(prefix = "rpc", name = "server-enabled", havingValue = "true", matchIfMissing = true) @ConditionalOnMissingBean @@ -60,47 +56,69 @@ public RpcServer rpcServer(RpcConfig rpcConfig) { return new RpcServer(); } - /** - * 创建 RPC Bean 后处理器 - */ @Bean public RpcPostProcessor rpcPostProcessor() { return new RpcPostProcessor(); } - /** - * 启动 RpcServer - */ @Bean @ConditionalOnProperty(prefix = "rpc", name = "server-enabled", havingValue = "true", matchIfMissing = true) - public RpcServerRunner rpcServerRunner(RpcServer rpcServer) { - return new RpcServerRunner(rpcServer); + public RpcServerLifecycle rpcServerLifecycle(RpcServer rpcServer) { + return new RpcServerLifecycle(rpcServer); } /** - * 使用 CommandLineRunner 启动 RpcServer + * 将 RPC Server 纳入 Spring 生命周期：Context 启动时同步启动，关闭时优雅释放服务端资源。 */ - @Slf4j - public static class RpcServerRunner implements org.springframework.boot.CommandLineRunner { + public static class RpcServerLifecycle implements SmartLifecycle { private final RpcServer rpcServer; + private final AtomicBoolean running = new AtomicBoolean(false); - public RpcServerRunner(RpcServer rpcServer) { + public RpcServerLifecycle(RpcServer rpcServer) { this.rpcServer = rpcServer; } @Override - public void run(String... args) { - log.info("启动 RPC Server..."); - Thread serverThread = new Thread(() -> { - try { - rpcServer.start(); - } catch (InterruptedException e) { - log.error("RPC Server 启动失败", e); - Thread.currentThread().interrupt(); - } - }, "rpc-server-thread"); - serverThread.setDaemon(true); - serverThread.start(); + public void start() { + if (!running.compareAndSet(false, true)) { + return; + } + try { + rpcServer.start(); + log.info("RPC Server 已由 Spring SmartLifecycle 启动"); + } catch (InterruptedException e) { + running.set(false); + Thread.currentThread().interrupt(); + throw new IllegalStateException("RPC Server 启动被中断", e); + } catch (RuntimeException e) { + running.set(false); + throw e; + } + } + + @Override + public void stop() { + if (!running.getAndSet(false)) { + return; + } + rpcServer.close(); + log.info("RPC Server 已由 Spring SmartLifecycle 停止"); + } + + @Override + public boolean isRunning() { + return running.get(); + } + + @Override + public boolean isAutoStartup() { + return true; + } + + @Override + public int getPhase() { + // 启动尽量靠后，关闭尽量靠前，先摘除 RPC 流量再销毁其他业务 Bean。 + return Integer.MAX_VALUE; } } } diff --git a/rpc-spring-boot-starter/src/test/java/com/xiaoyu/rpc/spring/RpcServerLifecycleTest.java b/rpc-spring-boot-starter/src/test/java/com/xiaoyu/rpc/spring/RpcServerLifecycleTest.java new file mode 100644 index 0000000..ddc79f3 --- /dev/null +++ b/rpc-spring-boot-starter/src/test/java/com/xiaoyu/rpc/spring/RpcServerLifecycleTest.java @@ -0,0 +1,50 @@ +package com.xiaoyu.rpc.spring; + +import com.xiaoyu.rpc.core.server.RpcServer; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +@DisplayName("Spring RPC 生命周期测试") +class RpcServerLifecycleTest { + + @Test + @DisplayName("SmartLifecycle 启动和停止委托给 RpcServer") + void testStartAndStop() throws Exception { + RpcServer rpcServer = mock(RpcServer.class); + RpcAutoConfiguration.RpcServerLifecycle lifecycle = + new RpcAutoConfiguration.RpcServerLifecycle(rpcServer); + + lifecycle.start(); + assertTrue(lifecycle.isRunning()); + verify(rpcServer, times(1)).start(); + + lifecycle.stop(); + assertFalse(lifecycle.isRunning()); + verify(rpcServer, times(1)).close(); + + lifecycle.stop(); + verify(rpcServer, times(1)).close(); + } + + @Test + @DisplayName("RpcServer 启动失败时 Spring 生命周期保持未运行状态并向上抛错") + void testStartupFailurePropagates() throws Exception { + RpcServer rpcServer = mock(RpcServer.class); + doThrow(new IllegalStateException("port occupied")).when(rpcServer).start(); + RpcAutoConfiguration.RpcServerLifecycle lifecycle = + new RpcAutoConfiguration.RpcServerLifecycle(rpcServer); + + IllegalStateException error = assertThrows(IllegalStateException.class, lifecycle::start); + + assertTrue(error.getMessage().contains("port occupied")); + assertFalse(lifecycle.isRunning()); + } +} diff --git a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/transport/netty/NettyTransportServer.java b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/transport/netty/NettyTransportServer.java index 05bf47c..11d0eb0 100644 --- a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/transport/netty/NettyTransportServer.java +++ b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/transport/netty/NettyTransportServer.java @@ -90,12 +90,20 @@ protected void initChannel(SocketChannel ch) { }); serverChannel = bootstrap.bind(port).sync().channel(); - log.info("RPC Server (Netty) started on port {}, bossThreads={}, workerThreads={}, businessThreads={}, " + log.info("RPC Server (Netty) ready on port {}, bossThreads={}, workerThreads={}, businessThreads={}, " + "businessQueueCapacity={}", port, bossThreads, workerThreads, businessThreads, businessQueueCapacity); - serverChannel.closeFuture().sync(); - } finally { + } catch (InterruptedException | RuntimeException e) { stop(); + throw e; + } + } + + @Override + public void awaitTermination() throws InterruptedException { + Channel channel = serverChannel; + if (channel != null) { + channel.closeFuture().sync(); } }