前言
学过dubbo的应该知道dubbo底层基于Netty实现，为了加强对Netty的理解，这篇文章我们来仿照dubbo手撸一个简易版本的RPC框架
结构理解
先来看一张图
原理还是比较简单 : 代理 + 线程池 + Netty 下面做一些解释：
- 首先需要定义一个统一的API接口，例：UserApi , 服务端(provider)需要实现这个接口，提供相应的方法UserApiImpl#save，客户端通过远程来调用该接口。
- 然后需要约定一个协议，服务器如何才能识别到客户端要调用哪个接口？：我这里用 “接口权限定名#方法名#参数” ，的方式来，因为是一个简单版本的RPC。服务端解析该内容就能匹配对应的接口的实现类，然后调用该方法。并把方法的返回值通过Netty写回给客户端
- 使用的编解码器都是比价简单的String的编解码器
- 提供者/服务端(provider)正常拉起NettyServer ，启动即可等待客户端的连接。
- 消费者/客户端(consumer)需要调用UserApi API接口，但是在消费者这边是没有实现类的，消费者要做的事情就是在发起 UserApi#save调用的时候，底层通过Netty向服务端通信。内容就是 “接口权限定名#方法#参数”。
- 消费者这边最好的是方式就是为接口生成代理，在代理类去约定协议，组装通信的内容，然后这里还用到了线程池，把发送Netty通信的工作交给新开的线程去处理。
- 请求发送给服务端，服务端返回一个结果，通过Netty拿到结果返回给消费者即可,整个调用过程结束。
统一API
首先定义一个统一的接口，提供一个save方法
publicinterfaceUserApi { Stringsave(Stringdata); }
提供者
第一步：提供者编写一个实现类实现该接口,save方法，返回一个“success” , 消费者通过远程来调用该方法
//实现类publicclassUserApiImplimplementsUserApi { publicStringsave(Stringdata) { System.out.println("调用方法 UserApiImpl#save ，参数： "+data); return"success"; } }
第二步：编写NettyServer，监听的IP和端口通过方法传入，这里没有什么改动
//提供者方启动publicclassNettyServer { publicvoidstart(Stringaddress,intport ){ NioEventLoopGroupbossGroup=newNioEventLoopGroup(); NioEventLoopGroupworkGroup=newNioEventLoopGroup(); ServerBootstrapbootstrap=newServerBootstrap(); bootstrap.group(bossGroup,workGroup); bootstrap.channel(NioServerSocketChannel.class); bootstrap.childHandler(newChannelInitializer<SocketChannel>() { protectedvoidinitChannel(SocketChannelch) throwsException { ChannelPipelinepipeline=ch.pipeline(); pipeline.addLast("decoder",newStringDecoder()); pipeline.addLast("encoder",newStringEncoder()); pipeline.addLast(newServerHandler()); } }); try { ChannelFuturesync=bootstrap.bind(address, port).sync(); sync.channel().closeFuture().sync(); } catch (InterruptedExceptione) { e.printStackTrace(); }finally { bossGroup.shutdownGracefully(); workGroup.shutdownGracefully(); } } }
第三步：编写Provider启动入口
//启动提供者publicclassProviderStart { publicstaticvoidmain(String[] args) { newNettyServer().start("127.0.0.1",1000); } }
第四步：编写ServerHandler，在channelRead0读取到数据，需要解析内容，匹配相关的service接口，且调用方法，把结果写回给客户端
publicclassServerHandlerextendsSimpleChannelInboundHandler<String> { //这里没有Spring的容器，就搞一个MapConcurrentHashMap<String, Object>applicationContext=newConcurrentHashMap<>(); publicServerHandler(){ applicationContext.put("cn.itsource.rpc.api.UserApi",newUserApiImpl()); } //消息内容约定 cn.itsource.rpc.api.UserApi#save#数据protectedvoidchannelRead0(ChannelHandlerContextctx, Stringmsg) throwsException { System.out.println("服务端收到请求 -> "+msg); String[] msgs=msg.split("#"); StringinterfaceClass=msgs[0]; StringmethodName=msgs[1]; Stringdata=msgs[2]; //调研Bean的方法Objectobj=applicationContext.get(interfaceClass); Class<?>aClass=obj.getClass(); Methodmethod=aClass.getMethod(methodName,data.getClass()); //调用方法Objectresult=method.invoke(obj,data); if(resultinstanceofString){ ctx.writeAndFlush((String)result); }else{ System.out.println("类型不支持"); } } }
上面只是简单模拟了一下servie调用，整合Sping可以根据类型去容器中获取Bean。到这服务端编写完成
消费者
消费者要复杂一些，第一步：先编写NettyClient
publicclassNettyClient { //客户端处理器,是一个 CallableprivateClientHandlerclientHandler=null; //初始化netty客户端publicvoidinit(Stringaddress, intport){ NioEventLoopGroupworkGroup=newNioEventLoopGroup(); Bootstrapbootstrap=newBootstrap(); bootstrap.group(workGroup); bootstrap.channel(NioSocketChannel.class); clientHandler=newClientHandler(); bootstrap.handler(newChannelInitializer<SocketChannel>() { protectedvoidinitChannel(SocketChannelch) throwsException { ChannelPipelinepipeline=ch.pipeline(); pipeline.addLast(newStringEncoder()); pipeline.addLast(newStringDecoder()); //一个客户端一个handlerpipeline.addLast(clientHandler); } }); try { ChannelFuturesync=bootstrap.connect(address, port).sync(); } catch (InterruptedExceptione) { e.printStackTrace(); } } //执行任务的线程池ExecutorServiceexecutorService=Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()); publicObjectgetBean(Class<UserApi>userApiClass) { //创建 userApi 接口的代理ObjectproxyInstance=Proxy.newProxyInstance(userApiClass.getClassLoader(), newClass[]{userApiClass}, (proxy, method, args) -> { //初始化客户端，连接Netty服务端if(clientHandler==null)init("127.0.0.1",1000); //把协议头#数据 ，传递给handlerclientHandler.setContent(userApiClass.getName()+"#"+method.getName()+"#"+args[0]); //把请求交给线程池处理，clientHandler就是一个线程returnexecutorService.submit(clientHandler).get(); }); returnproxyInstance; } }
NettyClient中提供了两个方法，init和 getBean方法。在init方法中去初始化NettyClient ，ClientHandler提升为了成员变量，因为下getBean面生成代理的时候会用到，然后getBean方法为接口生成了代理。在代理中把需要发送的内容“类权限定名#方法名#数据”交给Handler,把clientHandler交给线程池去执行。
第二步：编写Handler ，handler是一个Callable 如下
publicclassClientHandlerextendsSimpleChannelInboundHandler<String>implementsCallable<Object> { //协议头：cn.itsource.rpc.api.UserApi#save#dataprivateStringcontent=null; //上下文privateChannelHandlerContextctx; //服务器返回的结果privateObjectresult; publicvoidsetContent(Stringcontent){ this.content=content; } publicvoidchannelActive(ChannelHandlerContextctx) throwsException { //当和服务器建立连接，就需要保存ChannelHandlerContext//在call方法中发请求会用到ChannelHandlerContextthis.ctx=ctx; } protectedsynchronizedvoidchannelRead0(ChannelHandlerContextctx, Stringmsg) throwsException { System.out.println("客户端收到结果 = "+msg); //拿到服务器返回的结果this.result=msg; //唤醒call方法取走结果notify(); } publicsynchronizedObjectcall() throwsException { //发送请求if(ctx==null){ System.out.println("RPC连接失败..."); returnnull; } //把内容写给服务端ctx.writeAndFlush(content); //服务器返回的结果是在 channelRead0 方法中拿到，这里等待wait(); returnresult; } }
在 channelActive方法中，当和服务端建立连接就把ChannelHandlerContext上下文提升为成员变量，方便在call方法中使用。 当call方法被调用，使用ChannelHandlerContext把内容写给服务端，这个时候call方法需要等待服务端返回结果，使用了wait().
当服务端返回结果给客户端，客户端通过channelRead0方法拿到结果，并赋值给result成员变量，然后notify();唤醒wait的call方法，在call方法中就可以拿到结果返回。
第三步：就是通过NettyClient拿到UserApi代理，然后调用save方法
//消费者启动publicclassConsumerStart { publicstaticvoidmain(String[] args) { NettyClientnettyClient=newNettyClient(); UserApiuserApi= (UserApi)nettyClient.getBean(UserApi.class); for(inti=0 ; i<10 ; i++){ Stringcontent="data"+i; System.out.println("客户端发送数据，内容："+content); Stringresult=userApi.save(content); System.out.println("客户端收到返回的数据，内容："+result); } } }
第四步 ： 依次启动服务端，客户端，观看效果
客户端
服务端能拿到客户端传过去的数据，客户端也能收到服务端返回的结果。
