likes
comments
collection
share

【BlossomRPC】接入注册中心

作者站长头像
站长
· 阅读数 2

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);
        }
    }


    @Override
    public 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.RpcService
            namingService.registerInstance(instance.getServiceName(), instance.getServiceIp(),
                    instance.getServicePort());
        } catch (NacosException e) {
            log.error("register the ServiceInstance to Nacos failed!!!");
            throw new RuntimeException(e);
        }
    }

    @Override
    public 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);
        }

    }

    @Override
    public 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 implements
        ApplicationContextAware, 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;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    private Environment environment;

    @Override
    public 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 implements
        BeanDefinitionRegistryPostProcessor ,
        EnvironmentAware,
        ApplicationContextAware {

    private Environment environment;


    @Override
    public 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();

        // 创建BeanDefinition
        BeanDefinition 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);
        }
    }

    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
        // No implementation required for this method in this context
    }

    @Override
    public void setEnvironment(Environment environment) {
        this.environment = environment;
    }

    private ApplicationContext applicationContext;
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }


}

好的，这里我们已Nacos为例，对服务进行启动。 成功完成服务实例的注册 【BlossomRPC】接入注册中心启动多个服务实例也可以 【BlossomRPC】接入注册中心 同理，对于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;
    }

    @Override
    public 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);
        }
    }

    @Override
    public 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 {


    @Override
    public boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {
        try {
            Class.forName(ZK_DISCOVERY_CLASS);
            return true;
        } catch (ClassNotFoundException e) {
            return false;
        }
    }

自研配置中心

上文提到，有可能用户会使用自己的注册中心。所以我提供了基于spi机制的方式，来让用户引入自己的注册中心。 【BlossomRPC】接入注册中心 用户在项目启动的时候通过SPI的方式提供自己实现的注册中心代码即可。 如果不存在会扫描是否存在Nacos/Zk，如果都不存在，就报错，否则优先使用用户自定义的配置中心。

到此为止，一个简单的自研配置中心就完成了。