简单RPC之Socket实现

最近在学习RPC，看了些文章和代码，RPC的底层是通过SOCKET通信来实现的，这篇文章就是关于RPC的Socket简单实现。

rpc的相关概念可以参考这篇文章：http://blog.csdn.net/yinwenjie/article/details/49453303

一、技术方案

使用比较原始的方案实现RPC框架，采用Socket通信、动态代理与反射与Java原生的序列化。

二、RPC框架架构

1）服务提供者，运行在服务器端，提供服务接口定义与服务实现类。

2）服务中心，运行在服务器端，负责将本地服务发布成远程服务，管理远程服务，提供给服务消费者使用。

3）服务消费者，运行在客户端，通过远程代理对象调用远程服务。

三、RPC架构图



四、具体实现

1、client端代码

package org.weir.rpc.BioRpc.client;

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

public class BioClientInvocationHandler implements InvocationHandler{
Class serviceinterface = null;
int port;
public BioClientInvocationHandler(Class rpcInterface,int port){
this.serviceinterface = rpcInterface;
this.port = port;
}
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
// TODO Auto-generated method stub
Socket socket = null;
ObjectOutputStream output = null;
ObjectInputStream input = null;
try {
// 2.创建Socket客户端，根据指定地址连接远程服务提供者
socket = new Socket();
socket.connect(new InetSocketAddress(port));

// 3.将远程服务调用所需的接口类、方法名、参数列表等编码后发送给服务提供者
output = new ObjectOutputStream(socket.getOutputStream());
output.writeUTF(serviceinterface.getName());
output.writeUTF(method.getName());
output.writeObject(method.getParameterTypes());
output.writeObject(args);

// 4.同步阻塞等待服务器返回应答，获取应答后返回
input = new ObjectInputStream(socket.getInputStream());
return input.readObject();
} finally {
if (socket != null)
socket.close();
if (output != null)
output.close();
if (input != null)
input.close();
}
}

}

package org.weir.rpc.BioRpc.client;

import java.lang.reflect.Proxy;

public class BIOClient<T> {
public int port;

public <T> T refer(Class<T> rpcInterface) {
return getProxy(rpcInterface);
}

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

@SuppressWarnings("unchecked")
public <T> T getProxy(Class<T> rpcInterface) {
return (T) Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[] { rpcInterface },
new BioClientInvocationHandler(rpcInterface,port));
}

}

2、server端代码

server端接口类，为了以后扩展，所以写出接口形式

package org.weir.rpc.BioRpc.server;

import java.io.IOException;

public interface BIOServer {
public void stop();

public void start() throws IOException;

public void register(Class serviceInterface, Class impl);

public boolean isRunning();

public int getPort();
}

server端实现类

package org.weir.rpc.BioRpc.server;

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 BIOServiceCenter implements BIOServer {
private static int port;
private static final HashMap<String, Class> serviceRegistry = new HashMap<String, Class>();
private static boolean isRunning = false;

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

@Override
public void stop() {
// TODO Auto-generated method stub
isRunning = false;
}

@Override
public void start() throws IOException {
// TODO Auto-generated method stub
ServerSocket serverSocket = new ServerSocket();
serverSocket.bind(new InetSocketAddress(port));
System.out.println("[weir-rpc]->start server!!!");
ObjectOutputStream output = null;
try {
while (true) {
Socket socket = serverSocket.accept();
ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
String serviceName = ois.readUTF();
String methodName = ois.readUTF();
Class<?>[] parameterTypes = (Class<?>[]) ois.readObject();
Object[] arguments = (Object[]) ois.readObject();
Class<?> serviceClass = serviceRegistry.get(serviceName);
if (serviceClass == null) {
throw new ClassNotFoundException("[weir-rpc]" + serviceName + "not found");
}
Method method = serviceClass.getMethod(methodName, parameterTypes);
Object result = method.invoke(serviceClass.newInstance(), arguments);
// 3.将执行结果反序列化，通过socket发送给客户端
output = new ObjectOutputStream(socket.getOutputStream());
output.writeObject(result);
}
} catch (Exception e) {
e.printStackTrace();
}

}

@Override
public void register(Class serviceInterface, Class impl) {
// TODO Auto-generated method stub
serviceRegistry.put(serviceInterface.getName(), impl);
}

@Override
public boolean isRunning() {
// TODO Auto-generated method stub
return isRunning;
}

@Override
public int getPort() {
// TODO Auto-generated method stub
return port;
}

}

3、test类

package org.weir.rpc;

import java.io.IOException;

import org.weir.rpc.BioRpc.client.BIOClient;
import org.weir.rpc.BioRpc.server.BIOServer;
import org.weir.rpc.BioRpc.server.BIOServiceCenter;
import org.weir.rpc.api.RpcService;
import org.weir.rpc.api.impl.RpcServiceImpl;

public class TestBIORpc {
public static void main(String[] args) throws IOException {
new Thread(new Runnable() {
public void run() {
try {
BIOServer serviceServer = new BIOServiceCenter(8088);
serviceServer.register(RpcService.class, RpcServiceImpl.class);
serviceServer.start();
} catch (IOException e) {
e.printStackTrace();
}
}
}).start();
BIOClient bioClient = new BIOClient(8089);

RpcService service = (RpcService) bioClient.getProxy(RpcService.class);
System.out.println(service.hello("hello"));
}
}

4、远程service类接口

package org.weir.rpc.api;

public interface RpcService {
public String hello(String hi);
}

service实现类

package org.weir.rpc.api.impl;

import org.weir.rpc.api.RpcService;

public class RpcServiceImpl implements RpcService{

@Override
public String hello(String hi) {
// TODO Auto-generated method stub
System.out.println("[weir-Rpc]->hello"+hi);
return "[weir-Rpc]->hello"+hi;
}

}

运行test类，结果如下



五、总结

    RPC调用框架就是屏蔽了底层调用的具体细节，使得调用远程服务可以像调用本地服务一样方便。

     这里实现的简单RPC框架是使用Java语言开发，与Java语言高度耦合，并且通信方式采用的Socket是基于BIO实现的，IO效率不高，还有Java原生的序列化机制占内存太多，运行效率也不高。可以考虑从下面几种方法改进。
可以使用NIO或直接使用Netty替代BIO实现；
使用开源的序列化机制，如JSON、kryo、hessian等序列化机制，提高效率。
这篇文件的服务注册直接使用HashMap来注册，如果server服务宕机会失去服务的注册信息，所以服务注册可以使用Zookeeper进行管理，能够让应用更加稳定。
同步等待机制对于复杂计算服务不太合适，可以考虑使用异步的通信方式。
可以专门封装一个消息对象，用来通信，更加面向对象。