溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
如果大家對RPC有一些了解的話，或多或者都會聽到過一些大名鼎鼎的RPC框架，比如Dobbo、gRPC。但是大部分人對于他們底層的實現原理其實不甚了解。
有一種比較好的學習方式：就是如果你想要了解一個框架的原理，你可以嘗試去寫一個簡易版的框架出來，就比如如果你想理解Spring IOC的思想，最好的方式就是自己實現一個小型的IOC容器，自己慢慢體會。
所以本文嘗試帶領大家去設計一個小型的RPC框架，同時對于框架會保持一些拓展點。
通過閱讀本文，你可以收獲：
理解RPC框架最核心的理念
學習在設計框架的時候，如何保持拓展性
本文會依賴一些組件，他們是實現RPC框架必要的一些知識，文中會盡量降低這些知識帶來的障礙。但是，最好期望讀者有以下知識基礎：
Zookeeper基本入門
Netty基本入門
RPC框架應該長什么樣子
我們首先來看一下：一個RPC框架是什么東西？
我們最直觀的感覺就是：
集成了RPC框架之后，通過配置一個注冊中心的地址。一個應用(稱為服務提供者)將某個接口(interface)“暴露”出去，另外一個應用(稱為服務消費者)通過“引用”這個接口(interface)，然后調用了一下，就很神奇的可以調用到另外一個應用的方法了
給我們的感覺就好像調用了一個本地方法一樣。即便兩個應用不是在同一個JVM中甚至兩個應用都不在同一臺機器中。
那他們是如何做到的呢？
其實啊，當我們的服務消費者調用某個RPC接口的方法之后，它的底層會通過動態代理，然后經過網絡調用，去到服務提供者的機器上，然后執行對應的方法。
接著方法的結果通過網絡傳輸返回到服務消費者那里，然后就可以拿到結果了。
整個過程如下圖：
那么這個時候，可能有人會問了：服務消費者怎么知道服務提供者在哪臺機器的哪個端口呢？
這個時候，就需要“注冊中心”登場了，具體來說是這樣子的：
服務提供者在啟動的時候，將自己應用所在機器的信息提交到注冊中心上面。
服務消費者在啟動的時候，將需要消費的接口所在機器的信息抓回來
這樣一來，服務消費者就有了一份服務提供者所在的機器列表了
“服務消費者”拿到了“服務提供者”的機器列表就可以通過網絡請求來發起請求了。
網絡客戶端，我們應該采用什么呢？有幾種選擇：
使用JDK原生BIO（也就是ServerSocket那一套）。阻塞式IO方法，無法支撐高并發。
使用JDK原生NIO(Selector、SelectionKey那一套)。非阻塞式IO，可以支持高并發，但是自己實現復雜，需要處理各種網絡問題。
使用大名鼎鼎的NIO框架Netty，天然支持高并發，封裝好，API易用。
作為一個有追求的程序員，我們要求開發出來的框架要求支持高并發、又要求簡單、還要快。當然是選擇Netty來實現了，使用Netty的一些很基本的API就能滿足我們的需求。
網絡協議定義
當然了，既然我們要使用網絡傳輸數據。我們首先要定義一套網絡協議出來。
你可能又要問了，啥叫網絡協議？
網絡協議，通俗理解，意思就是說我們的客戶端發送的數據應該長什么樣子，服務端可以去解析出來知道要做什么事情。話不多說，上代碼：
假設我們現在服務提供者有兩個類：
//?com.study.rpc.test.producer.HelloService
public?interface?HelloService?{
?String?sayHello(TestBean?testBean);
}
//?com.study.rpc.test.producer.TestBean
public?class?TestBean?{
?private?String?name;
?private?Integer?age;
?public?TestBean(String?name,?Integer?age)?{
?this.name?=?name;
?this.age?=?age;
?}
?//?getter?setter
}現在我要調用HelloService.sayHello(TestBean testBean)這個方法
作為“服務消費者”，應該怎么定義我們的請求，從而讓服務端知道我是要調用這個方法呢？
這需要我們將這個接口信息產生一個唯一的標識: 這個標識會記錄了接口名、具體是那個方法、然后具體參數是什么！
然后將這些信息組織起來發送給服務端，我這里的方式是將信息保存為一個JSON格式的字符串來傳輸。
比如上面的接口我們傳輸的數據大概是這樣的：
{
"interfaces":?"interface=com.study.rpc.test.producer.HelloService&method=sayHello&
parameter=com.study.rpc.test.producer.TestBean",
"requestId":?"3",
"parameter":?{
"com.study.rpc.test.producer.TestBean":?{
"age":?20,
"name":?"張三"
}
}
}嗯，我這里用一個JSON來標識這次調用是調用哪個接口的哪個方法，其中interface標識了唯一的類，parameter標識了里面具體有哪些參數， 其中key就是參數的類全限定名，value就是這個類的JSON信息。
可能看到這里，大家可能有意見了: 數據不一定用JSON格式傳輸啊，而且使用JSON也不一定性能最高啊。
你使用JDK的Serializable配合Netty的ObjectDecoder來實現，這當然也可以，其實這里是一個拓展點，我們應該要提供多種序列化方式來供用戶選擇
但是這里選擇了JSON的原因是因為它比較直觀，對于寫文章來說比較合理。
開發服務提供者
嗯，搞定了網絡協議之后，我們開始開發“服務提供者”了。對于服務提供者，因為我們這里是寫一個簡單版本的RPC框架，為了保持簡潔。
我們不會引入類似Spring之類的容器框架，所以我們需要定義一個服務提供者的配置類，它用于定義這個服務提供者是什么接口，然后它具體的實例對象是什么：
public?class?ServiceConfig{
?public?Class?type;
?public?T?instance;
?public?ServiceConfig(Classtype,?T?instance)?{
?this.type?=?type;
?this.instance?=?instance;
?}
?public?ClassgetType()?{
?return?type;
?}
?public?void?setType(Classtype)?{
?this.type?=?type;
?}
?public?T?getInstance()?{
?return?instance;
?}
?public?void?setInstance(T?instance)?{
?this.instance?=?instance;
?}
}有了這個東西之后，我們就知道需要暴露哪些接口了。
為了框架有一個統一的入口，我定義了一個類叫做ApplicationContext，可以認為這是一個應用程序上下文，他的構造函數，接收2個參數，代碼如下：
public?ApplicationContext(String?registryUrl,?ListserviceConfigs){
?//?1.?保存需要暴露的接口配置
?this.serviceConfigs?=?serviceConfigs?==?null???new?ArrayList<>()?:?serviceConfigs;
?//?step?2:?實例化注冊中心
?initRegistry(registryUrl);
?//?step?3:?將接口注冊到注冊中心，從注冊中心獲取接口，初始化服務接口列表
?RegistryInfo?registryInfo?=?null;
?InetAddress?addr?=?InetAddress.getLocalHost();
?String?hostname?=?addr.getHostName();
?String?hostAddress?=?addr.getHostAddress();
?registryInfo?=?new?RegistryInfo(hostname,?hostAddress,?port);
?doRegistry(registryInfo);
?
?
?//?step?4：初始化Netty服務器，接受到請求，直接打到服務提供者的service方法中
?if?(!this.serviceConfigs.isEmpty())?{
?//?需要暴露接口才暴露
?nettyServer?=?new?NettyServer(this.serviceConfigs,?interfaceMethods);
?nettyServer.init(port);
?}
}注冊中心設計
這里分為幾個步驟，首先保存了接口配置，接著初始化注冊中心,因為注冊中心可能會提供多種來供用戶選擇，所以這里需要定義一個注冊中心的接口：
public?interface?Registry?{
?/**
?*?將生產者接口注冊到注冊中心
?*
?*?@param?clazz?類
?*?@param?registryInfo?本機的注冊信息
?*/
?void?register(Class?clazz,?RegistryInfo?registryInfo)?throws?Exception;
}這里我們提供一個注冊的方法,這個方法的語義是將clazz對應的接口注冊到注冊中心。接收兩個參數，一個是接口的class對象，另一個是注冊信息，
里面包含了本機的一些基本信息，如下：
public?class?RegistryInfo?{
?private?String?hostname;
?private?String?ip;
?private?Integer?port;
?public?RegistryInfo(String?hostname,?String?ip,?Integer?port)?{
?this.hostname?=?hostname;
?this.ip?=?ip;
?this.port?=?port;
?}
?//?getter?setter
}好了，定義好注冊中心，回到之前的實例化注冊中心的地方，代碼如下：
/**
?*?注冊中心
?*/
private?Registry?registry;
private?void?initRegistry(String?registryUrl)?{
?if?(registryUrl.startsWith("zookeeper://"))?{
?registryUrl?=?registryUrl.substring(12);
?registry?=?new?ZookeeperRegistry(registryUrl);
?}?else?if?(registryUrl.startsWith("multicast://"))?{
?registry?=?new?MulticastRegistry(registryUrl);
?}
}這里邏輯也非常簡單，就是根據url的schema來判斷是那個注冊中心
注冊中心這里實現了2個實現類，分別使用zookeeper作為注冊中心，另外一個是使用廣播的方式作為注冊中心。
廣播注冊中心這里僅僅是做個示范，內部沒有實現。我們主要是實現了zookeeper的注冊中心。
（當然了，如果有興趣，可以實現更多的注冊中心供用戶選擇，比如redis之類的，這里只是為了保持“拓展點”）
那么實例化完注冊中心之后，回到上面的代碼：
注冊服務提供者
//?step?3:?將接口注冊到注冊中心，從注冊中心獲取接口，初始化服務接口列表 RegistryInfo?registryInfo?=?null; InetAddress?addr?=?InetAddress.getLocalHost(); String?hostname?=?addr.getHostName(); String?hostAddress?=?addr.getHostAddress(); registryInfo?=?new?RegistryInfo(hostname,?hostAddress,?port); doRegistry(registryInfo);
這里邏輯很簡單，就是獲取本機的的基本信息構造成RegistryInfo，然后調用了doRegistry方法：
/**
?*?接口方法對應method對象
?*/
private?MapinterfaceMethods?=?new?ConcurrentHashMap<>();
private?void?doRegistry(RegistryInfo?registryInfo)?throws?Exception?{
?for?(ServiceConfig?config?:?serviceConfigs)?{
?Class?type?=?config.getType();
?registry.register(type,?registryInfo);
?Method[]?declaredMethods?=?type.getDeclaredMethods();
?for?(Method?method?:?declaredMethods)?{
?String?identify?=?InvokeUtils.buildInterfaceMethodIdentify(type,?method);
?interfaceMethods.put(identify,?method);
?}
?}
}這里做了2件事情：
將接口注冊到注冊中心中
對于每一個接口的每一個方法，生成一個唯一標識，保存在interfaceMethods集合中
下面分別分析這兩件事情，首先是注冊方法：
因為我們用到了zookeeper，為了方便，引入了zookeeper的客戶端框架curator
<dependency> ?<groupId>org.apache.curatorgroupId> ?<artifactId>curator-recipesartifactId> ?<version>2.3.0version> dependency>
接著看代碼：
public?class?ZookeeperRegistry?implements?Registry?{
?private?CuratorFramework?client;
?public?ZookeeperRegistry(String?connectString)?{
?RetryPolicy?retryPolicy?=?new?ExponentialBackoffRetry(1000,?3);
?client?=?CuratorFrameworkFactory.newClient(connectString,?retryPolicy);
?client.start();
?try?{
?Stat?myRPC?=?client.checkExists().forPath("/myRPC");
?if?(myRPC?==?null)?{
?client.create()
?.creatingParentsIfNeeded()
?.forPath("/myRPC");
?}
?System.out.println("Zookeeper?Client初始化完畢......");
?}?catch?(Exception?e)?{
?e.printStackTrace();
?}
?}
?@Override
?public?void?register(Class?clazz,?RegistryInfo?registryInfo)?throws?Exception?{
?//?1.?注冊的時候，先從zk中獲取數據
?//?2.?將自己的服務器地址加入注冊中心中
?//?為每一個接口的每一個方法注冊一個臨時節點，然后key為接口方法的唯一標識，data為服務地址列表
?Method[]?declaredMethods?=?clazz.getDeclaredMethods();
?for?(Method?method?:?declaredMethods)?{
?String?key?=?InvokeUtils.buildInterfaceMethodIdentify(clazz,?method);
?String?path?=?"/myRPC/"?+?key;
?Stat?stat?=?client.checkExists().forPath(path);
?ListregistryInfos;
?if?(stat?!=?null)?{
?//?如果這個接口已經有人注冊過了，把數據拿回來，然后將自己的信息保存進去
?byte[]?bytes?=?client.getData().forPath(path);
?String?data?=?new?String(bytes,?StandardCharsets.UTF_8);
?registryInfos?=?JSONArray.parseArray(data,?RegistryInfo.class);
?if?(registryInfos.contains(registryInfo))?{
?//?正常來說，zk的臨時節點，斷開連接后，直接就沒了，但是重啟會經常發現存在節點，所以有了這樣的代碼
?System.out.println("地址列表已經包含本機【"?+?key?+?"】，不注冊了");
?}?else?{
?registryInfos.add(registryInfo);
?client.setData().forPath(path,?JSONArray.toJSONString(registryInfos).getBytes());
?System.out.println("注冊到注冊中心，路徑為：【"?+?path?+?"】?信息為："?+?registryInfo);
?}
?}?else?{
?registryInfos?=?new?ArrayList<>();
?registryInfos.add(registryInfo);
?client.create()
?.creatingParentsIfNeeded()
?//?臨時節點，斷開連接就關閉
?.withMode(CreateMode.EPHEMERAL)
?.forPath(path,?JSONArray.toJSONString(registryInfos).getBytes());
?System.out.println("注冊到注冊中心，路徑為：【"?+?path?+?"】?信息為："?+?registryInfo);
?}
?}
?}
}zookeeper注冊中心在初始化的時候，會建立好連接。然后注冊的時候，針對clazz接口的每一個方法，都會生成一個唯一標識
這里使用了InvokeUtils.buildInterfaceMethodIdentify方法：
public?static?String?buildInterfaceMethodIdentify(Class?clazz,?Method?method)?{
?Map<String,?String>?map?=?new?LinkedHashMap<>();
?map.put("interface",?clazz.getName());
?map.put("method",?method.getName());
?Parameter[]?parameters?=?method.getParameters();
?if?(parameters.length?>?0)?{
?StringBuilder?param?=?new?StringBuilder();
?for?(int?i?=?0;?i?<?parameters.length;?i++)?{
?Parameter?p?=?parameters[i];
?param.append(p.getType().getName());
?if?(i?<?parameters.length?-?1)?{
?param.append(",");
?}
?}
?map.put("parameter",?param.toString());
?}
?return?map2String(map);
}
public?static?String?map2String(Map<String,?String>?map)?{
?StringBuilder?sb?=?new?StringBuilder();
?Iterator<map.entry<string,?<=""?span="">String>>?iterator?=?map.entrySet().iterator();
?while?(iterator.hasNext())?{
?Map.Entry<String,?String>?entry?=?iterator.next();
?sb.append(entry.getKey()?+?"="?+?entry.getValue());
?if?(iterator.hasNext())?{
?sb.append("&");
?}
?}
?return?sb.toString();
}其實就是對接口的方法使用他們的限定名和參數來組成一個唯一的標識，比如 HelloService#sayHello(TestBean)生成的大概是這樣的：
interface=com.study.rpc.test.producer.HelloService&method=sayHello& parameter=com.study.rpc.test.producer.TestBean
接下來的邏輯就簡單了，在Zookeeper中的/myRPC路徑下面建立臨時節點，節點名稱為我們上面的接口方法唯一標識，數據內容為機器信息。
之所以采用臨時節點是因為：如果機器宕機了，連接斷開之后，消費者可以通過zookeeper的watcher機制感知到
大概看起來是這樣的：
?/myRPC/interface=com.study.rpc.test.producer.HelloService&method=sayHello&
?parameter=com.study.rpc.test.producer.TestBean
?[
?{
?"hostname":peer1,
?"port":8080
?},
?{
?"hostname":peer2,
?"port":8081
?}
?]通過這樣的方式，在服務消費的時候就可以拿到這樣的注冊信息，然后知道可以調用那臺機器的那個端口。
好了，注冊中心弄完了之后，我們回到前面說的注冊方法做的第二件事情,我們將每一個接口方法標識的方法放入了一個map中：
/** ?*?接口方法對應method對象 ?*/ private?Map<String,?Method>?interfaceMethods?=?new?ConcurrentHashMap<>();
這個的原因是因為，我們在收到網絡請求的時候，需要調用反射的方式調用method對象，所以存起來。
啟動網絡服務端接受請求
接下來我們就可以看第四步了：
//?step?4：初始化Netty服務器，接受到請求，直接打到服務提供者的service方法中
if?(!this.serviceConfigs.isEmpty())?{
?//?需要暴露接口才暴露
?nettyServer?=?new?NettyServer(this.serviceConfigs,?interfaceMethods);
?nettyServer.init(port);
}因為這里使用Netty來做的所以需要引入Netty的依賴：
<dependency> ?<groupId>io.nettygroupId> ?<artifactId>netty-allartifactId> ?<version>4.1.30.Finalversion> dependency>
接著來分析：
public?class?NettyServer?{
?/**
?*?負責調用方法的handler
?*/
?private?RpcInvokeHandler?rpcInvokeHandler;
?public?NettyServer(ListserverConfigs,?MapinterfaceMethods)?throws?InterruptedException?{
?this.rpcInvokeHandler?=?new?RpcInvokeHandler(serverConfigs,?interfaceMethods);
?}
?public?int?init(int?port)?throws?Exception?{
?EventLoopGroup?bossGroup?=?new?NioEventLoopGroup();
?EventLoopGroup?workerGroup?=?new?NioEventLoopGroup();
?ServerBootstrap?b?=?new?ServerBootstrap();
?b.group(bossGroup,?workerGroup)
?.channel(NioServerSocketChannel.class)
?.option(ChannelOption.SO_BACKLOG,?1024)
?.childHandler(new?ChannelInitializer()?{
?@Override
?protected?void?initChannel(SocketChannel?ch)?throws?Exception?{
?ByteBuf?delimiter?=?Unpooled.copiedBuffer("$$");
?//?設置按照分隔符“&&”來切分消息，單條消息限制為?1MB
?ch.pipeline().addLast(new?DelimiterBasedFrameDecoder(1024?*?1024,?delimiter));
?ch.pipeline().addLast(new?StringDecoder());
?ch.pipeline().addLast().addLast(rpcInvokeHandler);
?}
?});
?ChannelFuture?sync?=?b.bind(port).sync();
?System.out.println("啟動NettyService，端口為："?+?port);
?return?port;
?}
}這部分主要的都是netty的api，我們不做過多的說明，就簡單的說一下：
我們通過“&&”作為標識符號來區分兩條信息，然后一條信息的最大長度為1MB
所有邏輯都在RpcInvokeHandler中，這里面傳進去了配置的服務接口實例，以及服務接口實例每個接口方法唯一標識對應的Method對象的Map集合。
public?class?RpcInvokeHandler?extends?ChannelInboundHandlerAdapter?{
?/**
?*?接口方法唯一標識對應的Method對象
?*/
?private?Map<String,?Method>?interfaceMethods;
?/**
?*?接口對應的實現類
?*/
?private?Map<class,?Object>?interfaceToInstance;
?/**
?*?線程池，隨意寫的，不要吐槽
?*/
?private?ThreadPoolExecutor?threadPoolExecutor?=?new?ThreadPoolExecutor(10,
?50,?60,?TimeUnit.SECONDS,?new?LinkedBlockingQueue<>(100),
?new?ThreadFactory()?{
?AtomicInteger?m?=?new?AtomicInteger(0);
?@Override
?public?Thread?newThread(Runnable?r)?{
?return?new?Thread(r,?"IO-thread-"?+?m.incrementAndGet());
?}
?});
?public?RpcInvokeHandler(ListserviceConfigList,
?Map<String,?Method>?interfaceMethods)?{
?this.interfaceToInstance?=?new?ConcurrentHashMap<>();
?this.interfaceMethods?=?interfaceMethods;
?for?(ServiceConfig?config?:?serviceConfigList)?{
?interfaceToInstance.put(config.getType(),?config.getInstance());
?}
?}
?@Override
?public?void?channelRead(ChannelHandlerContext?ctx,?Object?msg)?throws?Exception?{
?try?{
?String?message?=?(String)?msg;
?//?這里拿到的是一串JSON數據，解析為Request對象，
?//?事實上這里解析網絡數據，可以用序列化方式，定一個接口，可以實現JSON格式序列化，或者其他序列化
?//?但是demo版本就算了。
?System.out.println("接收到消息："?+?msg);
?RpcRequest?request?=?RpcRequest.parse(message,?ctx);
?threadPoolExecutor.execute(new?RpcInvokeTask(request));
?}?finally?{
?ReferenceCountUtil.release(msg);
?}
?}
?@Override
?public?void?channelReadComplete(ChannelHandlerContext?ctx)?throws?Exception?{
?ctx.flush();
?}
?@Override
?public?void?exceptionCaught(ChannelHandlerContext?ctx,?Throwable?cause)?throws?Exception?{
?System.out.println("發生了異常..."?+?cause);
?cause.printStackTrace();
?ctx.close();
?}
?public?class?RpcInvokeTask?implements?Runnable?{
?private?RpcRequest?rpcRequest;
?RpcInvokeTask(RpcRequest?rpcRequest)?{
?this.rpcRequest?=?rpcRequest;
?}
?@Override
?public?void?run()?{
?try?{
?/*
?*?數據大概是這樣子的
?*?{"interfaces":"interface=com.study.rpc.test.producer.HelloService&method=sayHello?meter=com
?*?.study.rpc.test.producer.TestBean","requestId":"3","parameter":{"com.study.rpc.test.producer
?*?.TestBean":{"age":20,"name":"張三"}}}
?*/
?//?這里希望能拿到每一個服務對象的每一個接口的特定聲明
?String?interfaceIdentity?=?rpcRequest.getInterfaceIdentity();
?Method?method?=?interfaceMethods.get(interfaceIdentity);
?Map<String,?String>?map?=?string2Map(interfaceIdentity);
?String?interfaceName?=?map.get("interface");
?Class?interfaceClass?=?Class.forName(interfaceName);
?Object?o?=?interfaceToInstance.get(interfaceClass);
?String?parameterString?=?map.get("parameter");
?Object?result;
?if?(parameterString?!=?null)?{
?String[]?parameterTypeClass?=?parameterString.split(",");
?Map<String,?Object>?parameterMap?=?rpcRequest.getParameterMap();
?Object[]?parameterInstance?=?new?Object[parameterTypeClass.length];
?for?(int?i?=?0;?i?<?parameterTypeClass.length;?i++)?{
?String?parameterClazz?=?parameterTypeClass[i];
?parameterInstance[i]?=?parameterMap.get(parameterClazz);
?}
?result?=?method.invoke(o,?parameterInstance);
?}?else?{
?result?=?method.invoke(o);
?}
?//?寫回響應
?ChannelHandlerContext?ctx?=?rpcRequest.getCtx();
?String?requestId?=?rpcRequest.getRequestId();
?RpcResponse?response?=?RpcResponse.create(JSONObject.toJSONString(result),?interfaceIdentity,
?requestId);
?String?s?=?JSONObject.toJSONString(response)?+?"$$";
?ByteBuf?byteBuf?=?Unpooled.copiedBuffer(s.getBytes());
?ctx.writeAndFlush(byteBuf);
?System.out.println("響應給客戶端："?+?s);
?}?catch?(Exception?e)?{
?e.printStackTrace();
?}
?}
?
?public?static?Map<String,?String>?string2Map(String?str)?{
?String[]?split?=?str.split("&");
?Map<String,?String>?map?=?new?HashMap<>(16);
?for?(String?s?:?split)?{
?String[]?split1?=?s.split("=");
?map.put(split1[0],?split1[1]);
?}
?return?map;
?}
?}
}這里說明一下上面的邏輯：
channelRead方法用于接收消息，接收到的就是我們前面分析的那個JSON格式的數據，接著我們將消息解析成RpcRequest
public?class?RpcRequest?{
?private?String?interfaceIdentity;
?private?Map<String,?Object>?parameterMap?=?new?HashMap<>();
?private?ChannelHandlerContext?ctx;
?private?String?requestId;
?public?static?RpcRequest?parse(String?message,?ChannelHandlerContext?ctx)?throws?ClassNotFoundException?{
?/*
?*?{
?*?"interfaces":"interface=com.study.rpc.test.producer.HelloService&method=sayHello2?meter=java.lang
?*?.String,com.study.rpc.test.producer.TestBean",
?*?"parameter":{
?*?"java.lang.String":"haha",
?*?"com.study.rpc.test.producer.TestBean":{
?*?"name":"小王",
?*?"age":20
?*?}
?*?}
?*?}
?*/
?JSONObject?jsonObject?=?JSONObject.parseObject(message);
?String?interfaces?=?jsonObject.getString("interfaces");
?JSONObject?parameter?=?jsonObject.getJSONObject("parameter");
?Set<String>?strings?=?parameter.keySet();
?RpcRequest?request?=?new?RpcRequest();
?request.setInterfaceIdentity(interfaces);
?Map<String,?Object>?parameterMap?=?new?HashMap<>(16);
?String?requestId?=?jsonObject.getString("requestId");
?for?(String?key?:?strings)?{
?if?(key.equals("java.lang.String"))?{
?parameterMap.put(key,?parameter.getString(key));
?}?else?{
?Class?clazz?=?Class.forName(key);
?Object?object?=?parameter.getObject(key,?clazz);
?parameterMap.put(key,?object);
?}
?}
?request.setParameterMap(parameterMap);
?request.setCtx(ctx);
?request.setRequestId(requestId);
?return?request;
?}
}接著從request中解析出來需要調用的接口，然后通過反射調用對應的接口，得到結果后我們將響應封裝成PrcResponse寫回給客戶端：
public?class?RpcResponse?{
?private?String?result;
?private?String?interfaceMethodIdentify;
?private?String?requestId;
?public?String?getResult()?{
?return?result;
?}
?public?void?setResult(String?result)?{
?this.result?=?result;
?}
?public?static?RpcResponse?create(String?result,?String?interfaceMethodIdentify,?String?requestId)?{
?RpcResponse?response?=?new?RpcResponse();
?response.setResult(result);
?response.setInterfaceMethodIdentify(interfaceMethodIdentify);
?response.setRequestId(requestId);
?return?response;
?}
}里面包含了請求的結果JSON串，接口方法唯一標識，請求ID。數據大概看起來這個樣子：
{"interfaceMethodIdentify":"interface=com.study.rpc.test.producer.HelloService&method=sayHello&
parameter=com.study.rpc.test.producer.TestBean","requestId":"3",
"result":"\"牛逼,我收到了消息：TestBean{name='張三',?age=20}\""}通過這樣的信息，客戶端就可以通過響應結果解析出來。
測試服務提供者
既然我們代碼寫完了，現在需要測試一把：
首先我們先寫一個HelloService的實現類出來：
public?class?HelloServiceImpl?implements?HelloService?{
?@Override
?public?String?sayHello(TestBean?testBean)?{
?return?"牛逼,我收到了消息："?+?testBean;
?}
}接著編寫服務提供者代碼：
public?class?TestProducer?{
?public?static?void?main(String[]?args)?throws?Exception?{
?String?connectionString?=?"zookeeper://localhost1:2181,localhost2:2181,localhost3:2181";
?HelloService?service?=?new?HelloServiceImpl();
?ServiceConfig?config?=?new?ServiceConfig<>(HelloService.class,?service);
?ListserviceConfigList?=?new?ArrayList<>();
?serviceConfigList.add(config);
?ApplicationContext?ctx?=?new?ApplicationContext(connectionString,?serviceConfigList,
?null,?50071);
?}
}接著啟動起來,看到日志：
Zookeeper?Client初始化完畢......
注冊到注冊中心，路徑為：【/myRPC/interface=com.study.rpc.test.producer.HelloService&
method=sayHello?meter=com.study.rpc.test.producer.TestBean】
信息為：RegistryInfo{hostname='localhost',?ip='192.168.16.7',?port=50071}
啟動NettyService，端口為：50071這個時候，我們期望用NettyClient發送請求：
{
"interfaces":?"interface=com.study.rpc.test.producer.HelloService&
method=sayHello?meter=com.study.rpc.test.producer.TestBean",
"requestId":?"3",
"parameter":?{
"com.study.rpc.test.producer.TestBean":?{
"age":?20,
"name":?"張三"
}
}
}得到的響應應該是：
{"interfaceMethodIdentify":"interface=com.study.rpc.test.producer.HelloService&method=sayHello&
parameter=com.study.rpc.test.producer.TestBean","requestId":"3",
"result":"\"牛逼,我收到了消息：TestBean{name='張三',?age=20}\""}那么，可以編寫一個測試程序(這個程序僅僅用于中間測試用，讀者不必理解)：
public class NettyClient {
public static void main(String[] args) {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new StringDecoder());
ch.pipeline().addLast(new NettyClientHandler());
}
});
ChannelFuture sync = b.connect("127.0.0.1", 50071).sync();
sync.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
group.shutdownGracefully();
}
}
private static class NettyClientHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
JSONObject jsonObject = new JSONObject();
jsonObject.put("interfaces", "interface=com.study.rpc.test.producer" +
".HelloService&method=sayHello?meter=com.study.rpc.test.producer.TestBean");
JSONObject param = new JSONObject();
JSONObject bean = new JSONObject();
bean.put("age", 20);
bean.put("name", "張三");
param.put("com.study.rpc.test.producer.TestBean", bean);
jsonObject.put("parameter", param);
jsonObject.put("requestId", 3);
System.out.println("發送給服務端JSON為：" + jsonObject.toJSONString());
String msg = jsonObject.toJSONString() + "$$";
ByteBuf byteBuf = Unpooled.buffer(msg.getBytes().length);
byteBuf.writeBytes(msg.getBytes());
ctx.writeAndFlush(byteBuf);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("收到消息:" + msg);
}
}
}
啟動之后，看到控制臺輸出：
發送給服務端JSON為：{"interfaces":"interface=com.study.rpc.test.producer.HelloService&method=sayHello&
parameter=com.study.rpc.test.producer.TestBean","requestId":3,
"parameter":{"com.study.rpc.test.producer.TestBean":{"name":"張三","age":20}}}
收到消息:{"interfaceMethodIdentify":"interface=com.study.rpc.test.producer.HelloService&
method=sayHello?meter=com.study.rpc.test.producer.TestBean","requestId":"3",
"result":"\"牛逼,我收到了消息：TestBean{name='張三', age=20}\""}
bingo，完美實現了RPC的服務提供者。接下來我們只需要實現服務消費者就完成了。
開發服務消費者
服務消費者是同樣的處理，我們同樣要定義一個消費者的配置：
public?class?ReferenceConfig{
?private?Class?type;
?public?ReferenceConfig(Classtype)?{
?this.type?=?type;
?}
?public?ClassgetType()?{
?return?type;
?}
?public?void?setType(Classtype)?{
?this.type?=?type;
?}
}然后我們是統一入口，在ApplicationContext中修改代碼：
public?ApplicationContext(String?registryUrl,?ListserviceConfigs,
?ListreferenceConfigs,?int?port)?throws?Exception?{
?//?step?1:?保存服務提供者和消費者
?this.serviceConfigs?=?serviceConfigs?==?null???new?ArrayList<>()?:?serviceConfigs;
?this.referenceConfigs?=?referenceConfigs?==?null???new?ArrayList<>()?:?referenceConfigs;
?//?....
?
}
private?void?doRegistry(RegistryInfo?registryInfo)?throws?Exception?{
?for?(ServiceConfig?config?:?serviceConfigs)?{
?Class?type?=?config.getType();
?registry.register(type,?registryInfo);
?Method[]?declaredMethods?=?type.getDeclaredMethods();
?for?(Method?method?:?declaredMethods)?{
?String?identify?=?InvokeUtils.buildInterfaceMethodIdentify(type,?method);
?interfaceMethods.put(identify,?method);
?}
?}
?for?(ReferenceConfig?config?:?referenceConfigs)?{
?ListregistryInfos?=?registry.fetchRegistry(config.getType());
?if?(registryInfos?!=?null)?{
?interfacesMethodRegistryList.put(config.getType(),?registryInfos);
?initChannel(registryInfos);
?}
?}
}在注冊的時候，我們需要將需要消費的接口，通過注冊中心抓取出來，所以注冊中心要增加一個接口方法：
public?interface?Registry?{
?/**
?*?將生產者接口注冊到注冊中心
?*
?*?@param?clazz?類
?*?@param?registryInfo?本機的注冊信息
?*/
?void?register(Class?clazz,?RegistryInfo?registryInfo)?throws?Exception;
?/**
?*?為服務提供者抓取注冊表
?*
?*?@param?clazz?類
?*?@return?服務提供者所在的機器列表
?*/
?ListfetchRegistry(Class?clazz)?throws?Exception;
}獲取服務提供者的機器列表
具體在Zookeeper中的實現如下：
@Override
public?ListfetchRegistry(Class?clazz)?throws?Exception?{
?Method[]?declaredMethods?=?clazz.getDeclaredMethods();
?ListregistryInfos?=?null;
?for?(Method?method?:?declaredMethods)?{
?String?key?=?InvokeUtils.buildInterfaceMethodIdentify(clazz,?method);
?String?path?=?"/myRPC/"?+?key;
?Stat?stat?=?client.checkExists()
?.forPath(path);
?if?(stat?==?null)?{
?//?這里可以添加watcher來監聽變化，這里簡化了，沒有做這個事情
?System.out.println("警告：無法找到服務接口："?+?path);
?continue;
?}
?if?(registryInfos?==?null)?{
?byte[]?bytes?=?client.getData().forPath(path);
?String?data?=?new?String(bytes,?StandardCharsets.UTF_8);
?registryInfos?=?JSONArray.parseArray(data,?RegistryInfo.class);
?}
?}
?return?registryInfos;
}其實就是去zookeeper獲取節點中的數據，得到接口所在的機器信息，獲取到的注冊信息諸侯，就會調用以下代碼：
if?(registryInfos?!=?null)?{
?//?保存接口和服務地址
?interfacesMethodRegistryList.put(config.getType(),?registryInfos);
?//?初始化網絡連接
?initChannel(registryInfos);
}
private?void?initChannel(ListregistryInfos)?throws?InterruptedException?{
?for?(RegistryInfo?info?:?registryInfos)?{
?if?(!channels.containsKey(info))?{
?System.out.println("開始建立連接："?+?info.getIp()?+?",?"?+?info.getPort());
?NettyClient?client?=?new?NettyClient(info.getIp(),?info.getPort());
?client.setMessageCallback(message?->?{
?//?這里收單服務端返回的消息，先壓入隊列
?RpcResponse?response?=?JSONObject.parseObject(message,?RpcResponse.class);
?responses.offer(response);
?synchronized?(ApplicationContext.this)?{
?ApplicationContext.this.notifyAll();
?}
?});
?//?等待連接建立
?ChannelHandlerContext?ctx?=?client.getCtx();
?channels.put(info,?ctx);
?}
?}
}我們會針對每一個唯一的RegistryInfo建立一個連接，然后有這樣一段代碼：
client.setMessageCallback(message?->?{
?//?這里收單服務端返回的消息，先壓入隊列
?RpcResponse?response?=?JSONObject.parseObject(message,?RpcResponse.class);
?responses.offer(response);
?synchronized?(ApplicationContext.this)?{
?ApplicationContext.this.notifyAll();
?}
});設置一個callback，用于收到消息的時候，回調這里的代碼，這部分我們后面再分析。
然后在client.getCtx()的時候，同步阻塞直到連接完成,建立好連接后通過，NettyClient的代碼如下：
public?class?NettyClient?{
?private?ChannelHandlerContext?ctx;
?private?MessageCallback?messageCallback;
?public?NettyClient(String?ip,?Integer?port)?{
?EventLoopGroup?group?=?new?NioEventLoopGroup();
?try?{
?Bootstrap?b?=?new?Bootstrap();
?b.group(group)
?.channel(NioSocketChannel.class)
?.option(ChannelOption.TCP_NODELAY,?true)
?.handler(new?ChannelInitializer()?{
?@Override
?protected?void?initChannel(SocketChannel?ch)?throws?Exception?{
?ByteBuf?delimiter?=?Unpooled.copiedBuffer("$$".getBytes());
?//?設置按照分隔符“&&”來切分消息，單條消息限制為?1MB
?ch.pipeline().addLast(new?DelimiterBasedFrameDecoder(1024?*?1024,?delimiter));
?ch.pipeline().addLast(new?StringDecoder());
?ch.pipeline().addLast(new?NettyClientHandler());
?}
?});
?ChannelFuture?sync?=?b.connect(ip,?port).sync();
?}?catch?(Exception?e)?{
?e.printStackTrace();
?}
?}
?public?void?setMessageCallback(MessageCallback?callback)?{
?this.messageCallback?=?callback;
?}
?public?ChannelHandlerContext?getCtx()?throws?InterruptedException?{
?System.out.println("等待連接成功...");
?if?(ctx?==?null)?{
?synchronized?(this)?{
?wait();
?}
?}
?return?ctx;
?}
?private?class?NettyClientHandler?extends?ChannelInboundHandlerAdapter?{
?@Override
?public?void?channelRead(ChannelHandlerContext?ctx,?Object?msg)?throws?Exception?{
?try?{
?String?message?=?(String)?msg;
?if?(messageCallback?!=?null)?{
?messageCallback.onMessage(message);
?}
?}?finally?{
?ReferenceCountUtil.release(msg);
?}
?}
?@Override
?public?void?channelActive(ChannelHandlerContext?ctx)?throws?Exception?{
?NettyClient.this.ctx?=?ctx;
?System.out.println("連接成功："?+?ctx);
?synchronized?(NettyClient.this)?{
?NettyClient.this.notifyAll();
?}
?}
?@Override
?public?void?channelReadComplete(ChannelHandlerContext?ctx)?throws?Exception?{
?ctx.flush();
?}
?@Override
?public?void?exceptionCaught(ChannelHandlerContext?ctx,?Throwable?cause)?throws?Exception?{
?cause.printStackTrace();
?}
?}
?public?interface?MessageCallback?{
?void?onMessage(String?message);
?}
}這里主要是用了wait()和notifyAll()來實現同步阻塞等待連接建立。
建立好連接后，我們保存到集合中：
//?等待連接建立 ChannelHandlerContext?ctx?=?client.getCtx(); channels.put(info,?ctx);
發送請求
好了，到了這里我們為每一個需要消費的接口建立了網絡連接，接下來要做的事情就是提供一個接口給用戶獲取服務提供者實例：
我把這個方法寫在ApplicationContext中：
/**
?*?負責生成requestId的類
?*/
private?LongAdder?requestIdWorker?=?new?LongAdder();
/**
?*?獲取調用服務
?*/
@SuppressWarnings("unchecked")
publicT?getService(Classclazz)?{
?return?(T)?Proxy.newProxyInstance(getClass().getClassLoader(),?new?Class[]{clazz},?new?InvocationHandler()?{
?@Override
?public?Object?invoke(Object?proxy,?Method?method,?Object[]?args)?throws?Throwable?{
?String?methodName?=?method.getName();
?if?("equals".equals(methodName)?||?"hashCode".equals(methodName))?{
?throw?new?IllegalAccessException("不能訪問"?+?methodName?+?"方法");
?}
?if?("toString".equals(methodName))?{
?return?clazz.getName()?+?"#"?+?methodName;
?}
?//?step?1:?獲取服務地址列表
?ListregistryInfos?=?interfacesMethodRegistryList.get(clazz);
?if?(registryInfos?==?null)?{
?throw?new?RuntimeException("無法找到服務提供者");
?}
?//?step?2：?負載均衡
?RegistryInfo?registryInfo?=?loadBalancer.choose(registryInfos);
?ChannelHandlerContext?ctx?=?channels.get(registryInfo);
?String?identify?=?InvokeUtils.buildInterfaceMethodIdentify(clazz,?method);
?String?requestId;
?synchronized?(ApplicationContext.this)?{
?requestIdWorker.increment();
?requestId?=?String.valueOf(requestIdWorker.longValue());
?}
?Invoker?invoker?=?new?DefaultInvoker(method.getReturnType(),?ctx,?requestId,?identify);
?inProgressInvoker.put(identify?+?"#"?+?requestId,?invoker);
?return?invoker.invoke(args);
?}
?});
}這里主要是通過動態代理來實現的，首先通過class來獲取對應的機器列表，接著通過loadBalancer來選擇一個機器。這個LoaderBalance是一個接口：
public?interface?LoadBalancer?{
?/**
?*?選擇一個生產者
?*
?*?@param?registryInfos?生產者列表
?*?@return?選中的生產者
?*/
?RegistryInfo?choose(ListregistryInfos);
}在ApplicationContext初始化的時候可以選擇不同的實現，我這里主要實現了一個簡單的隨機算法（后續可以拓展為其他的，比如RoundRobin之類的）：
public?class?RandomLoadbalancer?implements?LoadBalancer?{
?@Override
?public?RegistryInfo?choose(ListregistryInfos)?{
?Random?random?=?new?Random();
?int?index?=?random.nextInt(registryInfos.size());
?return?registryInfos.get(index);
?}
}接著構造接口方法的唯一標識identify，還有一個requestId。
為什么需要一個requestId呢？
這是因為我們在處理響應的時候，需要找到某個響應是對應的哪個請求，但是僅僅使用identify是不行的
因為我們同一個應用程序中可能會有多個線程同時調用同一個接口的同一個方法，這樣的identify是相同的。
所以我們需要用 identify + requestId的方式來判斷，reqeustId是一個自增的LongAddr。服務端在響應的時候會將requestId返回。
接著我們構造了一個Invoker，把它放入inProgressInvoker的集合中。調用了其invoke方法：
Invoker?invoker?=?new?DefaultInvoker(method.getReturnType(),?ctx,?requestId,?identify);
inProgressInvoker.put(identify?+?"#"?+?requestId,?invoker);
//?阻塞等待結果
return?invoker.invoke(args);
public?class?DefaultInvokerimplements?Invoker{
?private?ChannelHandlerContext?ctx;
?private?String?requestId;
?private?String?identify;
?private?Cla***eturnType;
?private?T?result;
?DefaultInvoker(Cla***eturnType,?ChannelHandlerContext?ctx,?String?requestId,?String?identify)?{
?this.returnType?=?returnType;
?this.ctx?=?ctx;
?this.requestId?=?requestId;
?this.identify?=?identify;
?}
?@SuppressWarnings("unckecked")
?@Override
?public?T?invoke(Object[]?args)?{
?JSONObject?jsonObject?=?new?JSONObject();
?jsonObject.put("interfaces",?identify);
?JSONObject?param?=?new?JSONObject();
?if?(args?!=?null)?{
?for?(Object?obj?:?args)?{
?param.put(obj.getClass().getName(),?obj);
?}
?}
?jsonObject.put("parameter",?param);
?jsonObject.put("requestId",?requestId);
?System.out.println("發送給服務端JSON為："?+?jsonObject.toJSONString());
?String?msg?=?jsonObject.toJSONString()?+?"$$";
?ByteBuf?byteBuf?=?Unpooled.buffer(msg.getBytes().length);
?byteBuf.writeBytes(msg.getBytes());
?ctx.writeAndFlush(byteBuf);
?waitForResult();
?return?result;
?}
?@Override
?public?void?setResult(String?result)?{
?synchronized?(this)?{
?this.result?=?JSONObject.parseObject(result,?returnType);
?notifyAll();
?}
?}
?private?void?waitForResult()?{
?synchronized?(this)?{
?try?{
?wait();
?}?catch?(InterruptedException?e)?{
?e.printStackTrace();
?}
?}
?}
}我們可以看到調用Invoker的invoke方法之后，會運行到waitForResult()這里，這里已經把請求通過網絡發送出去了，但是就會被卡住。
這是因為我們的網絡請求的結果不是同步返回的，有可能是客戶端同時發起很多個請求，所以我們不可能在這里讓他同步阻塞等待的。
接受響應
那么對于服務消費者而言，把請求發送出去但是卡住了，這個時候當服務端處理完之后，會把消息返回給客戶端。返回的入口在
NettyClient的onChannelRead中：
@Override
public?void?channelRead(ChannelHandlerContext?ctx,?Object?msg)?throws?Exception?{
?try?{
?String?message?=?(String)?msg;
?if?(messageCallback?!=?null)?{
?messageCallback.onMessage(message);
?}
?}?finally?{
?ReferenceCountUtil.release(msg);
?}
}這里通過callback回調出去。是否還記的我們在初始化NettyClient的時候，會設置一個callback？
/**
?*?響應隊列
?*/
private?ConcurrentLinkedQueueresponses?=?new?ConcurrentLinkedQueue<>();
client.setMessageCallback(message?->?{
?//?這里收單服務端返回的消息，先壓入隊列
?RpcResponse?response?=?JSONObject.parseObject(message,?RpcResponse.class);
?responses.offer(response);
?synchronized?(ApplicationContext.this)?{
?ApplicationContext.this.notifyAll();
?}
});這里接受消息之后，解析成為一個RpcResponse對象，然后壓入responses隊列中，這樣我們就把所有的請求響應放入隊列中。
但是這樣一來，我們應該怎么把響應結果返回給調用的地方呢？
我們可以這樣做：起一個或多個后臺線程，然后從隊列中拿出響應，然后根據響應從我們之前保存的inProcessInvoker中找出對應的Invoker，然后把結果返回回去
public?ApplicationContext(....){
?
?//.....
?
?//?step?5：啟動處理響應的processor
?initProcessor();
?
}
private?void?initProcessor()?{
?//?事實上，這里可以通過配置文件讀取，啟動多少個processor
?int?num?=?3;
?processors?=?new?ResponseProcessor[num];
?for?(int?i?=?0;?i?<?3;?i++)?{
?processors[i]?=?createProcessor(i);
?}
}
/**
?*?處理響應的線程
?*/
private?class?ResponseProcessor?extends?Thread?{
?@Override
?public?void?run()?{
?System.out.println("啟動響應處理線程："?+?getName());
?while?(true)?{
?//?多個線程在這里獲取響應，只有一個成功
?RpcResponse?response?=?responses.poll();
?if?(response?==?null)?{
?try?{
?synchronized?(ApplicationContext.this)?{
?//?如果沒有響應，先休眠
?ApplicationContext.this.wait();
?}
?}?catch?(InterruptedException?e)?{
?e.printStackTrace();
?}
?}?else?{
?System.out.println("收到一個響應："?+?response);
?String?interfaceMethodIdentify?=?response.getInterfaceMethodIdentify();
?String?requestId?=?response.getRequestId();
?String?key?=?interfaceMethodIdentify?+?"#"?+?requestId;
?Invoker?invoker?=?inProgressInvoker.remove(key);
?invoker.setResult(response.getResult());
?}
?}
?}
}這里面如果從隊列中拿不到數據，就會調用wait()方法等待
這里需要注意的是，在callbak中獲取到響應的時候我們是會調用notifyAll()來喚醒這里的線程的：
responses.offer(response);
synchronized?(ApplicationContext.this)?{
?ApplicationContext.this.notifyAll();
}這里被喚醒之后，就會有多個線程去爭搶那個響應，因為隊列是線程安全的，所以這里多個線程可以獲取到響應結果。
接著拿到結果之后，通過identify + requestId構造成唯一的請求標識，從inProgressInvoker中獲取對應的invoker，然后通過setResult將結果設置進去：
String?key?=?interfaceMethodIdentify?+?"#"?+?requestId;
Invoker?invoker?=?inProgressInvoker.remove(key);
invoker.setResult(response.getResult());
@Override
public?void?setResult(String?result)?{
?synchronized?(this)?{
?this.result?=?JSONObject.parseObject(result,?returnType);
?notifyAll();
?}
}這里設置進去之后，就會將結果用json反序列化成為用戶需要的結果，然后調用其notifyAll方法喚醒invoke方法被阻塞的線程：
?@SuppressWarnings("unckecked")
?@Override
?public?T?invoke(Object[]?args)?{
?JSONObject?jsonObject?=?new?JSONObject();
?jsonObject.put("interfaces",?identify);
?JSONObject?param?=?new?JSONObject();
?if?(args?!=?null)?{
?for?(Object?obj?:?args)?{
?param.put(obj.getClass().getName(),?obj);
?}
?}
?jsonObject.put("parameter",?param);
?jsonObject.put("requestId",?requestId);
?System.out.println("發送給服務端JSON為："?+?jsonObject.toJSONString());
?String?msg?=?jsonObject.toJSONString()?+?NettyServer.DELIMITER;
?ByteBuf?byteBuf?=?Unpooled.buffer(msg.getBytes().length);
?byteBuf.writeBytes(msg.getBytes());
?ctx.writeAndFlush(byteBuf);
?//?這里被喚醒
?waitForResult();
?return?result;
?}然后就可以返回結果了，返回的結果就會返回給用戶了。
整體測試
到了這里我們的生產者和消費者的代碼都寫完了，我們來整體測試一遍。生產者的代碼是和之前的一致的：
public?class?TestProducer?{
?public?static?void?main(String[]?args)?throws?Exception?{
?String?connectionString?=?"zookeeper://localhost1:2181,localhost2:2182,localhost3:2181";
?HelloService?service?=?new?HelloServiceImpl();
?ServiceConfig?config?=?new?ServiceConfig<>(HelloService.class,?service);
?ListserviceConfigList?=?new?ArrayList<>();
?serviceConfigList.add(config);
?ApplicationContext?ctx?=?new?ApplicationContext(connectionString,?serviceConfigList,?null,?50071);
?}
}消費者測試代碼：
public?class?TestConsumer?{
?public?static?void?main(String[]?args)?throws?Exception?{
?String?connectionString?=?"zookeeper://localhost1:2181,localhost2:2182,localhost3:2181";
?ReferenceConfigconfig?=?new?ReferenceConfig<>(HelloService.class);
?ApplicationContext?ctx?=?new?ApplicationContext(connectionString,?null,?Collections.singletonList(config),
?50070);
?HelloService?helloService?=?ctx.getService(HelloService.class);
?System.out.println("sayHello(TestBean)結果為："?+?helloService.sayHello(new?TestBean("張三",?20)));
?}
}接著啟動生產者，然后啟動消費者：
生產者得到的日志如下：
Zookeeper?Client初始化完畢......
注冊到注冊中心，路徑為：【/myRPC/interface=com.study.rpc.test.producer.HelloService&
method=sayHello?meter=com.study.rpc.test.producer.TestBean】
信息為：RegistryInfo{hostname='localhost',?ip='192.168.16.7',?port=50071}
啟動NettyService，端口為：50071
啟動響應處理線程：Response-processor-0
啟動響應處理線程：Response-processor-2
啟動響應處理線程：Response-processor-1
接收到消息：{"interfaces":"interface=com.study.rpc.test.producer.HelloService&
method=sayHello?meter=com.study.rpc.test.producer.TestBean","requestId":"1",
"parameter":{"com.study.rpc.test.producer.TestBean":{"age":20,"name":"張三"}}}
響應給客戶端：{"interfaceMethodIdentify":"interface=com.study.rpc.test.producer.HelloService&
method=sayHello?meter=com.study.rpc.test.producer.TestBean","requestId":"1",
"result":"\"牛逼,我收到了消息：TestBean{name='張三',?age=20}\""}消費者得到的日志為：
Zookeeper Client初始化完畢......
開始建立連接：192.168.16.7, 50071
等待連接成功...
啟動響應處理線程：Response-processor-1
啟動響應處理線程：Response-processor-0
啟動響應處理線程：Response-processor-2
連接成功：ChannelHandlerContext(NettyClient$NettyClientHandler#0,
[id: 0xb7a59701, L:/192.168.16.7:58354 - R:/192.168.16.7:50071])
發送給服務端JSON為：{"interfaces":"interface=com.study.rpc.test.producer.HelloService&
method=sayHello?meter=com.study.rpc.test.producer.TestBean","requestId":"1",
"parameter":{"com.study.rpc.test.producer.TestBean":{"age":20,"name":"張三"}}}
收到一個響應：RpcResponse{result='"牛逼,我收到了消息：TestBean{name='張三', age=20}"',
interfaceMethodIdentify='interface=com.study.rpc.test.producer.HelloService&
method=sayHello?meter=com.study.rpc.test.producer.TestBean', requestId='1'}
sayHello(TestBean)結果為：牛逼,我收到了消息：TestBean{name='張三', age=20}
總結
通過完成這個RPC框架，大家應該會大致對RPC的實現原理有個感性的認識，這里總結一下特性：
支持多種注冊中心，可配置(雖然只實現了zookeeper，但是我們拓展是非常簡單的)
支持負載均衡
當然了還有非常多的不足之處，這是無可否認的，隨意寫出來的框架和工業級使用的框架比較還是不一樣
我這里列舉一些不完美的地方把(有興趣的可以搞搞)：
實現序列化框架的拓展，多種序列化供用戶選擇
網絡請求錯誤處理，這里實現非常簡陋，健壯性很差
注冊中心不支持故障感知和自動恢復
調用監控，性能指標
歡迎工作一到五年的Java工程師朋友們加入我的個人粉絲群（Java架構技術棧：728987924）群內提供免費的Java架構學習資料（里面有高可用、高并發、高性能及分布式、Jvm性能調優、Spring源碼，MyBatis，Netty,Redis,Kafka,Mysql,Zookeeper,Tomcat,Docker,Dubbo,Nginx等多個知識點的架構資料）
合理利用自己每一分每一秒的時間來學習提升自己，不要再用"沒有時間“來掩飾自己思想上的懶惰！趁年輕，使勁拼，給未來的自己一個交代！
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
