簡單實現Java的RMI——遠程方法調用

一、RMI簡介：

說到RMI就不得不說RPC了。

RPC：（Remote Procedure Call），遠程過程調用。

RMI（Remote Method Invocation），遠程方法調用。

RPC和RMI是有區別的，RPC中是通過網絡服務協議向遠程主機發送請求，RPC遠程主機就去搜索與之相匹配的類和方法，找到后就執行方法並把結果編碼，通過網絡協議發回。

而RMI是通過客戶端的對象作為遠程接口進行遠程方法的調用。RMI只適用於Java語言。

二、RMI的運行機理：

涉及兩個網絡端。其核心思想是，一個端可以通過調用另一個端的方法，實現相關功能。
一個端“執行”一個方法，而這個方法的實際執行是在另一端進行的！

當然，兩個端都應該有相同的類，自然會擁有相同的方法。
一個端所謂的執行這個方法，其實是通過調用這個類的代理對象的方法，在其中攔截這個方法，在這個方法中
實際上是將執行這個方法的參數和類名稱、方法名稱，通過網絡通訊傳輸給另一端；另一端根據得到的方法名稱、
類名稱和參數，實際執行那個方法，再將方法執行結果回傳給對端。
要注意的問題：
1、實際執行方法的一端，我們可以認為是RMI服務器端，偽執行一端，自然是RMI客戶端；
2、偽執行端不應該自己完成參數、方法名稱和類名稱的傳遞工作；也就是說，對於RMI客戶端用戶而言，他只面對一個類的一個方法，
    直接執行就好；
3、RMI服務器端可能接收多個RMI客戶端有關這個方法的執行請求，每個客戶端的執行當然應該是獨立的，應該用線程實現；
4、RMI服務器端在執行了相關方法，並回傳方法執行結果后，應該斷開與RMI客戶端的連接。

下面是我要實現它的一個思路。

 

上圖由於截圖原因，給點補充說明：RpcClientExecutor的作用是建立和服務器的連接，並接受消息和發送消息，具體是發送方法的序列號和參數類型。

1.首先：應該是RpcBeanDefinition:

package com.xupt.rpc.core;

import java.lang.reflect.Method;

public class RpcBeanDefination {   //將類，類方法，類對象封裝在Definition中。

    private Class<?> klass;
    private Method method;
    private Object object;
    
    RpcBeanDefination() {
    }

給該類所有的成員都有getter和setter方法就不需要說了，這個類，將執行的哪個類的哪個方法和類的對象封裝起來，以后這個類將形成Map

的值，下面來介紹的RpcBeanFactory會重點介紹。

2.RpcBeanFactory

 

package com.xupt.rpc.core;

import java.util.HashMap;
import java.util.Map;

public class RpcBeanFactory {

    private final Map<String, RpcBeanDefination> beanMap;
    
    RpcBeanFactory() {
        beanMap = new HashMap<>();
    }
    
    void rpcBeanRegistry(String beanId,RpcBeanDefination defination) {
        RpcBeanDefination rbd = beanMap.get(beanId);
        if(rbd != null) {
            return;
        }
        beanMap.put(beanId, defination);
    }
    
    RpcBeanDefination getBean(String beanId) {
        return beanMap.get(beanId);
    }
}

 

此類是將序列號作為Map中的鍵，RpcBeanDeifintion作為值放入Map中，用BeanId來找對應客戶端那邊序列號相同的方法。

3.下來是RpcBeanRegistry:

package com.xupt.rpc.core;

import java.lang.reflect.Method;

public class RpcBeanRegistry {
    
    RpcBeanRegistry() {
    }
    
    //給客戶端提供
     static void registryInterface(RpcBeanFactory rpcBeanFactory,Class<?> interfaces) {
        doregistry(rpcBeanFactory,interfaces,null);
    }
     
    //內部使用，注冊
     private static void doregistry(RpcBeanFactory rpcBeanFactory , Class<?> interfaces ,Object object) {
          //得到接口中的所有的方法，行程方法的數組
         Method[] methods = interfaces.getDeclaredMethods();
            for(Method method : methods) {  //遍歷這些方法
                String beanId = String.valueOf(method.toString().hashCode());//將方法序列化。
                RpcBeanDefination rpcBeanDefination = new RpcBeanDefination();
                
                //將得到的實現接口的那個類和它的方法以及對象放進RpcBeanDefination()中。
                rpcBeanDefination.setKlass(interfaces);
                rpcBeanDefination.setMethod(method);
                rpcBeanDefination.setObject(object);
                
                rpcBeanFactory.rpcBeanRegistry(beanId, rpcBeanDefination);
            }
    }
    
     //服務端使用，知道實現類的對象。
     static void registryInterface(RpcBeanFactory rpcBeanFactory,Class<?> interfaces,Object object) {
         //判斷此類是否實現了這個接口。
        if(!interfaces.isAssignableFrom(object.getClass())){
            return;
        }
        doregistry(rpcBeanFactory,interfaces,object);
    }
    
     //服務器端使用，知道類，創建一個對象。
    static void registryInterface(RpcBeanFactory rpcBeanFactory,Class<?> interfaces,Class<?> klass) {
        //判斷該類是否實現了接口。
        if(!interfaces.isAssignableFrom(klass)){
            return;
        }
        try {
            doregistry(rpcBeanFactory, interfaces, klass.newInstance());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

這個類是同時給客戶端和服務器使用的，所以有一個私有方法來完成類方法的獲得和序列號的產生（method.toString().hashcode()）。然后將類、方法、對象放進RpcBeanDefinition中，將得到的序列號作為鍵和rpcBeanDeifinyion作為值放進Map中，形成鍵值對，方便客戶端和服務器的調用。

4.下來是RpcServer:

1 package com.xupt.rpc.core;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;

public class RpcServer implements Runnable {
    
    private ServerSocket server;
    private int port;
    private boolean goon;
    private final RpcBeanFactory rpcBeanFactory;
    private static long executorId;
    
    public RpcServer() {
        rpcBeanFactory = new RpcBeanFactory();
        this.goon = false;
    }

    public void setPort(int port) {
        this.port = port;
    }
    
    public void startRpcServer() throws Exception {
        if(this.port == 0) {
            return;
        }
        server = new ServerSocket(port);
        this.goon = true;
        new Thread(this,"Rpc_Server").start();//啟動線程
    }
    
    public void stopRpcServer() {   //關閉服務器
        if (this.server != null && !this.server.isClosed()) {
            try {
                this.server.close();
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                this.server = null;
            }
        }
    }
    
    RpcBeanFactory getRpcBeanFactory() {
        return rpcBeanFactory;
    }
    
    //用注冊的方法知道類實現的接口，類的對象，通過對象執行類的方法。
    public void rpcRegistry(Class<?> interfaces,Object object) {
        RpcBeanRegistry.registryInterface(rpcBeanFactory, interfaces, object);
    }
    
    public void rpcRegistry(Class<?> interfaces, Class<?> imClass) {
        RpcBeanRegistry.registryInterface(rpcBeanFactory, interfaces,imClass);
    }
    
    @Override
    public void run() {
        while(goon) {
            try {
                Socket socket = server.accept();//不斷的偵聽
                
                new RpcServerExecutor(socket, this,++executorId);
                
            } catch (Exception e) {
                goon = false;
                e.printStackTrace();
            }
        }
        stopRpcServer();
    }
}

上述服務器端的基本任務已經完成了。

下來我們來看看客戶端：

首先還是：RpcClientExecutor，它是建立與服務器端的連接，以及接收，發送消息，具體發送的是方法的序列號和參數類型。

package com.xupt.rpc.core;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;

public class RpcClientExecutor {
	
	private String rpcServerIp;  //ip地址
	private int rpcServerPort;	//端口號
	
	RpcClientExecutor() {
	}

	RpcClientExecutor(String rpcServerIp, int rpcServerPort) {
		this.rpcServerIp = rpcServerIp;
		this.rpcServerPort = rpcServerPort;
	}

	String getRpcServerIp() {
		return rpcServerIp;
	}

	void setRpcServerIp(String rpcServerIp) {
		this.rpcServerIp = rpcServerIp;
	}

	int getRpcServerPort() {
		return rpcServerPort;
	}

	void setRpcServerPort(int rpcServerPort) {
		this.rpcServerPort = rpcServerPort;
	}
	
	//關閉客戶端
	private void closeSocket(ObjectInputStream ois, ObjectOutputStream oos, Socket socket) {
		try {
			if (ois != null) {
				ois.close();
			}
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			ois = null;
		}
		try {
			if (oos != null) {
				oos.close();
			}
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			oos = null;
		}
		try {
			if (socket != null && !socket.isClosed()) {
				socket.close();
			}
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			socket = null;
		}
	}
	
	@SuppressWarnings("unchecked")
	<T> T rpExecutor(String beanId,Object[] params) throws Exception {
		//連接服務器端
		Socket socket = new Socket(rpcServerIp, rpcServerPort);
		//發送方法的序列號和參數類型。
		ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
		oos.writeUTF(beanId);
		oos.writeObject(params);
		
		//必須將這句放在前面三句的后面。
		ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
		//接收服務器端返回的結果。
		Object result = ois.readObject();
		
		closeSocket(ois,oos,socket);
		
		return (T) result;
		
		
	}
}

 下來是：RpcClient：這個類是產生代理，用代理對象偽執行相關的方法，然后在真正的來連接服務器。

package com.xupt.rpc.core;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;

public class RpcClient {
    
    private RpcClientExecutor rpcClientExecutor;
    
    public RpcClient(String rpcServerIp,int rpcServerport) {
        this.rpcClientExecutor = new RpcClientExecutor(rpcServerIp, rpcServerport);
    }
    
    @SuppressWarnings("unchecked")
    public <T> T getProxy(Class<?> klass) {
        return (T) Proxy.newProxyInstance(
                klass.getClassLoader(),
                new Class[] {klass},
                new InvocationHandler() {
                    
                    @Override
                    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                        String BeanId = String.valueOf(method.toString().hashCode());
                        Object result = rpcClientExecutor.rpExecutor(BeanId, args);
                        
                        return result;
                    }
                }
            );
    }
}

上述方法產生代理就不多做解釋，不明白請看上一篇的代理機制之初見吧。。。。

以上就是我的RMI的簡單實現，入如果有錯誤，請指正！也希望它對你有所幫助。