本篇內(nèi)容主要講解“java rpc框架的原理是什么”，感興趣的朋友不妨來(lái)看看。本文介紹的方法操作簡(jiǎn)單快捷，實(shí)用性強(qiáng)。下面就讓小編來(lái)帶大家學(xué)習(xí)“java rpc框架的原理是什么”吧!

公司主營(yíng)業(yè)務(wù)：網(wǎng)站建設(shè)、成都網(wǎng)站制作、移動(dòng)網(wǎng)站開(kāi)發(fā)等業(yè)務(wù)。幫助企業(yè)客戶真正實(shí)現(xiàn)互聯(lián)網(wǎng)宣傳，提高企業(yè)的競(jìng)爭(zhēng)能力。創(chuàng)新互聯(lián)公司是一支青春激揚(yáng)、勤奮敬業(yè)、活力青春激揚(yáng)、勤奮敬業(yè)、活力澎湃、和諧高效的團(tuán)隊(duì)。公司秉承以“開(kāi)放、自由、嚴(yán)謹(jǐn)、自律”為核心的企業(yè)文化，感謝他們對(duì)我們的高要求，感謝他們從不同領(lǐng)域給我們帶來(lái)的挑戰(zhàn)，讓我們激情的團(tuán)隊(duì)有機(jī)會(huì)用頭腦與智慧不斷的給客戶帶來(lái)驚喜。創(chuàng)新互聯(lián)公司推出雨花臺(tái)免費(fèi)做網(wǎng)站回饋大家。

RPC基礎(chǔ)概念

     RPC就是遠(yuǎn)程過(guò)程調(diào)用協(xié)議，其作用就是客戶端與服務(wù)端之間的遠(yuǎn)程調(diào)用，就像本地自己調(diào)用一樣，讓服務(wù)端進(jìn)行服務(wù)化，功能唯一性，負(fù)載熱點(diǎn)流量。

RPC框架的實(shí)現(xiàn)

    RPC能夠讓本地應(yīng)用簡(jiǎn)單、高效地調(diào)用服務(wù)器中的過(guò)程（服務(wù)）。它主要應(yīng)用在分布式系統(tǒng)。如Hadoop中的IPC組件。但怎樣實(shí)現(xiàn)一個(gè)RPC框架呢？

從下面幾個(gè)方面思考，僅供參考：

1.通信模型：假設(shè)通信的為A機(jī)器與B機(jī)器，A與B之間有通信模型，在Java中一般基于BIO或NIO；。

2.過(guò)程（服務(wù)）定位：使用給定的通信方式，與確定IP與端口及方法名稱(chēng)確定具體的過(guò)程或方法；

3.遠(yuǎn)程代理對(duì)象：本地調(diào)用的方法(服務(wù))其實(shí)是遠(yuǎn)程方法的本地代理，因此可能需要一個(gè)遠(yuǎn)程代理對(duì)象，對(duì)于Java而言，遠(yuǎn)程代理對(duì)象可以使用Java的動(dòng)態(tài)對(duì)象實(shí)現(xiàn)，封裝了調(diào)用遠(yuǎn)程方法調(diào)用；

4.序列化，將對(duì)象名稱(chēng)、方法名稱(chēng)、參數(shù)等對(duì)象信息進(jìn)行網(wǎng)絡(luò)傳輸需要轉(zhuǎn)換成二進(jìn)制傳輸，這里可能需要不同的序列化技術(shù)方案。如:thrift,protobuf,Arvo等。

下面用簡(jiǎn)單的原生java socket來(lái)實(shí)現(xiàn)rpc調(diào)用，方便大家更深層了解rpc原理

服務(wù)類(lèi)接口

package socketrpc;
2 
3 public interface IHello {
4      String sayHello(String string);
5 }

服務(wù)實(shí)現(xiàn)類(lèi)

1 package socketrpc.server;
 2 
 3 import socketrpc.IHello;
 4 
 5 public class HelloServiceImpl implements IHello {
 6 
 7     @Override
 8     public String sayHello(String string) {
 9         return "你好:".concat ( string );
10     }
11 }

客戶端實(shí)現(xiàn)

package socketrpc.client;

import socketrpc.IHello;

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.net.Socket;

public class RpcClientProxy implements InvocationHandler {

    private  Class serviceInterface;
    private InetSocketAddress addr;

    public RpcClientProxy(Class serviceInterface, String ip,String port) {
        this.serviceInterface = serviceInterface;
        this.addr = new InetSocketAddress(ip, Integer.parseInt ( port ));
    }

    public T getClientIntance(){
        return (T) Proxy.newProxyInstance (serviceInterface.getClassLoader(),new Class[]{serviceInterface},this);
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

        Socket socket = null;
        ObjectOutputStream output = null;
        ObjectInputStream input = null;

        try {
            // 2.創(chuàng)建Socket客戶端，根據(jù)指定地址連接遠(yuǎn)程服務(wù)提供者
            socket = new Socket();
            socket.connect(addr);

            // 3.將遠(yuǎn)程服務(wù)調(diào)用所需的接口類(lèi)、方法名、參數(shù)列表等編碼后發(fā)送給服務(wù)提供者
            output = new ObjectOutputStream(socket.getOutputStream());
            output.writeUTF(serviceInterface.getName());
            output.writeUTF(method.getName());
            output.writeObject(method.getParameterTypes());
            output.writeObject(args);

            // 4.同步阻塞等待服務(wù)器返回應(yīng)答，獲取應(yīng)答后返回
            input = new ObjectInputStream(socket.getInputStream());
            return input.readObject();
        } finally {
            if (socket != null) socket.close();
            if (output != null) output.close();
            if (input != null) input.close();
        }
    }

    public static void main(String[] args) {
        RpcClientProxy client = new RpcClientProxy<>(IHello.class,"localhost","6666");
        IHello hello = (IHello) client.getClientIntance ();
        System.out.println (hello.sayHello ( "socket rpc" ));
    }
}

服務(wù)端實(shí)現(xiàn)

package socketrpc.server;

import socketrpc.IHello;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashMap;

public class RpcServer {

    private static final HashMap> serviceRegistry = new HashMap<>();
    private  int port;

    public RpcServer(int port) {
        this.port =port;
    }

    public RpcServer register(Class serviceInterface, Class impl) {
        serviceRegistry.put(serviceInterface.getName(), impl);
        return this;
    }

    public void run() throws IOException {

        ServerSocket server = new ServerSocket();
        server.bind(new InetSocketAddress (port));
        System.out.println("start server");
        ObjectInputStream input =null;
        ObjectOutputStream output =null;
        Socket socket=null;
        try {
            while(true){
                socket = server.accept ();
                input =new ObjectInputStream(socket.getInputStream());
                String serviceName = input.readUTF();
                String methodName = input.readUTF();
                System.out.println (methodName);
                Class[] parameterTypes = (Class[]) input.readObject();
                Object[] arguments = (Object[]) input.readObject();
                Class serviceClass = serviceRegistry.get(serviceName);
                if (serviceClass == null) {
                    throw new ClassNotFoundException(serviceName + " not found");
                }
                Method method = serviceClass.getMethod(methodName, parameterTypes);
                Object result = method.invoke(serviceClass.newInstance(), arguments);
                output = new ObjectOutputStream (socket.getOutputStream());
                output.writeObject(result);
            }
        } catch (Exception e){
            e.printStackTrace();
        }finally {
            if (output != null) {
                try {
                    output.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (input != null) {
                try {
                    input.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (socket != null) {
                try {
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

    }

    public static void main(String[] args) throws IOException {
        new RpcServer ( 6666 ).register ( IHello.class,HelloServiceImpl.class).run ();
    }

}

啟動(dòng)服務(wù)端和客戶端后運(yùn)行結(jié)果如下

客戶端代碼簡(jiǎn)單講解

private  Class serviceInterface;
 2     private InetSocketAddress addr;
 3 
 4     public RpcClientProxy(Class serviceInterface, String ip,String port) {
 5         this.serviceInterface = serviceInterface;
 6         this.addr = new InetSocketAddress(ip, Integer.parseInt ( port ));
 7     }
 8 
 9     public T getClientIntance(){
10         return (T) Proxy.newProxyInstance (serviceInterface.getClassLoader(),new Class[]{serviceInterface},this);
11     }

傳入接口類(lèi)和ip端口,調(diào)用getClientIntance方法時(shí),對(duì)當(dāng)前接口進(jìn)行代理,實(shí)際調(diào)用方法為

1 @Override
 2     public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
 3 
 4         Socket socket = null;
 5         ObjectOutputStream output = null;
 6         ObjectInputStream input = null;
 7 
 8         try {
 9             // 2.創(chuàng)建Socket客戶端，根據(jù)指定地址連接遠(yuǎn)程服務(wù)提供者
10             socket = new Socket();
11             socket.connect(addr);
12 
13             // 3.將遠(yuǎn)程服務(wù)調(diào)用所需的接口類(lèi)、方法名、參數(shù)列表等編碼后發(fā)送給服務(wù)提供者
14             output = new ObjectOutputStream(socket.getOutputStream());
15             output.writeUTF(serviceInterface.getName());
16             output.writeUTF(method.getName());
17             output.writeObject(method.getParameterTypes());
18             output.writeObject(args);
19 
20             // 4.同步阻塞等待服務(wù)器返回應(yīng)答，獲取應(yīng)答后返回
21             input = new ObjectInputStream(socket.getInputStream());
22             return input.readObject();
23         } finally {
24             if (socket != null) socket.close();
25             if (output != null) output.close();
26             if (input != null) input.close();
27         }
28     }

遵循下面步驟

1：創(chuàng)建socket，并和遠(yuǎn)程進(jìn)行三次連接握手【socket.connect(addr)】。

2：封裝socket輸出流【ObjectOutputStream】。

3：輸出類(lèi)名稱(chēng)，方法名稱(chēng)，參數(shù)類(lèi)型和參數(shù)值給server。

4：獲取socket輸入流，等待server返回結(jié)果。

服務(wù)端代碼簡(jiǎn)單講解

1 public RpcServer(int port) {
2         this.port =port;
3     }
4 
5     public RpcServer register(Class serviceInterface, Class impl) {
6         serviceRegistry.put(serviceInterface.getName(), impl);
7         return this;
8     }

初始化服務(wù)端，將服務(wù)類(lèi)注冊(cè)到hashMap【模擬spring上下文】

1 public void run() throws IOException {
 2 
 3         ServerSocket server = new ServerSocket();
 4         server.bind(new InetSocketAddress (port));
 5         System.out.println("start server");
 6         ObjectInputStream input =null;
 7         ObjectOutputStream output =null;
 8         Socket socket=null;
 9         try {
10             while(true){
11                 socket = server.accept ();
12                 input =new ObjectInputStream(socket.getInputStream());
13 
14                 String serviceName = input.readUTF();
15                 String methodName = input.readUTF();
16                 System.out.println (methodName);
17                 Class[] parameterTypes = (Class[]) input.readObject();
18                 Object[] arguments = (Object[]) input.readObject();
19                 Class serviceClass = serviceRegistry.get(serviceName);
20                 if (serviceClass == null) {
21                     throw new ClassNotFoundException(serviceName + " not found");
22                 }
23                 Method method = serviceClass.getMethod(methodName, parameterTypes);
24                 Object result = method.invoke(serviceClass.newInstance(), arguments);
25                 output = new ObjectOutputStream (socket.getOutputStream());
26                 output.writeObject(result);
27             }
28         } catch (Exception e){
29             e.printStackTrace();
30         }finally {
31             if (output != null) {
32                 try {
33                     output.close();
34                 } catch (IOException e) {
35                     e.printStackTrace();
36                 }
37             }
38             if (input != null) {
39                 try {
40                     input.close();
41                 } catch (IOException e) {
42                     e.printStackTrace();
43                 }
44             }
45             if (socket != null) {
46                 try {
47                     socket.close();
48                 } catch (IOException e) {
49                     e.printStackTrace();
50                 }
51             }
52         }
53 
54     }

服務(wù)端執(zhí)行做了以下幾件事：

1：綁定端口，阻塞等待客戶端調(diào)用【socket = server.accept ()】。

2：封裝輸入流【socket.getInputStream()】。

3：從輸入流中獲取到接口名，方法名，參數(shù)類(lèi)型，參數(shù)值。

4：找到初始化時(shí)hashmap中的服務(wù)類(lèi)。

5：反射獲取服務(wù)實(shí)現(xiàn)類(lèi)方法并根據(jù)請(qǐng)求參數(shù)進(jìn)行服務(wù)調(diào)用。

6：封裝輸出流【ObjectOutputStream】，并且返回結(jié)果。

到目前為止，整個(gè)簡(jiǎn)單的socket實(shí)現(xiàn)的RPC服務(wù)就已經(jīng)全部完成了，可以優(yōu)化的部分。

1：序列化局限，原生序列化只能序列化實(shí)現(xiàn)了【Serializable】接口的服務(wù)類(lèi)，并且序列化復(fù)雜對(duì)象時(shí)，內(nèi)容龐大效率極低，需要高效的序列化協(xié)議進(jìn)行序列化參數(shù)方法等必要請(qǐng)求入?yún)?/p>

2：BIO性能局限，socket服務(wù)端采用默認(rèn)的BIO來(lái)阻塞獲取輸入流，效率低下，需采用NIO等異步非阻塞服務(wù)端方案，例如netty,mina和java nio等。

3：在大型企業(yè)級(jí)RPC解決方案中，客戶端和服務(wù)端的長(zhǎng)連接需要一直保持，否則每次調(diào)用時(shí)都要重新進(jìn)行三次握手和四次揮手，這樣頻繁的創(chuàng)建tcp連接對(duì)機(jī)器性能是極大的損耗，對(duì)socket的連接可以采用apache pool2連接池等方案

4：服務(wù)端負(fù)載，需要考慮服務(wù)自動(dòng)發(fā)現(xiàn)，讓客戶端在不需要重啟的情況下能動(dòng)態(tài)感知服務(wù)端的變化，從而實(shí)現(xiàn)熱部署等?？梢圆捎棉k法定時(shí)自動(dòng)輪詢，zookeeper等。

5：服務(wù)端服務(wù)類(lèi)執(zhí)行異常，客戶端感知等。

到此，相信大家對(duì)“java rpc框架的原理是什么”有了更深的了解，不妨來(lái)實(shí)際操作一番吧！這里是創(chuàng)新互聯(lián)網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！