利用netty手写rpc框架

前言：利用netty异步事件驱动的网络通信模型，来实现rpc通信

一、大致目录结构：

二、两个端：服务端（发布），客户端（订阅消费），上代码：

1.服务端（发布）：

RPCServer：

代码：

public class RpcServer {private Map<String, Object> registryMap = new HashMap<>();private List<String> classCache = new ArrayList<>();// 1.实现发布；实现方式：// 1.1查找指定目录下的所有接口，放入一个集合中// 1.2遍历所有接口，放入一个map中存放接口路径及接口名称// 1.3发送方法相关信息public void publish(String providerPackage) throws Exception {getProviderClass(providerPackage);doRegister();NioEventLoopGroup parentGroup = new NioEventLoopGroup();NioEventLoopGroup childGroup = new NioEventLoopGroup();try {ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.group(parentGroup, childGroup).option(ChannelOption.SO_BACKLOG, 1024).childOption(ChannelOption.SO_KEEPALIVE, true).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {ChannelPipeline pipeline = socketChannel.pipeline();pipeline.addLast(new ObjectEncoder());pipeline.addLast(new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));pipeline.addLast(new RpcServerHandler(registryMap));}});ChannelFuture future = serverBootstrap.bind(9999).sync();System.out.println("服务端监听9999端口，启动成功。。。");future.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();} finally {parentGroup.shutdownGracefully();childGroup.shutdownGracefully();}}private void doRegister() throws Exception {if (classCache.size() > 0) {for (String className :classCache){Class<?> clazz = Class.forName(className);String interfaceName = clazz.getInterfaces()[0].getName();registryMap.put(interfaceName, clazz.newInstance());}}}/*** 获取当前目录下的所有接口，汇总到一个集合中* @param providerPackage*/private void getProviderClass(String providerPackage) {URL resource =         this.getClass().getClassLoader().getResource(providerPackage.replaceAll("\\.", "/"));File file = null;if (resource != null) {file = new File(resource.getFile());}if (file != null) {for (File f : Objects.requireNonNull(file.listFiles())) {if (f.isDirectory()) {getProviderClass(providerPackage + "." + f.getName());} else if (f.getName().endsWith(".class")) {String fileName = f.getName().replace(".class", "").trim();classCache.add(providerPackage + "." + fileName);}}}}}

服务端处理器：用于处理消费端发送过来的接口数据

代码：

public class RpcServerHandler extends ChannelInboundHandlerAdapter {private Map<String, Object> registryMap;public RpcServerHandler(Map<String, Object> registryMap) {this.registryMap = registryMap;}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {System.out.println("rpc-server收到客户端消息：" + msg);if (msg instanceof InvokeMessage) {InvokeMessage method = (InvokeMessage) msg;Object result = "rpc-server端没有该方法";// 判断是否存在该方法if (registryMap.containsKey(method.getClassName())) {Object provider = registryMap.get(method.getClassName());result = provider.getClass().getMethod(method.getMethodName(), method.getParamTypes()).invoke(provider, method.getParamValues());}// 把方法返回结果，发给订阅者ctx.channel().writeAndFlush(result);ctx.close();}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.close();}
}

接口的实现类：这里的接口是客户端的接口Api

代码：

public class SsenServiceApiImpl implements SsenServiceApi {@Overridepublic String hellRpc(String name) {return name + "实现类方法";}
}

2.客户端：（订阅消费）

这里采用JDK原生的基于接口的动态代理f发

public class RpcProxy {// JDK基于接口的动态代理，用于创建代理对象public static <T> T create(Class<?> clazz) {return (T) Proxy.newProxyInstance(clazz.getClassLoader(),new Class[]{clazz},new InvocationHandler() {@Overridepublic Object invoke(Object proxy, Method method, Object[] args) throws Throwable {if (Object.class.equals(method.getDeclaringClass())) {return method.invoke(proxy, method, args);}// 通过netty将接口信息发送给提供者，获取指定方法return RpcInvoke(clazz, method, args);}});}private static Object RpcInvoke(Class<?> clazz, Method method, Object[] args) {NioEventLoopGroup eventGroup = new NioEventLoopGroup();RpcClientHandler rpcClientHandler = new RpcClientHandler();try {Bootstrap bootstrap = new Bootstrap();bootstrap.group(eventGroup).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {ChannelPipeline pipeline = socketChannel.pipeline();pipeline.addLast(new ObjectEncoder());pipeline.addLast(new ObjectDecoder(Integer.MAX_VALUE,ClassResolvers.cacheDisabled(null)));pipeline.addLast(rpcClientHandler);}});// 绑定指定服务地址ChannelFuture future = bootstrap.connect("localhost", 9999).sync();// 指定接口信息发送给提供者InvokeMessage invokeMessage = new InvokeMessage();invokeMessage.setClassName(clazz.getName());invokeMessage.setMethodName(method.getName());invokeMessage.setParamTypes(method.getParameterTypes());invokeMessage.setParamValues(args);future.channel().writeAndFlush(invokeMessage).sync();future.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();} finally {eventGroup.shutdownGracefully();}return rpcClientHandler.getResult();}}