所谓RPC就是远程方法调用(Remote Process Call ),简单的来说就是通过MQ,TCP,HTTP或者自己写的网络协议来传输我要调用对方的什么接口,对方处理之后再把结果返回给我.就这么简单的一个过程。
运行时,一次客户机对服务器的RPC调用,其内部操作大致有如下十步：
1、调用客户端句柄；执行传送参数
2、调用本地系统内核发送网络消息
3、消息传送到远程主机
4、服务器句柄得到消息并取得参数
5、执行远程过程
6、执行的过程将结果返回服务器句柄
7、服务器句柄返回结果，调用远程系统内核
8、消息传回本地主机
9、客户句柄由内核接收消息
10、客户接收句柄返回的数据
之前一篇文章简单RPC之Socket实现我们通过socket通信实现了简单的RPC调用，接下来我们基于Netty来实现一个简单的RPC调用过程，当然还有很多不完善的地方，只供参考学习RPC使用。
一、首先定义消息传递的实体类
public class ClassInfo implements Serializable {
private static final long serialVersionUID = -8970942815543515064L;
private String className;//类名
private String methodName;//函数名称
private Class<?>[] types;//参数类型
private Object[] objects;//参数列表
public String getClassName() {
return className;
}
public void setClassName(String className) {
this.className = className;
}
public String getMethodName() {
return methodName;
}
public void setMethodName(String methodName) {
this.methodName = methodName;
}
public Class<?>[] getTypes() {
return types;
}
public void setTypes(Class<?>[] types) {
this.types = types;
}
public Object[] getObjects() {
return objects;
}
public void setObjects(Object[] objects) {
this.objects = objects;
}
}
二、创建Netty操作的服务端，以及具体操作
（1）服务端
public class RPCServer {
private int port;
public RPCServer(int port){
this.port = port;
}
public void start(){
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap().group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
.localAddress(port).childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
pipeline.addLast(new LengthFieldPrepender(4));
pipeline.addLast("encoder", new ObjectEncoder());
pipeline.addLast("decoder", new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));
pipeline.addLast(new InvokerHandler());
}
}).option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true);
ChannelFuture future = serverBootstrap.bind(port).sync();
System.out.println("Server start listen at " + port );
future.channel().closeFuture().sync();
} catch (Exception e) {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception {
int port;
if (args.length > 0) {
port = Integer.parseInt(args[0]);
} else {
port = 8080;
}
new RPCServer(port).start();
}
}
（2）服务端操作，由服务端我们看到具体的数据传输操作是进行序列化的，具体的操作还是比较简单的，就是获取发送过来的信息，这样就可以通过反射获得类名，根据函数名和参数值，执行具体的操作，将执行结果发送给客户端。
public class InvokerHandler extends ChannelInboundHandlerAdapter {
public static ConcurrentHashMap<String, Object> classMap = new ConcurrentHashMap<String,Object>();
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ClassInfo classInfo = (ClassInfo)msg;
Object claszz = null;
if(!classMap.containsKey(classInfo.getClassName())){
try {
claszz = Class.forName(classInfo.getClassName()).newInstance();
classMap.put(classInfo.getClassName(), claszz);
} catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) {
e.printStackTrace();
}
}else {
claszz = classMap.get(classInfo.getClassName());
}
Method method = claszz.getClass().getMethod(classInfo.getMethodName(), classInfo.getTypes());
Object result = method.invoke(claszz, classInfo.getObjects());
ctx.write(result);
ctx.flush();
ctx.close();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
三、客户端，通过代理机制来触发远程调用
（1）客户端，当执行具体的函数时会调用远程操作，将具体操作的类、函数及参数信息发送到服务端
public class RPCProxy {
@SuppressWarnings("unchecked")
public static <T> T create(Object target){
return (T) Proxy.newProxyInstance(target.getClass().getClassLoader(),target.getClass().getInterfaces(), new InvocationHandler(){
@Override
public Object invoke(Object proxy, Method method, Object[] args)
throws Throwable {
ClassInfo classInfo = new ClassInfo();
classInfo.setClassName(target.getClass().getName());
classInfo.setMethodName(method.getName());
classInfo.setObjects(args);
classInfo.setTypes(method.getParameterTypes());
ResultHandler resultHandler = new ResultHandler();
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
pipeline.addLast("encoder", new ObjectEncoder());
pipeline.addLast("decoder", new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));
pipeline.addLast("handler",resultHandler);
}
});
ChannelFuture future = b.connect("localhost", 8080).sync();
future.channel().writeAndFlush(classInfo).sync();
future.channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
return resultHandler.getResponse();
}
});
}
}
（2）获取远程调用返回的结果值
public class ResultHandler extends ChannelInboundHandlerAdapter {
private Object response;
public Object getResponse() {
return response;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
response=msg;
System.out.println("client接收到服务器返回的消息:" + msg);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("client exception is general");
}
}
四、接口、实现类及Main操作
接口：
public interface HelloRpc {
String hello(String name);
}
实现类：
public class HelloRpcImpl implements HelloRpc {
@Override
public String hello(String name) {
return "hello "+name;
}
}
Main操作：
public class Main {
public static void main(String [] args){
HelloRpc helloRpc = new HelloRpcImpl();
helloRpc = RPCProxy.create(helloRpc);
System.err.println(helloRpc.hello("rpc"));
}
}
完整代码地址github