文章目录
- 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);
}
}
@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为例，对服务进行启动。
成功完成服务实例的注册
启动多个服务实例也可以
同理，对于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机制的方式，来让用户引入自己的注册中心。
用户在项目启动的时候通过SPI的方式提供自己实现的注册中心代码即可。
如果不存在会扫描是否存在Nacos/Zk，如果都不存在，就报错，否则优先使用用户自定义的配置中心。
到此为止，一个简单的自研配置中心就完成了。
