一、RPC簡介

最近看hadoop底層通信，都是通過RPC實現的。

RPC(Remote Procedure Call Protocol)遠程調用： 遠程過程調用是一種常用的分布式網絡通信協議,它允許運行于 一臺計算機的程序調用另一臺計算機的子程序，同時將網絡的通信細節隱藏起來， 使得用戶無須額外地為這個交互作用編程。分布式系統之間的通信大都通過RPC實現

二、RPC請求過程

client發起服務調用請求

client stub代理程序將調用的方法，參數按照一定格式封裝，通過服務方的地址，發起網絡請求

消息通過網絡發送到服務端，server stub接收到消息，進行解包，反射調用本地對應的服務

本地服務執行將結果返回給server stub,然后server stub會將結果消息打包返回到客戶端

client stub接收消息解碼，得到最終結果.

三、RPC框架架構

要寫一個RPC框架，需要哪些組成部分？

序列化方式。序列化主要作用是將結構化對象轉為字節流以便于通過網絡進行傳輸或寫入持久存儲。

遠程代理對象，一般使用jdk動態代理或者cglib代理

服務暴露 設置注冊中心Zookeeper

網絡通信，基于事件驅動的Reactor模式

四、RPC框架示例

服務提供者，運行在服務器端，提供服務接口定義與服務實現類

服務發布者，運行在服務器端，負責將本地服務發布成遠程服務，管理遠程服務，提供給服務消費者使用

服務消費者，運行在客戶端，通過遠程代理對象調用遠程服務

服務端代碼

服務接口：

//計算學生年齡和的接口public interface CalculateService { String cal(Student sta, Student stb);}public class CalculateServiceImpl implements CalculateService { @Override public String cal(Student sta, Student stb) { return "學生年齡之和：" + (sta.getAge() + stb.getAge()); }}

服務發布

public class PublishUtilI { //服務接口集合 private static List serviceList; private static ThreadPoolExecutor executor = new ThreadPoolExecutor(5,10,10, TimeUnit.SECONDS, new LinkedBlockingQueue(10)); public static void publish(int port,Object... services) throws IOException { serviceList= Arrays.asList(services); ServerSocket server = new ServerSocket(port); Socket client; while (true) { //阻塞等待請求 client = server.accept(); //使用線程池處理請求 executor.submit(new ServerHandler(client, serviceList)); } }}

反射調用服務

讀取客戶端發送的服務名

判斷服務是否發布

如果發布，反射調用服務端對應服務

返回結果給客戶端

public class ServerHandler implements Runnable { private Socket client = null; private List serviceList = null; public ServerHandler(Socket client, List service) { this.client = client; this.serviceList = service; } @Override public void run() { try ( ObjectInputStream input = new ObjectInputStream(client.getInputStream()); ObjectOutputStream output = new ObjectOutputStream(client.getOutputStream()) ) { // 讀取客戶端要訪問那個service Class serviceClass = (Class) input.readObject(); // 找到該服務類 Object obj = findService(serviceClass); if (obj == null) { output.writeObject(serviceClass.getName() + "服務未發現"); } else { //利用反射調用該方法，返回結果 String methodName = input.readUTF(); //讀取UTF編碼的String字符串 //讀取參數類型 Class>[] parameterTypes = (Class>[]) input.readObject(); //讀取參數 Object[] arguments = (Object[]) input.readObject(); Method method = obj.getClass().getMethod(methodName, parameterTypes); //反射執行方法 Object result = method.invoke(obj, arguments); output.writeObject(result); } } catch (Exception e) { e.printStackTrace(); } } private Object findService(Class serviceClass) { for (Object obj : serviceList) { boolean isFather = serviceClass.isAssignableFrom(obj.getClass()); if (isFather) { return obj; } } return null; }}

客戶端代碼

public class Client { public static void main(String[] args) { CallProxyHandler handler = new CallProxyHandler("127.0.0.1

總結

以上是生活随笔為你收集整理的调用线程必须为sta_Java手写分布式系统远程调用RPC框架的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。