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

简介: 原理还是比较简单 : 代理 + 线程池 + Netty 下面做一些解释:首先需要定义一个统一的API接口,例:UserApi , 服务端(provider)需要实现这个接口,提供相应的方法UserApiImpl#save,客户端通过远程来调用该接口。然后需要约定一个协议,服务器如何才能识别到客户端要调用哪个接口?:我这里用 “接口权限定名#方法名#参数” ,的方式来,因为是一个简单版本的RPC。服务端解析该内容就能匹配对应的接口的实现类,然后调用该方法。并把方法的返回值通过Netty写回给客户端使用的编解码器都是比价简单的String的编解码器

前言

学过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 {
@OverridepublicStringsave(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>() {
@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();
        }
    }
}

第三步:编写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#数据@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("类型不支持");
        }
    }
}

上面只是简单模拟了一下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>() {
@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;
    }
}

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;
    }
@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;
    }
}

在 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);
        }
    }
}

第四步 : 依次启动服务端,客户端,观看效果

客户端

服务端能拿到客户端传过去的数据,客户端也能收到服务端返回的结果。

目录
相关文章
|
2月前
|
Dubbo Java 应用服务中间件
微服务学习 | Springboot整合Dubbo+Nacos实现RPC调用
微服务学习 | Springboot整合Dubbo+Nacos实现RPC调用
|
2月前
|
缓存 网络协议 算法
Netty的基础入门(上)
Netty的基础入门(上)
129 0
|
2月前
|
XML Dubbo Java
【Dubbo3高级特性】「框架与服务」服务的异步调用实践以及开发模式
【Dubbo3高级特性】「框架与服务」服务的异步调用实践以及开发模式
99 0
|
2月前
|
负载均衡 Dubbo Java
Dubbo 3.x:探索阿里巴巴的开源RPC框架新技术
随着微服务架构的兴起,远程过程调用(RPC)框架成为了关键组件。Dubbo,作为阿里巴巴的开源RPC框架,已经演进到了3.x版本,带来了许多新特性和技术改进。本文将探讨Dubbo 3.x中的一些最新技术,包括服务注册与发现、负载均衡、服务治理等,并通过代码示例展示其使用方式。
154 9
|
2月前
|
监控 负载均衡 Dubbo
Dubbo 框架揭秘:分布式架构的精髓与魔法【一】
Dubbo 框架揭秘:分布式架构的精髓与魔法【一】
190 0
|
2月前
|
Dubbo Cloud Native 网络协议
【Dubbo3技术专题】「服务架构体系」第一章之Dubbo3新特性要点之RPC协议分析介绍
【Dubbo3技术专题】「服务架构体系」第一章之Dubbo3新特性要点之RPC协议分析介绍
58 1
|
1月前
|
网络协议 Java 物联网
Netty是什么?深入理解高性能网络框架
Netty是什么?深入理解高性能网络框架
100 1
|
2月前
|
负载均衡 Dubbo Java
理解 Netty 和 Dubbo 之间的关系:相互作用与作用域
在现代分布式系统和微服务架构中,高性能、高可用的通信框架是成功的关键要素之一。Netty 和 Dubbo 是两个非常流行的开源项目,它们在分布式系统的构建中发挥着重要作用。
|
2月前
|
Dubbo Java 应用服务中间件
Java从入门到精通:3.2.2分布式与并发编程——了解分布式系统的基本概念,学习使用Dubbo、Spring Cloud等分布式框架
Java从入门到精通:3.2.2分布式与并发编程——了解分布式系统的基本概念,学习使用Dubbo、Spring Cloud等分布式框架
293 0
|
2月前
|
XML 网络协议 前端开发
Netty网络框架(三)
Netty网络框架
42 1