RPC框架-Gitee代碼(麻煩點個Starred, 支持一下吧)
RPC框架-GitHub代碼(麻煩點個Starred, 支持一下吧)

服務(wù)注冊

a.添加服務(wù)節(jié)點和主機節(jié)點

主要完成服務(wù)注冊和發(fā)現(xiàn)的功能，其具體流程如下：

• 1.服務(wù)提供方將服務(wù)注冊到注冊中心中

• 2.消費端拉取服務(wù)列表。

• 3.消費端簡單的選取一個可以服務(wù)（后續(xù)會進行改造，實現(xiàn)負載均衡）

1.修改framework/common的Constants類：定義注冊中心的路徑常量

public class Constant {// 略........// 服務(wù)提供方的在注冊中心的基礎(chǔ)路徑public static final String BASE_PROVIDERS_PATH = "/dcyrpc-metadata/providers";// 服務(wù)調(diào)用方的在注冊中心的基礎(chǔ)路徑public static final String BASE_CONSUMERS_PATH = "/dcyrpc-metadata/consumers";
}

2.在core中引入common的依賴項

3.修改framework/core的DcyRpcBootstrap類：定義一些相關(guān)的基礎(chǔ)配置

• 定義相關(guān)變量：應用名稱, Config, 默認端口
• 定義Zookeeper實例
• 完善方法代碼：application() / registry() / protocol() / publish()
• publish()發(fā)布服務(wù)：將接口與匹配的實現(xiàn)注冊到服務(wù)中心

// 略......
private static final DcyRpcBootstrap dcyRpcBootstrap = new DcyRpcBootstrap();
// 定義一些相關(guān)的基礎(chǔ)配置
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;
}/*** 配置當前暴露的服務(wù)使用的協(xié)議* @param protocolConfig 協(xié)議的封裝* @return this*/
public DcyRpcBootstrap protocol(ProtocolConfig protocolConfig) {this.protocolConfig = protocolConfig;if (log.isDebugEnabled()) {log.debug("當前工程使用了：{}協(xié)議進行序列化", protocolConfig.toString());}return this;
}/*** --------------------------------服務(wù)提供方的相關(guān)api--------------------------------*//*** 發(fā)布服務(wù)：將接口與匹配的實現(xiàn)注冊到服務(wù)中心* @param service 封裝需要發(fā)布的服務(wù)* @return*/
public DcyRpcBootstrap publish(ServiceConfig<?> service) {// 服務(wù)名稱的節(jié)點String parentNode = Constant.BASE_PROVIDERS_PATH + "/" + service.getInterface().getName();// 判斷節(jié)點是否存在，不存在則創(chuàng)建節(jié)點(持久)if (!ZookeeperUtils.existNode(zooKeeper, parentNode, null)) {ZookeeperNode zookeeperNode = new ZookeeperNode(parentNode, null);ZookeeperUtils.createNode(zooKeeper, zookeeperNode, null, CreateMode.PERSISTENT);}// 創(chuàng)建本機的臨時節(jié)點，ip:port// 服務(wù)提供方的端口(一般自己設(shè)定)，還需要獲取ip的方法// /dcyrpc-metadata/providers/com.dcyrpc.DcyRpc/192.168.195.1:8088String 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("服務(wù){(diào)}，已經(jīng)被注冊", service.getInterface().getName());}return this;
}// 略....../*** 啟動netty服務(wù)*/
public void start() {try {Thread.sleep(10000);} catch (InterruptedException e) {throw new RuntimeException(e);}
}// 略......}

4.修改framework/common的utils.zookeeper.ZookeeperUtils類：添加方法：判斷節(jié)點是否存在

/*** 判斷節(jié)點是否存在* @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("判斷節(jié)點：{} 是否存在時發(fā)生異常：", node, e);throw new ZookeeperException(e);}
}

5.修改framework/common的exceptions.ZookeeperException類：完善內(nèi)容

public class ZookeeperException extends RuntimeException{public ZookeeperException() {super();}public ZookeeperException(Throwable cause) {super(cause);}
}

6.在framework/common的utils包下，創(chuàng)建NetUtils類：Network工具類

/*** Network utils*/
@Slf4j
public class NetUtils {public static String getIp() {try {// 獲取所有的網(wǎng)卡信息Enumeration<NetworkInterface> interfaces = NetworkInterface.getNetworkInterfaces();while (interfaces.hasMoreElements()) {NetworkInterface iface = interfaces.nextElement();// 過濾非回環(huán)接口和虛擬接口if (iface.isLoopback() || iface.isVirtual() || !iface.isUp()) {continue;}Enumeration<InetAddress> addresses = iface.getInetAddresses();while (addresses.hasMoreElements()) {InetAddress addr = addresses.nextElement();// 過濾IPv6地址和回環(huán)地址if (addr instanceof Inet6Address || addr.isLoopbackAddress()) {continue;}String ipAddress = addr.getHostAddress();System.out.println("局域網(wǎng)IP地址：" + ipAddress);return ipAddress;}}throw new NetworkException();} catch (SocketException e) {log.error("獲取局域網(wǎng)IP時發(fā)送異常", e);throw new NetworkException(e);}}
}

7.在framework/common的exceptions包下創(chuàng)建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，甚至是自己獨立開發(fā)注冊中心。為后來的擴展提供可能性，所以在整個工程中我們再也不能單獨的面向具體的對象編程，而是面向抽象，我們將抽象出**【注冊中心】**整個抽象的概念

1.在core下com.dcyrpc下創(chuàng)建discovery包，創(chuàng)建Registry接口：抽象注冊中心接口：注冊服務(wù)，發(fā)現(xiàn)服務(wù)，下線服務(wù)

/*** 抽象注冊中心：注冊服務(wù)，發(fā)現(xiàn)服務(wù)，下線服務(wù)*/
public interface Registry {/*** 注冊服務(wù)* @param serviceConfig 服務(wù)的配置內(nèi)容*/public void register(ServiceConfig<?> serviceConfig);
}

2.在core下com.dcyrpc.discovery下創(chuàng)建AbstractRegistry抽象類：提煉共享內(nèi)容，還可以做模板方法 (待開發(fā))

/*** 提煉共享內(nèi)容，還可以做模板方法* 所有注冊中心都有的公共方法*/
public abstract class AbstractRegistry implements Registry{
}

3.在core下com.dcyrpc.discovery下創(chuàng)建impl包，創(chuàng)建ZookeeperRegistry類繼承AbstractRegistry

• 把DcyRpcBootstrap類里的publish()方法，提煉到該類中

@Slf4j
public class ZookeeperRegistry extends AbstractRegistry {private ZooKeeper zooKeeper = ZookeeperUtils.createZookeeper();@Overridepublic void register(ServiceConfig<?> service) {// 服務(wù)名稱的節(jié)點String parentNode = Constant.BASE_PROVIDERS_PATH + "/" + service.getInterface().getName();// 判斷節(jié)點是否存在，不存在則創(chuàng)建節(jié)點(持久)if (!ZookeeperUtils.existNode(zooKeeper, parentNode, null)) {ZookeeperNode zookeeperNode = new ZookeeperNode(parentNode, null);ZookeeperUtils.createNode(zooKeeper, zookeeperNode, null, CreateMode.PERSISTENT);}// 創(chuàng)建本機的臨時節(jié)點，ip:port// 服務(wù)提供方的端口(一般自己設(shè)定)，還需要獲取ip的方法// /dcyrpc-metadata/providers/com.dcyrpc.DcyRpc/192.168.195.1:8088// TODO:后續(xù)處理端口問題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("服務(wù){(diào)}，已經(jīng)被注冊", service.getInterface().getName());}}
}

4.修改DcyRpcBootstrap

// 略.....
private static final DcyRpcBootstrap dcyRpcBootstrap = new DcyRpcBootstrap();// 定義一些相關(guān)的基礎(chǔ)配置
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;
}// 略...../*** 發(fā)布服務(wù)：將接口與匹配的實現(xiàn)注冊到服務(wù)中心* @param service 封裝需要發(fā)布的服務(wù)* @return*/
public DcyRpcBootstrap publish(ServiceConfig<?> service) {// 抽象了注冊中心的概念，使用注冊中心的一個實現(xiàn)完成注冊zookeeperRegistry.register(service);return this;
}/*** 批量發(fā)布服務(wù)* @param services 封裝需要發(fā)布的服務(wù)集合* @return this*/
public DcyRpcBootstrap publish(List<ServiceConfig<?>> services) {for (ServiceConfig<?> service : services) {this.publish(service);}return this;
}// 略.....}

5.修改RegistryConfig類，將類放入discovery包下

public class RegistryConfig {// 定義連接的 urlprivate 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("未發(fā)現(xiàn)合適的注冊中心");}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中創(chuàng)建DiscoveryException類：服務(wù)注冊與發(fā)現(xiàn)異常處理

/*** 服務(wù)注冊與發(fā)現(xiàn)異常處理*/
public class DiscoveryException extends RuntimeException{public DiscoveryException() {super();}public DiscoveryException(String message) {super(message);}public DiscoveryException(Throwable cause) {super(cause);}
}

c.本地服務(wù)列表

服務(wù)調(diào)用方需要通過服務(wù)中心發(fā)現(xiàn)服務(wù)列表

• 1.使用Map進行服務(wù)列表的存儲
• 2.使用動態(tài)代理生成代理對象
• 3.從注冊中心，尋找一個可用的服務(wù)

1.修改DcyRpcBootstrap部分代碼：使用Map進行服務(wù)列表的存儲

// 維護已經(jīng)發(fā)布且暴露的服務(wù)列表 key:interface的全限定名  value:ServiceConfig
private static final Map<String, ServiceConfig<?>> SERVERS_LIST = new HashMap<>(16);/*** 發(fā)布服務(wù)：將接口與匹配的實現(xiàn)注冊到服務(wù)中心* @param service 封裝需要發(fā)布的服務(wù)* @return*/
public DcyRpcBootstrap publish(ServiceConfig<?> service) {// 抽象了注冊中心的概念，使用注冊中心的一個實現(xiàn)完成注冊zookeeperRegistry.register(service);// 1.當服務(wù)調(diào)用方，通過接口、方法名、具體的方法參數(shù)列表 發(fā)起調(diào)用，提供方怎么知道使用哪一個實現(xiàn)//  (1) new 1 個//  (2) spring beanFactory.getBean(Class)//  (3) 自己維護映射關(guān)系SERVERS_LIST.put(service.getInterface().getName(),  service);return this;
}

2.修改ReferenceConfig部分代碼

// 略.....
private Registry registry;/*** 代理設(shè)計模式，生成一個API接口的代理對象* @return 代理對象*/
public T get() {// 使用動態(tài)代理完成工作ClassLoader classLoader = Thread.currentThread().getContextClassLoader();Class[] classes = new Class[]{interfaceRef};// 使用動態(tài)代理生成代理對象Object helloProxy = Proxy.newProxyInstance(classLoader, classes, new InvocationHandler() {@Overridepublic Object invoke(Object proxy, Method method, Object[] args) throws Throwable {// 1.發(fā)現(xiàn)服務(wù)，從注冊中心，尋找一個可用的服務(wù)// 傳入服務(wù)的名字，返回ip+端口 （InetSocketAddress可以封裝端口/ip/host name）InetSocketAddress address = registry.lookup(interfaceRef.getName());if (log.isInfoEnabled()){log.debug("服務(wù)調(diào)用方，發(fā)現(xiàn)了服務(wù){(diào)}的可用主機{}", interfaceRef.getName(), address);}// 2.使用netty連接服務(wù)器，發(fā)送 調(diào)用的 服務(wù)名字+方法名字+參數(shù)列表，得到結(jié)果return null;}});return (T) helloProxy;
}public Registry getRegistry() {return registry;
}public void setRegistry(Registry registry) {this.registry = registry;
}

3.修改Registry接口，添加發(fā)現(xiàn)服務(wù)的接口

/*** 從注冊中心拉取一個可用的服務(wù)* @param serviceName 服務(wù)名稱* @return 服務(wù)的ip+端口*/
InetSocketAddress lookup(String serviceName);

4.修改ZookeeperRegistry部分代碼，實現(xiàn)發(fā)現(xiàn)服務(wù)的業(yè)務(wù)邏

@Override
public InetSocketAddress lookup(String serviceName) {// 1.找到對應服務(wù)的節(jié)點String serviceNode = Constant.BASE_PROVIDERS_PATH + "/" + serviceName;// 2.從zk中獲取它的子節(jié)點，List<String> children = ZookeeperUtils.getChildren(zooKeeper, serviceNode, null);// 獲取所有的可用的服務(wù)列表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("未發(fā)現(xiàn)任何可用的服務(wù)主機");}return inetSocketAddressList.get(0);
}

5.修改ZookeeperUtils部分代碼，添加與實現(xiàn)獲取子節(jié)點的方法

/*** 查詢一個節(jié)點的子元素* @param zooKeeper* @param serviceNode 服務(wù)節(jié)點* @return 子元素列表*/
public static List<String> getChildren(ZooKeeper zooKeeper, String serviceNode, Watcher watcher) {try {return zooKeeper.getChildren(serviceNode, watcher);} catch (KeeperException | InterruptedException e) {log.error("獲取節(jié)點{}的子元素時發(fā)生異常：{}", serviceNode, e);throw new ZookeeperException(e);}
}