宜春市网站建设_网站建设公司_安全防护_seo优化

文 / Kenyon，资深软件架构师，15年软件开发和技术管理经验，从程序员做到企业技术高管，专注技术管理、架构设计、AI技术应用和落地。

由于公众号推流的原因，请在关注页右上角加星标，这样才能及时收到新文章的推送。

摘要：本文完成了RPC框架的剩余核心功能，包括基于Nacos的服务注册中心、多种负载均衡策略（随机、轮询、最小连接数）及服务端核心实现，提供了完整的使用示例（服务定义、服务端/客户端实现）和单元测试，最终构建了一个功能完备的RPC框架。

引言

在前面的两篇文章中，我们完成了RPC框架整体的架构设计，并实现了一些核心的组件。今天，我们将完成RPC框架剩余的功能，这些功能包括服务注册中心、服务端核心实现、负载均衡等，在文章最后我们将提供完整的使用示例和测试。

一、服务注册中心（Registry Center）实现

为了支持多种不同的注册中心，我们基于单一职责原则、里氏替换原则和接口隔离原则，来对服务注册中心的接口和实现进行设计。

// 服务注册中心接口
public interface RegistryCenter {// 注册服务void register(String serviceName, InetSocketAddress address) throws Exception;// 注销服务void unregister(String serviceName, InetSocketAddress address) throws Exception;// 发现服务List<InetSocketAddress> discover(String serviceName) throws Exception;// 订阅服务变化void subscribe(String serviceName, ServiceChangeListener listener) throws Exception;// 取消订阅void unsubscribe(String serviceName, ServiceChangeListener listener) throws Exception;// 关闭连接void close();
}// 服务变化监听器接口
public interface ServiceChangeListener {void onServiceChange(String serviceName, List<InetSocketAddress> addresses);
}// 使用Nacos实现的服务注册中心
public class NacosRegistryCenter implements RegistryCenter {private static final Logger logger = LoggerFactory.getLogger(NacosRegistryCenter.class);private final NamingService namingService;private final Map<String, List<ServiceChangeListener>> listeners = new ConcurrentHashMap<>();private final String groupName;//构造函数public NacosRegistryCenter(String serverAddr) throws NacosException {this(serverAddr, "DEFAULT_GROUP");}//构造函数public NacosRegistryCenter(String serverAddr, String groupName) throws NacosException {this.groupName = groupName;Properties properties = new Properties();properties.setProperty("serverAddr", serverAddr);this.namingService = NacosFactory.createNamingService(properties);}@Overridepublic void register(String serviceName, InetSocketAddress address) throws Exception {logger.debug("Registering service: {} at {} in group {}", serviceName, address, groupName);Instance instance = new Instance();instance.setIp(address.getAddress().getHostAddress());instance.setPort(address.getPort());// 设置权重instance.setWeight(1.0);// 设置为临时实例，服务下线后自动删除instance.setEphemeral(true);// 设置为健康状态instance.setHealthy(true);namingService.registerInstance(serviceName, groupName, instance);logger.debug("Successfully registered service: {} at {} in group {}", serviceName, address, groupName);}@Overridepublic void unregister(String serviceName, InetSocketAddress address) throws Exception {// Nacos会在服务下线后自动移除临时实例，这里不需要手动注销logger.debug("Unregistering service: {} at {} (will be automatically removed)", serviceName, address);}@Overridepublic List<InetSocketAddress> discover(String serviceName) throws Exception {logger.debug("Discovering service: {} in group {}", serviceName, groupName);// 获取所有健康的服务实例List<Instance> instances = namingService.getAllInstances(serviceName, groupName);logger.debug("Found {} instances for service: {}", instances.size(), serviceName);for (Instance instance : instances) {logger.debug("Instance: {}:{}", instance.getIp(), instance.getPort());}// 添加监听器以监听服务变化namingService.subscribe(serviceName, groupName, new NacosEventListener(serviceName));List<InetSocketAddress> addresses = new ArrayList<>();for (Instance instance : instances) {if (instance.isHealthy()) {addresses.add(new InetSocketAddress(instance.getIp(), instance.getPort()));}}logger.debug("Returning {} healthy addresses for service: {}", addresses.size(), serviceName);return addresses;}@Overridepublic void subscribe(String serviceName, ServiceChangeListener listener) throws Exception {listeners.computeIfAbsent(serviceName, k -> new ArrayList<>()).add(listener);// 立即触发一次回调，获取当前服务列表listener.onServiceChange(serviceName, discover(serviceName));}@Overridepublic void unsubscribe(String serviceName, ServiceChangeListener listener) throws Exception {List<ServiceChangeListener> serviceListeners = listeners.get(serviceName);if (serviceListeners != null) {serviceListeners.remove(listener);}}@Overridepublic void close() {try {if (namingService != null) {namingService.shutDown();}} catch (Exception e) {logger.error("Error closing Nacos naming service", e);}}//Nacos事件监听器private class NacosEventListener implements EventListener {private final String serviceName;public NacosEventListener(String serviceName) {this.serviceName = serviceName;}@Overridepublic void onEvent(Event event) {if (event instanceof NamingEvent) {NamingEvent namingEvent = (NamingEvent) event;List<Instance> instances = namingEvent.getInstances();List<InetSocketAddress> addresses = instances.stream().filter(Instance::isHealthy).map(instance -> new InetSocketAddress(instance.getIp(), instance.getPort())).collect(Collectors.toList());// 触发监听器List<ServiceChangeListener> serviceListeners = listeners.get(serviceName);if (serviceListeners != null) {for (ServiceChangeListener listener : serviceListeners) {try {listener.onServiceChange(serviceName, addresses);} catch (Exception e) {logger.error("Error notifying service change listener", e);}}}}}}
}

二、负载均衡策略（Load Balancer）实现

同样，为了支持多种不同的负载均衡策略，我们也基于里氏替换原则和设计模式里面的策略模式来定义组件的接口和实现类。

// 定义负载均衡接口
public interface LoadBalancer {InetSocketAddress select(List<InetSocketAddress> addresses);
}// 随机负载均衡
public class RandomLoadBalancer implements LoadBalancer {private final Random random = new Random();@Overridepublic InetSocketAddress select(List<InetSocketAddress> addresses) {if (addresses.isEmpty()) {return null;}return addresses.get(random.nextInt(addresses.size()));}
}// 轮询负载均衡
public class RoundRobinLoadBalancer implements LoadBalancer {private final AtomicInteger index = new AtomicInteger(0);@Overridepublic InetSocketAddress select(List<InetSocketAddress> addresses) {if (addresses.isEmpty()) {return null;}return addresses.get(Math.abs(index.getAndIncrement() % addresses.size()));}
}// 最小连接数负载均衡（简化版）
public class LeastConnectionLoadBalancer implements LoadBalancer {// 模拟连接数统计private final Map<InetSocketAddress, AtomicInteger> connectionCount = new ConcurrentHashMap<>();@Overridepublic InetSocketAddress select(List<InetSocketAddress> addresses) {if (addresses.isEmpty()) {return null;}return addresses.stream().min(Comparator.comparingInt(address -> connectionCount.getOrDefault(address, new AtomicInteger(0)).get())).orElse(null);}// 记录连接数变化public void incrementConnection(InetSocketAddress address) {connectionCount.computeIfAbsent(address, k -> new AtomicInteger(0)).incrementAndGet();}public void decrementConnection(InetSocketAddress address) {AtomicInteger count = connectionCount.get(address);if (count != null) {count.decrementAndGet();}}
}
//ServiceProxy.java服务代理类构造函数在使用负载均衡时的使用方式
public ServiceProxy(Class<?> serviceClass, RegistryCenter registryCenter) {// 默认使用随机负载均衡，也可以在构造函数中传入其他负载均衡策略this(serviceClass, registryCenter, new RandomLoadBalance());
}

三、服务端核心实现

// RPC服务端核心类
public class RpcServer {private static final Logger logger = LoggerFactory.getLogger(RpcServer.class);private final TransportServer transportServer;private final RegistryCenter registryCenter;private final ServiceRegistry serviceRegistry;private final int port;private final AtomicBoolean started = new AtomicBoolean(false);private final Set<String> registeredServices = new HashSet<>();//构造函数public RpcServer(int port, RegistryCenter registryCenter) {this.port = port;this.registryCenter = registryCenter;this.serviceRegistry = new ServiceRegistry();this.transportServer = new NettyTransportServer();logger.debug("RpcServer created with registryCenter: {}", registryCenter);}//注册服务public void registerService(Class<?> serviceClass, Object serviceImpl) {if (!serviceClass.isAssignableFrom(serviceImpl.getClass())) {throw new IllegalArgumentException("Service implementation must implement the service interface");}String serviceName = serviceClass.getName();serviceRegistry.registerService(serviceName, serviceImpl);registeredServices.add(serviceName);logger.debug("Registered service: {} with implementation: {}", serviceName, serviceImpl.getClass().getName());}//启动服务public void start() throws Exception {logger.debug("Starting RpcServer...");if (!started.compareAndSet(false, true)) {logger.debug("Server already started, returning.");return;}try {// 启动网络传输服务logger.debug("Starting transport server on port: {}", port);transportServer.start(port, new RpcRequestHandler(serviceRegistry));logger.debug("RPC server started on port: {}", port);} catch (Exception e) {logger.error("Failed to start transport server", e);throw e;}// 注册服务到服务中心logger.debug("Registry center: {}", registryCenter);logger.debug("Number of registered services: {}", registeredServices.size());if (registryCenter != null) {InetSocketAddress address = new InetSocketAddress("localhost", port);logger.debug("Attempting to register {} services", registeredServices.size());for (String serviceName : registeredServices) {logger.debug("Registering service: {} at {}", serviceName, address);registryCenter.register(serviceName, address);}} else {logger.debug("Registry center is null, skipping service registration");}}//停止服务public void stop() throws Exception {if (!started.compareAndSet(true, false)) {return;}// 从服务中心注销服务if (registryCenter != null) {InetSocketAddress address = new InetSocketAddress("localhost", port);for (String serviceName : registeredServices) {registryCenter.unregister(serviceName, address);}registryCenter.close();}// 停止网络传输服务transportServer.stop();logger.debug("RPC server stopped");}//获取服务端口public int getPort() {return port;}//服务是否已启动public boolean isStarted() {return started.get();}
}

四、RPC框架的使用示例

这里的示例我们使用计算服务来展示RPC框架的使用。

1. 定义服务接口

我们先定义一个简单的计算器服务接口，包含加法、减法、乘法和除法运算。

// 计算器服务接口
public interface CalculatorService {//加法运算int add(int a, int b);//减法运算int subtract(int a, int b);//乘法运算int multiply(int a, int b);//除法运算int divide(int a, int b) throws IllegalArgumentException;
}

2. 服务端实现

以下是计算器服务接口的具体实现，为了方便查看被调用的情况，我们添加了一些日志记录和异常处理。

// 计算器服务实现
@RpcService(CalculatorService.class)
public class CalculatorServiceImpl implements CalculatorService {private static final Logger logger = LoggerFactory.getLogger(CalculatorServiceImpl.class);@Overridepublic int add(int a, int b) {int result = a + b;logger.info("Calculated: {} + {} = {}", a, b, result);return result;}@Overridepublic int subtract(int a, int b) {int result = a - b;logger.info("Calculated: {} - {} = {}", a, b, result);return result;}@Overridepublic int multiply(int a, int b) {int result = a * b;logger.info("Calculated: {} * {} = {}", a, b, result);return result;}@Overridepublic int divide(int a, int b) throws IllegalArgumentException {if (b == 0) {throw new IllegalArgumentException("除数不能为0");}int result = a / b;logger.info("Calculated: {} / {} = {}", a, b, result);return result;}
}// 服务端示例的启动类
public class ServerExample {public static void main(String[] args) {// 默认使用Nacos作为注册中心RegistryCenter registryCenter = new NacosRegistryCenter("localhost:8848");// 如果想使用ZooKeeper作为注册中心的话就打开下面这行注释即可// RegistryCenter registryCenter = new ZookeeperRegistryCenter("localhost:2181");int port = 8081;RpcServer server = new RpcServer(port, registryCenter);CalculatorService calculatorService = new CalculatorServiceImpl();server.registerService(CalculatorService.class, calculatorService);server.start();}
}

3. 客户端实现

我们实现客户端代码，使用代理模式调用远程服务。

// 客户端启动类
public class ClientExample {public static void main(String[] args) {// 使用Nacos作为注册中心RegistryCenter registryCenter = new NacosRegistryCenter("localhost:8848");// 或者使用ZooKeeper作为注册中心// RegistryCenter registryCenter = new ZookeeperRegistryCenter("localhost:2181");// 使用轮询负载均衡策略RoundRobinLoadBalance loadBalance = new RoundRobinLoadBalance();CalculatorService calculatorService = (CalculatorService) Proxy.newProxyInstance(CalculatorService.class.getClassLoader(),new Class<?>[]{CalculatorService.class},new ServiceProxy(CalculatorService.class, registryCenter, loadBalance));// 调用远程服务int result = calculatorService.add(10, 5);System.out.println("10 + 5 = " + result);}
}

五、框架测试与优化

1. 单元测试示例

// 服务注册中心测试
public class ServerTest {private static final Logger logger = LoggerFactory.getLogger(ServerTest.class);private RegistryCenter registryCenter;private RpcServer server;private Thread serverThread;private int port = 8081; // 将端口作为实例变量@Beforepublic void setUp() throws Exception {// 检查系统属性中是否指定了端口String portProperty = System.getProperty("server.port");if (portProperty != null && !portProperty.isEmpty()) {try {port = Integer.parseInt(portProperty);logger.info("Using port from system property: {}", port);} catch (NumberFormatException e) {logger.warn("Invalid port specified in system property, using default port 8081");}}// 创建Nacos注册中心实例logger.info("Creating Nacos registry center...");registryCenter = new NacosRegistryCenter("localhost:8848");logger.info("Nacos registry center created: {}", registryCenter);// 创建RPC服务器logger.info("Creating RPC server with registry center...");server = new RpcServer(port, registryCenter);logger.info("RPC server created: {}", server);// 创建服务实现类实例CalculatorService calculatorService = new CalculatorServiceImpl();// 注册服务logger.info("Registering service...");server.registerService(CalculatorService.class, calculatorService);logger.info("Service registered.");// 在独立线程中启动服务器serverThread = new Thread(() -> {try {logger.info("Starting server on port {}...", port);server.start();logger.info("Server started.");} catch (Exception e) {logger.error("Failed to start RPC server", e);}});serverThread.start();// 等待服务器启动Thread.sleep(2000);logger.info("RPC server started successfully on port {}", port);logger.info("Service registered: {}", CalculatorService.class.getName());}@Testpublic void testServerRunning() throws InterruptedException {// 保持服务器运行一段时间用于测试logger.info("Server is running, keeping it alive for testing...");Thread.sleep(30000); // 保持运行30秒用于测试}@Afterpublic void tearDown() {try {// 关闭资源if (server != null) {server.stop();}if (registryCenter != null) {registryCenter.close();}// 等待服务器线程结束if (serverThread != null) {serverThread.join(5000); // 最多等待5秒}logger.info("Server stopped.");} catch (Exception e) {logger.error("Error stopping server", e);}}
}// 客户端测试
public class ClientTest {private static final Logger logger = LoggerFactory.getLogger(ClientTest.class);private RegistryCenter registryCenter;private CalculatorService calculatorService;@Beforepublic void setUp() throws Exception {// 创建注册中心logger.info("Creating Nacos registry center...");registryCenter = new NacosRegistryCenter("localhost:8848");logger.info("Nacos registry center created: {}", registryCenter);// 创建服务代理，使用轮询负载均衡策略logger.info("Creating service proxy with round robin load balance...");RoundRobinLoadBalance loadBalance = new RoundRobinLoadBalance();calculatorService = (CalculatorService) Proxy.newProxyInstance(CalculatorService.class.getClassLoader(),new Class<?>[]{CalculatorService.class},new ServiceProxy(CalculatorService.class, registryCenter, loadBalance));logger.info("RPC client initialized successfully");}@Testpublic void testCalculatorServiceAdd() throws Exception {logger.info("--- Testing Calculator Service Add ---");// 测试加法int result = calculatorService.add(10, 5);assertEquals(15, result);logger.info("10 + 5 = {}", result);}@Testpublic void testCalculatorServiceSubtract() throws Exception {logger.info("--- Testing Calculator Service Subtract ---");// 测试减法int result = calculatorService.subtract(10, 5);assertEquals(5, result);logger.info("10 - 5 = {}", result);}@Testpublic void testCalculatorServiceMultiply() throws Exception {logger.info("--- Testing Calculator Service Multiply ---");// 测试乘法int result = calculatorService.multiply(10, 5);assertEquals(50, result);logger.info("10 * 5 = {}", result);}@Testpublic void testCalculatorServiceDivide() throws Exception {logger.info("--- Testing Calculator Service Divide ---");// 测试除法int result = calculatorService.divide(10, 5);assertEquals(2, result);logger.info("10 / 5 = {}", result);}@Testpublic void testMultipleCallsWithLoadBalance() throws Exception {logger.info("--- Testing Calculator Service with Load Balance ---");// 多次调用以测试负载均衡for (int i = 0; i < 5; i++) {// 测试加法int result = calculatorService.add(10, 5);assertEquals(15, result);logger.info("Round {}: 10 + 5 = {}", i+1, result);// 稍微延时以便观察负载均衡效果Thread.sleep(500);}logger.info("All RPC calls completed successfully");}@Afterpublic void tearDown() {// 关闭资源if (registryCenter != null) {registryCenter.close();}}
}

2. 性能优化建议

为了提高RPC框架的性能，后续可以考虑按照以下的建议来进行优化：

1. 优化客户端的连接池管理方式，避免频繁创建和关闭连接
2. 增加异步调用模式的支持，提高框架的并发处理能力
3. 合并请求，让框架支持批量发送多个请求，从而减少网络的开销
4. 对请求和响应的数据进行压缩，减少网络的传输量
5. 缓存服务发现时返回的结果列表，减少客户端向注册中心发起的访问，同时也增加注册中心实例变化时的监听功能，当服务实例有变化时，能够及时更新本地的数据。
6. 合理配置线程池参数，提高系统吞吐量

六、系列总结

在这个"手搓RPC框架"系列的3篇文章里面，我们基于常见的架构设计原则、方法和模式，从0到1实现了一个功能完整的RPC框架。其中，我们重点应用了以下的这些架构设计的原则和方法：

1. 核心架构原则应用

• SOLID原则：每个组件只负责一个明确的功能，组件之间通过接口通信，实现高内聚低耦合。而且类、模块等软件实体基本都是面向接口编程，从而实现对扩展开放，对修改关闭，同时子类也都能平滑地替换父类而不破坏程序正确性。
• 高内聚低耦合：组件内部功能紧密相关，组件间通过接口通信，减少了组件之间的依赖关系，提高了系统的可维护性和可扩展性。
• KISS原则：实现简洁明了，避免过度设计，保持代码的可读性和可维护性。
• 依赖倒置原则：通过依赖注入实现解耦，减少了组件之间的直接依赖关系，提高了系统的灵活性和可测试性。
• 策略模式：负载均衡等功能支持多种策略切换，用户可以根据实际场景选择合适的策略，而无需修改框架代码。

2. 框架特点

• 通过面向接口编程的方式，使得框架能够支持自定义序列化、传输、负载均衡等组件，从而实现高度的可扩展性和可定制性。
• 简单的API接口，方便服务注册和调用。用户只需要定义服务接口，实现服务端的业务逻辑，即可完成服务的注册和调用。
• 服务自动发现、故障转移等机制，能够自动发现服务端实例，同时在实例故障时能够自动切换到其他可用的实例，提高系统的可用性和容错性。
• 支持连接池、异步调用等优化手段，提高系统的并发处理能力和吞吐量。

3. 未来扩展方向

• 支持更多序列化协议（如Protobuf、Kryo等）
• 实现服务治理功能（限流、降级、熔断等）
• 增加监控和追踪能力
• 支持分布式事务
• 实现集群部署和动态扩缩容

通过这个系列文章，我们即学习了RPC框架的设计和实现，同时也通过实现RPC框架的过程掌握了该如何将架构设计原则应用到实际项目中，构建一个高质量、可维护的软件系统。

项目我已经放到了GitHub上，欢迎star和fork。

---

互动话题：你对这个RPC框架的实现有什么建议或改进意见？你在实际工作中使用过哪些优秀的RPC框架？欢迎在评论区分享你的观点。

关于作者

Kenyon，资深软件架构师，15年的软件开发和技术管理经验，从程序员做到企业技术高管。多年企业数字化转型和软件架构设计经验，善于帮助企业构建高质量、可维护的软件系统，目前专注技术管理、架构设计、AI技术应用和落地；全网统一名称"六边形架构"，欢迎关注交流。

原创不易，转载请联系授权，如果觉得有帮助，请点赞、收藏、转发三连支持！