【BlossomRPC】接入注册中心

文章目录

Nacos
Zookeeper
自研配置中心

RPC项目

配置中心项目

网关项目

这是BlossomRPC项目的最后一篇文章了，接入完毕注册中心，一个完整的RPC框架就设计完成了。
对于项目对注册中心的整合，其实我们只需要再服务启动的时候将ip/port/servicename注册到注册中心上即可。
注册中心这里只是一个简单的ip/port信息的存储器而已。
那么我们就需要考虑用什么样的一种方式来引入注册中心。
这里可以考虑使用Spring的AutoConfiguration然后配合Conditional类型的注解进行判断使用那个注册中心。
这里我提供了一个RegisterService接口提供抽象的注册方法，只需要实现这个接口就可以再项目中引入自己实现的注册中心了。

public interface RegisterService {default void init(){}void register(RpcServiceInstance instance);default void unregister(RpcServiceInstance instance){}RpcServiceInstance discovery(RpcServiceInstance instance);
}

Nacos

我们首先以Nacos为例。实现所有的接口方法。

package blossom.project.rpc.nacos;import blossom.project.rpc.common.register.RegisterService;
import blossom.project.rpc.common.register.RpcServiceInstance;
import blossom.project.rpc.common.loadbalance.LoadBalanceStrategy;
import blossom.project.rpc.common.loadbalance.PollLoadBalance;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.NamingFactory;
import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.pojo.Instance;
import lombok.extern.slf4j.Slf4j;import java.util.List;
import java.util.Objects;/*** @author: ZhangBlossom* @date: 2023/12/19 23:46* @contact: QQ:4602197553* @contact: WX:qczjhczs0114* @blog: https://blog.csdn.net/Zhangsama1* @github: https://github.com/ZhangBlossom* NacosRegisterService类*/
@Slf4j
public class NacosRegisterService implements RegisterService {private NamingService namingService;private LoadBalanceStrategy<Instance> loadBalanceStrategy= new PollLoadBalance<>();public NacosRegisterService(){}public NacosRegisterService(String serverAddress) {try {this.namingService = NamingFactory.createNamingService(serverAddress);} catch (NacosException e) {throw new RuntimeException(e);}}public NacosRegisterService(String serverAddress, LoadBalanceStrategy loadBalanceStrategy) {try {this.namingService = NamingFactory.createNamingService(serverAddress);this.loadBalanceStrategy = loadBalanceStrategy;} catch (NacosException e) {throw new RuntimeException(e);}}@Overridepublic void register(RpcServiceInstance instance) {if (Objects.isNull(instance)) {log.info("the Reigster Service Info can not be null!!!");return;}log.info("start to register instance to Nacos: {}",instance);try {//注册服务  服务名称：blossom.project.rpc.core.service.RpcServicenamingService.registerInstance(instance.getServiceName(), instance.getServiceIp(),instance.getServicePort());} catch (NacosException e) {log.error("register the ServiceInstance to Nacos failed!!!");throw new RuntimeException(e);}}@Overridepublic void unregister(RpcServiceInstance instance) {if (Objects.isNull(instance)) {log.info("the Reigster Service Info can not be null!!!");return;}log.info("start to unregister instance to Nacos: {}",instance);try {//进行服务注销namingService.deregisterInstance(instance.getServiceName(), instance.getServiceIp(),instance.getServicePort());} catch (NacosException e) {log.error("unregister the ServiceInstance from Nacos failed!!!");throw new RuntimeException(e);}}@Overridepublic RpcServiceInstance discovery(RpcServiceInstance instance) {try {List<Instance> instances = namingService.selectInstances(instance.getServiceName(),instance.getGroupName(), true);Instance rpcInstance = loadBalanceStrategy.choose(instances);if (Objects.nonNull(rpcInstance)) {return RpcServiceInstance.builder().serviceIp(rpcInstance.getIp()).servicePort(rpcInstance.getPort()).serviceName(rpcInstance.getServiceName()).build();}return null;} catch (NacosException e) {log.error("discovery the ServiceInstance from Nacos failed!!!");throw new RuntimeException(e);}}}

之后，我们得考虑，如何让用户无感知的只需要启动项目和引入依赖，就可以完成服务的注册。
我们考虑使用@AutoConfiguration的方式来进行。
同时，还得预留一种情况，就是用于自研的注册中心。

package blossom.project.rpc.nacos;import blossom.project.rpc.common.constants.RpcCommonConstants;
import blossom.project.rpc.common.loadbalance.LoadBalanceStrategy;
import blossom.project.rpc.common.loadbalance.PollLoadBalance;
import blossom.project.rpc.common.register.RegisterService;
import org.springframework.beans.BeansException;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
import org.springframework.boot.autoconfigure.AutoConfigureOrder;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.EnvironmentAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Conditional;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.core.env.Environment;import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Map;
import java.util.Objects;/*** @author: ZhangBlossom* @date: 2023/12/22 18:50* @contact: QQ:4602197553* @contact: WX:qczjhczs0114* @blog: https://blog.csdn.net/Zhangsama1* @github: https://github.com/ZhangBlossom* NacosAutoConfiguration类*/
@Configuration
//@AutoConfiguration
//@AutoConfigureBefore(value = RegisterService.class)
@AutoConfigureOrder(value = Integer.MAX_VALUE)
//@Conditional(OnNacosClientClassCondition.class)
public class NacosAutoConfiguration implementsApplicationContextAware, EnvironmentAware {/*** 这个bean只会在存在nacos的依赖的时候才会创建** @return*/@Primary@Bean(name = "nacosRegisterService")@ConditionalOnMissingBean(name = "spiRegisterService")public RegisterService nacosRegisterService() {//创建注册中心// 优先检查是否存在 SPI 实现类// 获取Nacos相关配置，例如服务器地址等//String serverAddress = "localhost:8848"; // 从配置中读取Nacos服务器地址// ... 其他所需配置String registerAddress = environment.getProperty(RpcCommonConstants.REGISTER_ADDRESS);try {// 使用反射创建NamingService实例//Class<?> namingFactoryClass =//        Class.forName("com.alibaba.nacos.api.naming.NamingFactory");//Method createNamingServiceMethod =//        namingFactoryClass.getMethod("createNamingService", String.class);//Object namingServiceInstance = createNamingServiceMethod.invoke(null, serverAddress);// 创建NacosRegisterService实例Class<?> nacosRegisterServiceClass = Class.forName(RpcCommonConstants.NACOS_REGISTER_CLASS);Constructor<?> constructor = nacosRegisterServiceClass.getConstructor(String.class,LoadBalanceStrategy.class);return (RegisterService) constructor.newInstance(registerAddress, new PollLoadBalance<>());} catch (ClassNotFoundException | NoSuchMethodException | IllegalAccessException | InstantiationException |InvocationTargetException e) {throw new IllegalStateException("Failed to create NacosRegisterService", e);}}private ApplicationContext applicationContext;@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {this.applicationContext = applicationContext;}private Environment environment;@Overridepublic void setEnvironment(Environment environment) {this.environment = environment;}
}

通过这种方式，只要引入了Nacos的依赖，在项目启动的时候就会使用Nacos作为项目的注册中心。
同时，如果用户也提供了自己的注册中心，那么会优先使用用户自己的注册中心来进行服务注册。
而用户自己的注册中心的实现，使用的是SPI的方式。

package blossom.project.rpc.core.proxy.spring;import blossom.project.rpc.common.constants.RpcCommonConstants;
import blossom.project.rpc.common.loadbalance.LoadBalanceStrategy;
import blossom.project.rpc.common.loadbalance.PollLoadBalance;
import blossom.project.rpc.common.register.RegisterService;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.EnvironmentAware;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;import java.util.Map;
import java.util.ServiceLoader;@Configuration
public class SpringRegisterServicePostProcessor implementsBeanDefinitionRegistryPostProcessor ,EnvironmentAware,ApplicationContextAware {private Environment environment;@Overridepublic void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {ServiceLoader<RegisterService> serviceLoader = ServiceLoader.load(RegisterService.class);if (!serviceLoader.iterator().hasNext()) {// 没有通过SPI找到实现，加载Nacos或Zookeeper的实现String registerAddress = environment.getProperty(RpcCommonConstants.REGISTER_ADDRESS);registerServiceBeanDefinition(registry, registerAddress);} else {// 通过SPI找到了实现，将其注册到Spring容器registerServiceViaSpi(serviceLoader, registry);}}private void registerServiceViaSpi(ServiceLoader<RegisterService> serviceLoader, BeanDefinitionRegistry registry) {// 获取SPI的RegisterService实现RegisterService registerService = serviceLoader.iterator().next();// 创建BeanDefinitionBeanDefinition beanDefinition = BeanDefinitionBuilder.genericBeanDefinition(registerService.getClass()).getBeanDefinition();// 注册BeanDefinition到Spring容器registry.registerBeanDefinition("spiRegisterService", beanDefinition);}private void registerServiceBeanDefinition(BeanDefinitionRegistry registry, String registerAddress) {try {registerReflectiveService(registry, RpcCommonConstants.NACOS_REGISTER_CLASS, registerAddress);} catch (Exception e) {registerReflectiveService(registry, RpcCommonConstants.ZK_REGISTER_CLASS, registerAddress);}}private void registerReflectiveService(BeanDefinitionRegistry registry, String className, String registerAddress) {try {Class<?> registerServiceClass = Class.forName(className);BeanDefinition beanDefinition =BeanDefinitionBuilder.genericBeanDefinition(registerServiceClass).getBeanDefinition();registry.registerBeanDefinition(className, beanDefinition);System.out.println(registry.getBeanDefinition(className));} catch (Exception e) {throw new RuntimeException("Failed to register " + className, e);}}@Overridepublic void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {// No implementation required for this method in this context}@Overridepublic void setEnvironment(Environment environment) {this.environment = environment;}private ApplicationContext applicationContext;@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {this.applicationContext = applicationContext;}}

好的，这里我们已Nacos为例，对服务进行启动。
成功完成服务实例的注册
在这里插入图片描述启动多个服务实例也可以

同理，对于zk，也是一样的方法。

Zookeeper

@Slf4j
public class ZookeeperRegisterService implements RegisterService {private static final String REGISTRY_PATH = "/rpc_registry";/*** zk注册中心*/private final ServiceDiscovery<RpcServiceInstance> serviceDiscovery;private LoadBalanceStrategy<ServiceInstance<RpcServiceInstance>> loadBalanceStrategy;public ZookeeperRegisterService(String serverAddress,LoadBalanceStrategy loadBalanceStrategy) throws Exception {CuratorFramework client = CuratorFrameworkFactory.newClient(serverAddress,new ExponentialBackoffRetry(2000, 3));client.start();JsonInstanceSerializer<RpcServiceInstance> serializer = new JsonInstanceSerializer<>(RpcServiceInstance.class);this.serviceDiscovery =ServiceDiscoveryBuilder.builder(RpcServiceInstance.class).client(client).serializer(serializer).basePath(REGISTRY_PATH).build();this.serviceDiscovery.start();this.loadBalanceStrategy = loadBalanceStrategy;}@Overridepublic void register(RpcServiceInstance instance) {if (Objects.isNull(instance)) {log.info("the Reigster Service Info can not be null!!!");return;}log.info("start to register instance to Zookeeper: {}",instance);try {ServiceInstance<RpcServiceInstance> serviceInstance =ServiceInstance.<RpcServiceInstance>builder().name(instance.getServiceName()).address(instance.getServiceIp()).port(instance.getServicePort()).payload(instance).build();this.serviceDiscovery.registerService(serviceInstance);} catch (Exception e) {throw new RuntimeException(e);}}@Overridepublic RpcServiceInstance discovery(RpcServiceInstance instance) {Collection<ServiceInstance<RpcServiceInstance>> serviceInstances = null;try {serviceInstances = this.serviceDiscovery.queryForInstances(instance.getServiceName());ServiceInstance<RpcServiceInstance> serviceInstance =this.loadBalanceStrategy.choose((List<ServiceInstance<RpcServiceInstance>>) serviceInstances);if (serviceInstance != null) {return serviceInstance.getPayload();}return null;} catch (Exception e) {throw new RuntimeException(e);}}}public class OnZookeeperClientClassCondition implements Condition {@Overridepublic boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {try {Class.forName(ZK_DISCOVERY_CLASS);return true;} catch (ClassNotFoundException e) {return false;}}