摘要:項目版本源碼在上一博文中跟大家講了的實現思路思路畢竟只是思路那么這篇就帶著源碼給大家講解下實現過程中的各個具體問題讀懂本篇需要的基本知識若尚未清晰請自行了解后再閱讀本文動態代理框架的基本使用的基本配置最終項目的使用如下調用端代碼及配置測試類
項目1.0版本源碼
https://github.com/wephone/Me...
在上一博文中 跟大家講了RPC的實現思路 思路畢竟只是思路 那么這篇就帶著源碼給大家講解下實現過程中的各個具體問題
讀懂本篇需要的基本知識 若尚未清晰請自行了解后再閱讀本文java動態代理
netty框架的基本使用
spring的基本配置
最終項目的使用如下/** *調用端代碼及spring配置 */ @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration(locations={"file:src/test/java/rpcTest/ClientContext.xml"}) public class Client { @Test public void start(){ Service service= (Service) RPC.call(Service.class); System.out.println("測試Integer,Double類型傳參與返回String對象:"+service.stringMethodIntegerArgsTest(233,666.66)); //輸出string233666.66 } } /** *Service抽象及其實現 *調用與實現端共同依賴Service */ public interface Service { String stringMethodIntegerArgsTest(Integer a,Double b); } /** * ServiceImpl實現端對接口的具體實現 */ public class ServiceImpl implements Service { @Override public String stringMethodIntegerArgsTest(Integer a, Double b) { return "String"+a+b; } }
1.0版本分3個包
Client 調用端
Server 實現端
Core 核心方法
首先看這句代碼調用端只需如此調用
定義接口 傳入接口類類型 后面調用的接口內的方法 全部是由實現端實現
Service service= (Service) RPC.call(Service.class);
這句的作用其實就是生成調用端的動態代理
/** * 暴露調用端使用的靜態方法 為抽象接口生成動態代理對象 * TODO 考慮后面優化不在使用時仍需強轉 * @param cls 抽象接口的類類型 * @return 接口生成的動態代理對象 */ public static Object call(Class cls){ RPCProxyHandler handler=new RPCProxyHandler(); Object proxyObj=Proxy.newProxyInstance(cls.getClassLoader(),new Class>[]{cls},handler); return proxyObj; }
RPCProxyHandler為動態代理的方法被調用后的回調方法 每個方法被調用時都會執行這個invoke
/** * 代理抽象接口調用的方法 * 發送方法信息給服務端 加鎖等待服務端返回 * @param proxy * @param method * @param args * @return * @throws Throwable */ @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { RPCRequest request=new RPCRequest(); request.setRequestID(buildRequestID(method.getName())); request.setClassName(method.getDeclaringClass().getName());//返回表示聲明由此 Method 對象表示的方法的類或接口的Class對象 request.setMethodName(method.getName()); // request.setParameterTypes(method.getParameterTypes());//返回形參類型 request.setParameters(args);//輸入的實參 RPCRequestNet.requestLockMap.put(request.getRequestID(),request); RPCRequestNet.connect().send(request); //調用用結束后移除對應的condition映射關系 RPCRequestNet.requestLockMap.remove(request.getRequestID()); return request.getResult();//目標方法的返回結果 }
也就是收集對應調用的接口的信息 然后send給實現端
那么這個requestLockMap又是作何作用的呢
由于我們的網絡調用都是異步的
但是RPC調用都要做到同步 等待這個遠程調用方法完全返回后再繼續執行
所以將每個請求的request對象作為對象鎖 每個請求發送后加鎖 等到網絡異步調用返回后再釋放所
生成每個請求的ID 這里我用隨機數加時間戳
將請求ID和請求對象維護在靜態全局的一個map中 實現端通過ID來對應是哪個請求
異步調用返回后 通過ID notify喚醒對應請求對象的線程
netty異步返回的調用 釋放對象鎖
@Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { String responseJson= (String) msg; RPCResponse response= (RPCResponse) RPC.responseDecode(responseJson); synchronized (RPCRequestNet.requestLockMap.get(response.getRequestID())) { //喚醒在該對象鎖上wait的線程 RPCRequest request= (RPCRequest) RPCRequestNet.requestLockMap.get(response.getRequestID()); request.setResult(response.getResult()); request.notifyAll(); } }
接下來是RPCRequestNet.connect().send(request);方法
connect方法其實是單例模式返回RPCRequestNet實例
RPCRequestNet構造方法是使用netty對實現端進行TCP鏈接
send方法如下
try { //判斷連接是否已完成 只在連接啟動時會產生阻塞 if (RPCRequestHandler.channelCtx==null){ connectlock.lock(); //掛起等待連接成功 System.out.println("正在等待連接實現端"); connectCondition.await(); connectlock.unlock(); } //編解碼對象為json 發送請求 String requestJson= null; try { requestJson = RPC.requestEncode(request); } catch (JsonProcessingException e) { e.printStackTrace(); } ByteBuf requestBuf= Unpooled.copiedBuffer(requestJson.getBytes()); RPCRequestHandler.channelCtx.writeAndFlush(requestBuf); System.out.println("調用"+request.getRequestID()+"已發送"); //掛起等待實現端處理完畢返回 TODO 后續配置超時時間 synchronized (request) { //放棄對象鎖 并阻塞等待notify request.wait(); } System.out.println("調用"+request.getRequestID()+"接收完畢"); } catch (InterruptedException e) { e.printStackTrace(); }
condition和lock同樣是為了同步等待異步IO返回用的
send方法基本是編解碼json后發送給實現端
/** *實現端代碼及spring配置 */ @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration(locations={"file:src/test/java/rpcTest/ServerContext.xml"}) public class Server { @Test public void start(){ //啟動spring后才可啟動 防止容器尚未加載完畢 RPC.start(); } }
出了配置spring之外 實現端就一句 RPC.start()
其實就是啟動netty服務器
服務端的處理客戶端信息回調如下
@Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws IOException { String requestJson= (String) msg; System.out.println("receive request:"+requestJson); RPCRequest request= RPC.requestDeocde(requestJson); Object result=InvokeServiceUtil.invoke(request); //netty的write方法并沒有直接寫入通道(為避免多次喚醒多路復用選擇器) //而是把待發送的消息放到緩沖數組中,flush方法再全部寫到通道中 // ctx.write(resp); //記得加分隔符 不然客戶端一直不會處理 RPCResponse response=new RPCResponse(); response.setRequestID(request.getRequestID()); response.setResult(result); String respStr=RPC.responseEncode(response); ByteBuf responseBuf= Unpooled.copiedBuffer(respStr.getBytes()); ctx.writeAndFlush(responseBuf); }
主要是編解碼json 反射對應的方法 我們看看反射的工具類
/** * 反射調用相應實現類并結果 * @param request * @return */ public static Object invoke(RPCRequest request){ Object result=null;//內部變量必須賦值 全局變量才不用 //實現類名 String implClassName= RPC.getServerConfig().getServerImplMap().get(request.getClassName()); try { Class implClass=Class.forName(implClassName); Object[] parameters=request.getParameters(); int parameterNums=request.getParameters().length; Class[] parameterTypes=new Class[parameterNums]; for (int i = 0; i解析Parameters getClass獲取他們的類類型 反射調用對應的方法
這里需要注意一個點本文最初采用Gson處理json 但gson默認會把int類型轉為double類型 例如2變為2.0 不適用本場景 我也不想去專門適配
所以換用了jackson
常見json處理框架 反序列化為對象時 int,long等基本類型都會變成他們的包裝類Integer Long
所以本例程中 遠程調度接口方法的形參不可以使用int等基本類型
否則method.invoke(implObj,parameters);會找不到對應的方法報錯
因為parameters已經是包裝類了 而method還是int這些基本類 所以找不到對應方法
最后是借助spring配置基礎配置
我寫了兩個類 ServerConfig ClientConfig 作為調用端和服務端的配置
只需在spring中配置這兩個bean 并啟動IOC容器即可調用端
實現端
最后有個小問題我們的框架是作為一個依賴包引入的 我們不可能在我們的框架中讀取對應的spring xml
這樣完全是去了框架的靈活性
那我們怎么在運行過程中獲得我們所處于的IOC容器 已獲得我們的正確配置信息呢
答案是spring提供的ApplicationContextAware接口/** * Created by wephone on 17-12-26. */ public class ClientConfig implements ApplicationContextAware { private String host; private int port; //調用超時時間 private long overtime; public String getHost() { return host; } public void setHost(String host) { this.host = host; } public int getPort() { return port; } public void setPort(int port) { this.port = port; } public long getOvertime() { return overtime; } public void setOvertime(long overtime) { this.overtime = overtime; } /** * 加載Spring配置文件時,如果Spring配置文件中所定義的Bean類 * 如果該類實現了ApplicationContextAware接口 * 那么在加載Spring配置文件時,會自動調用ApplicationContextAware接口中的 * @param applicationContext * @throws BeansException */ @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { RPC.clientContext=applicationContext; } }這樣我們在RPC類內部就維護了一個靜態IOC容器的context
只需如此獲取配置
RPC.getServerConfig().getPort()public static ServerConfig getServerConfig(){ return serverContext.getBean(ServerConfig.class); }就這樣 這個RPC框架的核心部分 已經講述完畢了本例程僅為1.0版本
后續博客中 會加入異常處理 zookeeper支持 負載均衡策略等
博客:zookeeper支持
歡迎持續關注 歡迎star 提issue
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/68280.html
摘要:每個都可以通過其路徑唯一標識,同時每個節點還可以存儲少量數據。監聽機制,監聽某個當該發生變化時,會回調該,但是這個是一次性的,下次需要監聽時還得再注冊一次。 前面的文章中 我用netty實現了一個簡單的一對一的RPC 11個類實現簡單java rpc 接下來的文章中 我將使用zookeeper作為rpc調用的分布式注冊中心 從而實現多對多(多個調用者,多個提供者)的rpc調用,負載均...
摘要:等之所以支持跨語言,是因為他們自己定義了一套結構化數據存儲格式,如的,用于編解碼對象,作為各個語言通信的中間協議。 前段時間覺得自己一直用別人的框架,站在巨人的肩膀上,也該自己造造輪子了 一時興起 就著手寫起了RPC框架 這里寫了系列博客拿給大家分享下 這篇是開篇的思路篇 項目最終的代碼放在了我的github上https://github.com/wephone/Me... 歡迎sta...
摘要:支持相關規范和標準,包括同上。支持多種傳輸協議和協議綁定數據綁定。構建端還有其服務實現,接口使用注解,標明是一個遠程服務接口。然后編寫一個的啟動程序,并運行,我想你會成功的因為我看到了下圖是一種跨平臺的技術協議。 本博客 貓叔的博客,轉載請申明出處 學習系列 RPC框架是啥? RPC框架是啥之Java自帶RPC實現,RMI框架入門 Apache CXF一款WebService RP...
閱讀 1589·2023-04-26 01:54
閱讀 1621·2021-09-30 09:55
閱讀 2645·2021-09-22 16:05
閱讀 1856·2021-07-25 21:37
閱讀 2620·2019-08-29 18:45
閱讀 1886·2019-08-29 16:44
閱讀 1882·2019-08-29 12:34
閱讀 1346·2019-08-23 14:02