RPC框架-Gitee代码(麻烦点个Starred, 支持一下吧)
RPC框架-GitHub代码(麻烦点个Starred, 支持一下吧)
服务注册
- 服务注册
- a.添加服务节点和主机节点
- b.抽象注册中心
- c.本地服务列表
服务注册
a.添加服务节点和主机节点
主要完成服务注册和发现的功能，其具体流程如下：
-
1.服务提供方将服务注册到注册中心中
-
2.消费端拉取服务列表。
-
3.消费端简单的选取一个可以服务（后续会进行改造，实现负载均衡）
1.修改framework/common的Constants类：定义注册中心的路径常量
public class Constant {
// 略........
// 服务提供方的在注册中心的基础路径
public static final String BASE_PROVIDERS_PATH = "/dcyrpc-metadata/providers";
// 服务调用方的在注册中心的基础路径
public static final String BASE_CONSUMERS_PATH = "/dcyrpc-metadata/consumers";
}
2.在core中引入common的依赖项
3.修改framework/core的DcyRpcBootstrap类：定义一些相关的基础配置
- 定义相关变量：应用名称, Config, 默认端口
- 定义Zookeeper实例
- 完善方法代码：application() / registry() / protocol() / publish()
- publish()发布服务：将接口与匹配的实现注册到服务中心
// 略......
private static final DcyRpcBootstrap dcyRpcBootstrap = new DcyRpcBootstrap();
// 定义一些相关的基础配置
private String applicationName = "default";
private RegistryConfig registryConfig;
private ProtocolConfig protocolConfig;
private int port = 8088;
// 维护一个Zookeeper实例
private ZooKeeper zooKeeper;
// 略......
/**
* 定义当前应用的名字
* @param applicationName 应用名称
* @return
*/
public DcyRpcBootstrap application(String applicationName) {
this.applicationName = applicationName;
return this;
}
/**
* 配置一个注册中心
* @param registryConfig 注册中心
* @return this
*/
public DcyRpcBootstrap registry(RegistryConfig registryConfig) {
// 维护一个zookeeper实例，但是，如果这样写就会将zookeeper和当前的工程耦合
zooKeeper = ZookeeperUtils.createZookeeper();
this.registryConfig = registryConfig;
return this;
}
/**
* 配置当前暴露的服务使用的协议
* @param protocolConfig 协议的封装
* @return this
*/
public DcyRpcBootstrap protocol(ProtocolConfig protocolConfig) {
this.protocolConfig = protocolConfig;
if (log.isDebugEnabled()) {
log.debug("当前工程使用了：{}协议进行序列化", protocolConfig.toString());
}
return this;
}
/**
* --------------------------------服务提供方的相关api--------------------------------
*/
/**
* 发布服务：将接口与匹配的实现注册到服务中心
* @param service 封装需要发布的服务
* @return
*/
public DcyRpcBootstrap publish(ServiceConfig<?> service) {
// 服务名称的节点
String parentNode = Constant.BASE_PROVIDERS_PATH + "/" + service.getInterface().getName();
// 判断节点是否存在，不存在则创建节点(持久)
if (!ZookeeperUtils.existNode(zooKeeper, parentNode, null)) {
ZookeeperNode zookeeperNode = new ZookeeperNode(parentNode, null);
ZookeeperUtils.createNode(zooKeeper, zookeeperNode, null, CreateMode.PERSISTENT);
}
// 创建本机的临时节点，ip:port
// 服务提供方的端口(一般自己设定)，还需要获取ip的方法
// /dcyrpc-metadata/providers/com.dcyrpc.DcyRpc/192.168.195.1:8088
String node = parentNode + "/" + NetUtils.getIp() + ":" + port;
if (!ZookeeperUtils.existNode(zooKeeper, node, null)) {
ZookeeperNode zookeeperNode = new ZookeeperNode(node, null);
ZookeeperUtils.createNode(zooKeeper, zookeeperNode, null, CreateMode.EPHEMERAL);
}
if (log.isDebugEnabled()) {
log.debug("服务{}，已经被注册", service.getInterface().getName());
}
return this;
}
// 略......
/**
* 启动netty服务
*/
public void start() {
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
// 略......
}
4.修改framework/common的utils.zookeeper.ZookeeperUtils类：添加方法：判断节点是否存在
/**
* 判断节点是否存在
* @param zooKeeper
* @param node
* @param watcher
* @return true:存在 false:不存在
*/
public static boolean existNode(ZooKeeper zooKeeper, String node, Watcher watcher) {
try {
return zooKeeper.exists(node, watcher) != null;
} catch (KeeperException | InterruptedException e) {
log.error("判断节点：{} 是否存在时发生异常：", node, e);
throw new ZookeeperException(e);
}
}
5.修改framework/common的exceptions.ZookeeperException类：完善内容
public class ZookeeperException extends RuntimeException{
public ZookeeperException() {
super();
}
public ZookeeperException(Throwable cause) {
super(cause);
}
}
6.在framework/common的utils包下，创建NetUtils类：Network工具类
/**
* Network utils
*/
@Slf4j
public class NetUtils {
public static String getIp() {
try {
// 获取所有的网卡信息
Enumeration<NetworkInterface> interfaces = NetworkInterface.getNetworkInterfaces();
while (interfaces.hasMoreElements()) {
NetworkInterface iface = interfaces.nextElement();
// 过滤非回环接口和虚拟接口
if (iface.isLoopback() || iface.isVirtual() || !iface.isUp()) {
continue;
}
Enumeration<InetAddress> addresses = iface.getInetAddresses();
while (addresses.hasMoreElements()) {
InetAddress addr = addresses.nextElement();
// 过滤IPv6地址和回环地址
if (addr instanceof Inet6Address || addr.isLoopbackAddress()) {
continue;
}
String ipAddress = addr.getHostAddress();
System.out.println("局域网IP地址：" + ipAddress);
return ipAddress;
}
}
throw new NetworkException();
} catch (SocketException e) {
log.error("获取局域网IP时发送异常", e);
throw new NetworkException(e);
}
}
}
7.在framework/common的exceptions包下创建NetworkException类：编写自定义异常
public class NetworkException extends RuntimeException{
public NetworkException() {
super();
}
public NetworkException(String message) {
super(message);
}
public NetworkException(Throwable cause) {
super(cause);
}
}
b.抽象注册中心
在当前项目中我们的确使用的是zookeeper作为我们项目的注册中心。但是，我们希望在我们的项目是可以扩展使用其他类型的注册中心的，如nacos，redis，甚至是自己独立开发注册中心。为后来的扩展提供可能性，所以在整个工程中我们再也不能单独的面向具体的对象编程，而是面向抽象，我们将抽象出**【注册中心】**整个抽象的概念
1.在core下com.dcyrpc下创建discovery包，创建Registry接口：抽象注册中心接口：注册服务，发现服务，下线服务
/**
* 抽象注册中心：注册服务，发现服务，下线服务
*/
public interface Registry {
/**
* 注册服务
* @param serviceConfig 服务的配置内容
*/
public void register(ServiceConfig<?> serviceConfig);
}
2.在core下com.dcyrpc.discovery下创建AbstractRegistry抽象类：提炼共享内容，还可以做模板方法 (待开发)
/**
* 提炼共享内容，还可以做模板方法
* 所有注册中心都有的公共方法
*/
public abstract class AbstractRegistry implements Registry{
}
3.在core下com.dcyrpc.discovery下创建impl包，创建ZookeeperRegistry类继承AbstractRegistry
- 把
DcyRpcBootstrap类里的publish()方法，提炼到该类中
@Slf4j
public class ZookeeperRegistry extends AbstractRegistry {
private ZooKeeper zooKeeper = ZookeeperUtils.createZookeeper();
@Override
public void register(ServiceConfig<?> service) {
// 服务名称的节点
String parentNode = Constant.BASE_PROVIDERS_PATH + "/" + service.getInterface().getName();
// 判断节点是否存在，不存在则创建节点(持久)
if (!ZookeeperUtils.existNode(zooKeeper, parentNode, null)) {
ZookeeperNode zookeeperNode = new ZookeeperNode(parentNode, null);
ZookeeperUtils.createNode(zooKeeper, zookeeperNode, null, CreateMode.PERSISTENT);
}
// 创建本机的临时节点，ip:port
// 服务提供方的端口(一般自己设定)，还需要获取ip的方法
// /dcyrpc-metadata/providers/com.dcyrpc.DcyRpc/192.168.195.1:8088
// TODO:后续处理端口问题
String node = parentNode + "/" + NetUtils.getIp() + ":" + 8088;
if (!ZookeeperUtils.existNode(zooKeeper, node, null)) {
ZookeeperNode zookeeperNode = new ZookeeperNode(node, null);
ZookeeperUtils.createNode(zooKeeper, zookeeperNode, null, CreateMode.EPHEMERAL);
}
if (log.isDebugEnabled()) {
log.debug("服务{}，已经被注册", service.getInterface().getName());
}
}
}
4.修改DcyRpcBootstrap
// 略.....
private static final DcyRpcBootstrap dcyRpcBootstrap = new DcyRpcBootstrap();
// 定义一些相关的基础配置
private String applicationName = "default";
private RegistryConfig registryConfig;
private ProtocolConfig protocolConfig;
private int port = 8088;
// 注册中心
private Registry zookeeperRegistry;
// 略.....
/**
* 配置一个注册中心
* @param registryConfig 注册中心
* @return this
*/
public DcyRpcBootstrap registry(RegistryConfig registryConfig) {
// 维护一个zookeeper实例，但是，如果这样写就会将zookeeper和当前的工程耦合
// 使用 registryConfig 获取一个注册中心
this.zookeeperRegistry = registryConfig.getRegistry();
return this;
}
// 略.....
/**
* 发布服务：将接口与匹配的实现注册到服务中心
* @param service 封装需要发布的服务
* @return
*/
public DcyRpcBootstrap publish(ServiceConfig<?> service) {
// 抽象了注册中心的概念，使用注册中心的一个实现完成注册
zookeeperRegistry.register(service);
return this;
}
/**
* 批量发布服务
* @param services 封装需要发布的服务集合
* @return this
*/
public DcyRpcBootstrap publish(List<ServiceConfig<?>> services) {
for (ServiceConfig<?> service : services) {
this.publish(service);
}
return this;
}
// 略.....
}
5.修改RegistryConfig类，将类放入discovery包下
public class RegistryConfig {
// 定义连接的 url
private final String connectString;
public RegistryConfig(String connectString) {
this.connectString = connectString;
}
/**
* 可以使用简单工厂来完成
* @return 具体的注册中心实例
*/
public Registry getRegistry() {
// 1.获取注册中心
// 1.获取类型
// 2.获取主机地址
String registryType = getRegistryType(connectString, true).toLowerCase().trim();
if (registryType.equals("zookeeper")) {
String host = getRegistryType(connectString, false);
return new ZookeeperRegistry(host, Constant.TIME_OUT);
}
throw new DiscoveryException("未发现合适的注册中心");
}
private String getRegistryType(String connectString, boolean ifType) {
String[] typeAndHost = connectString.split("://");
if (typeAndHost.length != 2) {
throw new RuntimeException("给定的注册中心连接url不合法");
}
if (ifType){
return typeAndHost[0];
}else {
return typeAndHost[1];
}
}
}
6.在framework/common的exceptions中创建DiscoveryException类：服务注册与发现异常处理
/**
* 服务注册与发现异常处理
*/
public class DiscoveryException extends RuntimeException{
public DiscoveryException() {
super();
}
public DiscoveryException(String message) {
super(message);
}
public DiscoveryException(Throwable cause) {
super(cause);
}
}
c.本地服务列表
服务调用方需要通过服务中心发现服务列表
- 1.使用Map进行服务列表的存储
- 2.使用动态代理生成代理对象
- 3.从注册中心，寻找一个可用的服务
1.修改DcyRpcBootstrap部分代码：使用Map进行服务列表的存储
// 维护已经发布且暴露的服务列表 key:interface的全限定名 value:ServiceConfig
private static final Map<String, ServiceConfig<?>> SERVERS_LIST = new HashMap<>(16);
/**
* 发布服务：将接口与匹配的实现注册到服务中心
* @param service 封装需要发布的服务
* @return
*/
public DcyRpcBootstrap publish(ServiceConfig<?> service) {
// 抽象了注册中心的概念，使用注册中心的一个实现完成注册
zookeeperRegistry.register(service);
// 1.当服务调用方，通过接口、方法名、具体的方法参数列表 发起调用，提供方怎么知道使用哪一个实现
// (1) new 1 个
// (2) spring beanFactory.getBean(Class)
// (3) 自己维护映射关系
SERVERS_LIST.put(service.getInterface().getName(), service);
return this;
}
2.修改ReferenceConfig部分代码
// 略.....
private Registry registry;
/**
* 代理设计模式，生成一个API接口的代理对象
* @return 代理对象
*/
public T get() {
// 使用动态代理完成工作
ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
Class[] classes = new Class[]{interfaceRef};
// 使用动态代理生成代理对象
Object helloProxy = Proxy.newProxyInstance(classLoader, classes, new InvocationHandler() {
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
// 1.发现服务，从注册中心，寻找一个可用的服务
// 传入服务的名字，返回ip+端口 （InetSocketAddress可以封装端口/ip/host name）
InetSocketAddress address = registry.lookup(interfaceRef.getName());
if (log.isInfoEnabled()){
log.debug("服务调用方，发现了服务{}的可用主机{}", interfaceRef.getName(), address);
}
// 2.使用netty连接服务器，发送 调用的 服务名字+方法名字+参数列表，得到结果
return null;
}
});
return (T) helloProxy;
}
public Registry getRegistry() {
return registry;
}
public void setRegistry(Registry registry) {
this.registry = registry;
}
3.修改Registry接口，添加发现服务的接口
/**
* 从注册中心拉取一个可用的服务
* @param serviceName 服务名称
* @return 服务的ip+端口
*/
InetSocketAddress lookup(String serviceName);
4.修改ZookeeperRegistry部分代码，实现发现服务的业务逻
@Override
public InetSocketAddress lookup(String serviceName) {
// 1.找到对应服务的节点
String serviceNode = Constant.BASE_PROVIDERS_PATH + "/" + serviceName;
// 2.从zk中获取它的子节点，
List<String> children = ZookeeperUtils.getChildren(zooKeeper, serviceNode, null);
// 获取所有的可用的服务列表
List<InetSocketAddress> inetSocketAddressList = children.stream().map(ipString -> {
String[] ipAndPort = ipString.split(":");
String ip = ipAndPort[0];
int port = Integer.valueOf(ipAndPort[1]);
return new InetSocketAddress(ip, port);
}).toList();
if (inetSocketAddressList.size() == 0){
throw new DiscoveryException("未发现任何可用的服务主机");
}
return inetSocketAddressList.get(0);
}
5.修改ZookeeperUtils部分代码，添加与实现获取子节点的方法
/**
* 查询一个节点的子元素
* @param zooKeeper
* @param serviceNode 服务节点
* @return 子元素列表
*/
public static List<String> getChildren(ZooKeeper zooKeeper, String serviceNode, Watcher watcher) {
try {
return zooKeeper.getChildren(serviceNode, watcher);
} catch (KeeperException | InterruptedException e) {
log.error("获取节点{}的子元素时发生异常：{}", serviceNode, e);
throw new ZookeeperException(e);
}
}
