Netty入门到超神系列-手撸简单版RPC框架(仿Dubbo)

publicinterfaceUserApi {
Stringsave(Stringdata);
}

//实现类publicclassUserApiImplimplementsUserApi {
@OverridepublicStringsave(Stringdata) {
System.out.println("调用方法 UserApiImpl#save ，参数： "+data);
return"success";
    }
}

//提供者方启动publicclassNettyServer {
publicvoidstart(Stringaddress,intport ){
NioEventLoopGroupbossGroup=newNioEventLoopGroup();
NioEventLoopGroupworkGroup=newNioEventLoopGroup();
ServerBootstrapbootstrap=newServerBootstrap();
bootstrap.group(bossGroup,workGroup);
bootstrap.channel(NioServerSocketChannel.class);
bootstrap.childHandler(newChannelInitializer<SocketChannel>() {
@OverrideprotectedvoidinitChannel(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();
        }
    }
}

//启动提供者publicclassProviderStart {
publicstaticvoidmain(String[] args) {
newNettyServer().start("127.0.0.1",1000);
    }
}

publicclassServerHandlerextendsSimpleChannelInboundHandler<String> {
//这里没有Spring的容器，就搞一个MapConcurrentHashMap<String, Object>applicationContext=newConcurrentHashMap<>();
publicServerHandler(){
applicationContext.put("cn.itsource.rpc.api.UserApi",newUserApiImpl());
    }
//消息内容约定   cn.itsource.rpc.api.UserApi#save#数据@OverrideprotectedvoidchannelRead0(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("类型不支持");
        }
    }
}

publicclassNettyClient {
//客户端处理器,是一个 CallableprivateClientHandlerclientHandler=null;
//初始化netty客户端publicvoidinit(Stringaddress, intport){
NioEventLoopGroupworkGroup=newNioEventLoopGroup();
Bootstrapbootstrap=newBootstrap();
bootstrap.group(workGroup);
bootstrap.channel(NioSocketChannel.class);
clientHandler=newClientHandler();
bootstrap.handler(newChannelInitializer<SocketChannel>() {
@OverrideprotectedvoidinitChannel(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;
    }
}

publicclassClientHandlerextendsSimpleChannelInboundHandler<String>implementsCallable<Object> {
//协议头：cn.itsource.rpc.api.UserApi#save#dataprivateStringcontent=null;
//上下文privateChannelHandlerContextctx;
//服务器返回的结果privateObjectresult;
publicvoidsetContent(Stringcontent){
this.content=content;
    }
@OverridepublicvoidchannelActive(ChannelHandlerContextctx) throwsException {
//当和服务器建立连接，就需要保存ChannelHandlerContext//在call方法中发请求会用到ChannelHandlerContextthis.ctx=ctx;
    }
@OverrideprotectedsynchronizedvoidchannelRead0(ChannelHandlerContextctx, Stringmsg) throwsException {
System.out.println("客户端收到结果 = "+msg);
//拿到服务器返回的结果this.result=msg;
//唤醒call方法取走结果notify();
    }
@OverridepublicsynchronizedObjectcall() throwsException {
//发送请求if(ctx==null){
System.out.println("RPC连接失败...");
returnnull;
        }
//把内容写给服务端ctx.writeAndFlush(content);
//服务器返回的结果是在 channelRead0 方法中拿到，这里等待wait();
returnresult;
    }
}

//消费者启动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);
        }
    }
}