所谓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
时间: 2024-10-07 04:02:11