RPC-BDY(3)

-2022.7.15 by BDY

文章目录

RPC-BDY(3)
前言
一、Netty
二、NettyClient和NettyServer
三、自定义协议与编解码器（了解）
四、序列化接口
五、NettyServerHandler 和 NettyClientHandler
六、测试
七、知识点
总结

前言

这一节的主要内容是将传统的BIO改为更为高效的NIO传输，使用技术栈netty。
其次实现通用的序列化接口和自定义传输协议

一、Netty

这里直接链接别人的文章Netty

简介

Netty是一个异步事件驱动的网络应用程序框架用于快速开发可维护的高性能协议服务器和客户端。
Netty 对 JDK 自带的 NIO 的 API 进行了封装，解决了原生NIO的一系列问题

Netty模型

1.Netty 抽象出两组线程池 BossGroup 专门负责接收客户端的连接，WorkerGroup 专门负责网络的读写
2.BossGroup 和 WorkerGroup 类型都是 NioEventLoopGroup
3.NioEventLoopGroup 相当于一个事件循环组，这个组中含有多个事件循环，每一个事件循环是 NioEventLoop
4.NioEventLoop 表示一个不断循环的执行处理任务的线程，每个 NioEventLoop 都有一个 Selector，用于监听绑定在其上的 socket 的网络通讯
5.NioEventLoopGroup 可以有多个线程，即可以含有多个 NioEventLoop
6.每个 BossNioEventLoop 循环执行的步骤有 3 步
轮询 accept 事件
处理 accept 事件，与 client 建立连接，生成 NioScocketChannel，并将其注册到某个 worker NIOEventLoop 上的 Selector
处理任务队列的任务，即 runAllTasks
7.每个 Worker NIOEventLoop 循环执行的步骤
轮询 read，write 事件
处理 I/O 事件，即 read，write 事件，在对应 NioScocketChannel 处理
处理任务队列的任务，即 runAllTasks
8.每个 Worker NIOEventLoop 处理业务时，会使用 pipeline（管道），pipeline 中包含了 channel，即通过 pipeline 可以获取到对应通道，管道中维护了很多的处理器。

Netty核心组件

Bootstrap、ServerBootstrap
Bootstrap 意思是引导，一个 Netty 应用通常由一个 Bootstrap 开始，主要作用是配置整个 Netty 程序，串联各个组件，Netty 中 Bootstrap 类是客户端程序的启动引导类，ServerBootstrap 是服务端启动引导类。
Future、ChannelFuture
Netty 中所有的 IO 操作都是异步的，不能立刻得知消息是否被正确处理。但是可以过一会等它执行完成或者直接注册一个监听，具体的实现就是通过 Future 和 ChannelFutures，他们可以注册一个监听，当操作执行成功或失败时监听会自动触发注册的监听事件
Channel
Netty 网络通信的组件，能够用于执行网络 I/O 操作。
通过 Channel 可获得当前网络连接的通道的状态
通过 Channel 可获得网络连接的配置参数（例如接收缓冲区大小）
Channel 提供异步的网络 I/O 操作(如建立连接，读写，绑定端口)，异步调用意味着任何 I/O 调用都将立即返回，并且不保证在调用结束时所请求的 I/O 操作已完成
调用立即返回一个 ChannelFuture 实例，通过注册监听器到 ChannelFuture 上，可以 I/O 操作成功、失败或取消时回调通知调用方
支持关联 I/O 操作与对应的处理程序
不同协议、不同的阻塞类型的连接都有不同的 Channel 类型与之对应
Selector
Netty 基于 Selector 对象实现 I/O 多路复用，通过 Selector 一个线程可以监听多个连接的 Channel 事件。
ChannelHandler 及其实现类
ChannelHandler 是一个接口，处理 I/O 事件或拦截 I/O 操作，并将其转发到其 ChannelPipeline（业务处理链）中的下一个处理程序。
Pipeline 和 ChannelPipeline
ChannelPipeline 是一个 Handler 的集合，它负责处理和拦截 inbound 或者 outbound 的事件和操作，相当于一个贯穿 Netty 的链。（也可以这样理解：ChannelPipeline 是保存 ChannelHandler 的 List，用于处理或拦截 Channel 的入站事件和出站操作）
ChannelPipeline 实现了一种高级形式的拦截过滤器模式，使用户可以完全控制事件的处理方式，以及 Channel 中各个的 ChannelHandler 如何相互交互
在 Netty 中每个 Channel 都有且仅有一个 ChannelPipeline 与之对应，它们的组成关系如下。
EventLoopGroup 和其实现类 NioEventLoopGroup
EventLoopGroup 是一组 EventLoop 的抽象，Netty 为了更好的利用多核 CPU 资源，一般会有多个 EventLoop 同时工作，每个 EventLoop 维护着一个 Selector 实例。
EventLoopGroup 提供 next 接口，可以从组里面按照一定规则获取其中一个 EventLoop 来处理任务。在 Netty 服务器端编程中，我们一般都需要提供两个 EventLoopGroup，例如：BossEventLoopGroup 和 WorkerEventLoopGroup。
通常一个服务端口即一个 ServerSocketChannel 对应一个 Selector 和一个 EventLoop 线程。BossEventLoop 负责接收客户端的连接并将 SocketChannel 交给 WorkerEventLoopGroup 来进行 IO 处理。

二、NettyClient和NettyServer

1.nettyclient

/*** NIO方式消费侧客户端类* @author bdy*/
public class NettyClient implements RpcClient {private static final Logger logger = LoggerFactory.getLogger(NettyClient.class);private String host;private int port;private static final Bootstrap bootstrap;public NettyClient(String host, int port) {this.host = host;this.port = port;}static {EventLoopGroup group = new NioEventLoopGroup();bootstrap = new Bootstrap();bootstrap.group(group).channel(NioSocketChannel.class).option(ChannelOption.SO_KEEPALIVE, true).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new CommonDecoder()).addLast(new CommonEncoder(new JsonSerializer())).addLast(new NettyClientHandler());}});}@Overridepublic Object sendRequest(RpcRequest rpcRequest) {try {ChannelFuture future = bootstrap.connect(host, port).sync();logger.info("客户端连接到服务器 {}:{}", host, port);Channel channel = future.channel();if(channel != null) {channel.writeAndFlush(rpcRequest).addListener(future1 -> {if(future1.isSuccess()) {logger.info(String.format("客户端发送消息: %s", rpcRequest.toString()));} else {logger.error("发送消息时有错误发生: ", future1.cause());}});channel.closeFuture().sync();AttributeKey<RpcResponse> key = AttributeKey.valueOf("rpcResponse");//res赋值为空?????????RpcResponse rpcResponse = channel.attr(key).get();return rpcResponse.getData();}} catch (InterruptedException e) {logger.error("发送消息时有错误发生: ", e);}return null;}}

2.nettyserver

/*** NIO方式服务提供侧* @author bdy*/
public class NettyServer implements RpcServer {private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);@Overridepublic void start(int port) {EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO)).option(ChannelOption.SO_BACKLOG, 256).option(ChannelOption.SO_KEEPALIVE, true).childOption(ChannelOption.TCP_NODELAY, true).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new CommonEncoder(new JsonSerializer()));pipeline.addLast(new CommonDecoder());pipeline.addLast(new NettyServerHandler());}});ChannelFuture future = serverBootstrap.bind(port).sync();future.channel().closeFuture().sync();} catch (InterruptedException e) {logger.error("启动服务器时有错误发生: ", e);} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}}

三、自定义协议与编解码器（了解）

+---------------+---------------+-----------------+-------------+
|  Magic Number |  Package Type | Serializer Type | Data Length |
|    4 bytes    |    4 bytes    |     4 bytes     |   4 bytes   |
+---------------+---------------+-----------------+-------------+
|                          Data Bytes                           |
|                   Length: ${Data Length}                      |
+---------------------------------------------------------------+

4 字节魔数：表识一个协议包。
Package Type：标明这是一个调用请求还是调用响应，
Serializer Type ：标明了实际数据使用的序列化器，这个服务端和客户端应当使用统一标准；
Data Length ：就是实际数据的长度，设置这个字段主要防止粘包，最后就是经过序列化后的实际数据，可能是 RpcRequest 也可能是 RpcResponse 经过序列化后的字节，取决于 Package Type。

1.CommonEncoder


/*** 通用的编码拦截器* @author bdy*/
public class CommonEncoder extends MessageToByteEncoder {private static final int MAGIC_NUMBER = 0xCAFEBABE;private final CommonSerializer serializer;public CommonEncoder(CommonSerializer serializer) {this.serializer = serializer;}@Overrideprotected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {out.writeInt(MAGIC_NUMBER);if(msg instanceof RpcRequest) {out.writeInt(PackageType.REQUEST_PACK.getCode());} else {out.writeInt(PackageType.RESPONSE_PACK.getCode());}out.writeInt(serializer.getCode());byte[] bytes = serializer.serialize(msg);out.writeInt(bytes.length);out.writeBytes(bytes);}}

2.CommonDncoder


/*** 通用的解码拦截器* @author bdy*/
public class CommonDecoder extends ReplayingDecoder {private static final Logger logger = LoggerFactory.getLogger(CommonDecoder.class);private static final int MAGIC_NUMBER = 0xCAFEBABE;@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {int magic = in.readInt();if(magic != MAGIC_NUMBER) {logger.error("不识别的协议包: {}", magic);throw new RpcException(RpcError.UNKNOWN_PROTOCOL);}int packageCode = in.readInt();Class<?> packageClass;if(packageCode == PackageType.REQUEST_PACK.getCode()) {packageClass = RpcRequest.class;} else if(packageCode == PackageType.RESPONSE_PACK.getCode()) {packageClass = RpcResponse.class;} else {logger.error("不识别的数据包: {}", packageCode);throw new RpcException(RpcError.UNKNOWN_PACKAGE_TYPE);}int serializerCode = in.readInt();CommonSerializer serializer = CommonSerializer.getByCode(serializerCode);if(serializer == null) {logger.error("不识别的反序列化器: {}", serializerCode);throw new RpcException(RpcError.UNKNOWN_SERIALIZER);}int length = in.readInt();byte[] bytes = new byte[length];in.readBytes(bytes);Object obj = serializer.deserialize(bytes, packageClass);out.add(obj);}}

四、序列化接口

四个方法：序列化，反序列化，获得该序列化器的编号，已经根据编号获取序列化器

使用jackson作为序列化工具
注意：
在 RpcRequest 反序列化时，由于其中有一个字段是 Object 数组，在反序列化时序列化器会根据字段类型进行反序列化，而 Object 就是一个十分模糊的类型，会出现反序列化失败的现象，这时就需要 RpcRequest 中的另一个字段 ParamTypes 来获取到 Object 数组中的每个实例的实际类，辅助反序列化，这就是 handleRequest() 方法的作用。

1.CommonSerializer

/*** 通用的序列化反序列化接口* @author bdy*/
public interface CommonSerializer {byte[] serialize(Object obj);Object deserialize(byte[] bytes, Class<?> clazz);int getCode();static CommonSerializer getByCode(int code) {switch (code) {case 1:return new JsonSerializer();default:return null;}}}

2.JsonSerializer

/*** 使用JSON格式的序列化器* @author ziyang*/
public class JsonSerializer implements CommonSerializer {private static final Logger logger = LoggerFactory.getLogger(JsonSerializer.class);private ObjectMapper objectMapper = new ObjectMapper();@Overridepublic byte[] serialize(Object obj) {try {return objectMapper.writeValueAsBytes(obj);} catch (JsonProcessingException e) {logger.error("序列化时有错误发生: {}", e.getMessage());e.printStackTrace();return null;}}@Overridepublic Object deserialize(byte[] bytes, Class<?> clazz) {try {Object obj = objectMapper.readValue(bytes, clazz);if(obj instanceof RpcRequest) {obj = handleRequest(obj);}return obj;} catch (IOException e) {logger.error("反序列化时有错误发生: {}", e.getMessage());e.printStackTrace();return null;}}/*这里由于使用JSON序列化和反序列化Object数组，无法保证反序列化后仍然为原实例类型需要重新判断处理*/private Object handleRequest(Object obj) throws IOException {RpcRequest rpcRequest = (RpcRequest) obj;for(int i = 0; i < rpcRequest.getParamTypes().length; i ++) {Class<?> clazz = rpcRequest.getParamTypes()[i];if(!clazz.isAssignableFrom(rpcRequest.getParameters()[i].getClass())) {byte[] bytes = objectMapper.writeValueAsBytes(rpcRequest.getParameters()[i]);rpcRequest.getParameters()[i] = objectMapper.readValue(bytes, clazz);}}return rpcRequest;}@Overridepublic int getCode() {return SerializerCode.valueOf("JSON").getCode();}}

五、NettyServerHandler 和 NettyClientHandler

1.NettyServerhandler

NettyServerhandler 用于接收 RpcRequest，并且执行调用，将调用结果返回封装成 RpcResponse 发送出去。


/*** Netty中处理RpcRequest的Handler* @author bdy*/
public class NettyServerHandler extends SimpleChannelInboundHandler<RpcRequest> {private static final Logger logger = LoggerFactory.getLogger(NettyServerHandler.class);private static RequestHandler requestHandler;//服务处理private static ServiceRegistry serviceRegistry;//注册表static {requestHandler = new RequestHandler();serviceRegistry = new DefaultServiceRegistry();//这里获取注册表，表里的内容已经注册好了}//当类实例化后，相当于开启线程@Overrideprotected void channelRead0(ChannelHandlerContext ctx, RpcRequest msg) throws Exception {try {logger.info("服务器接收到请求: {}", msg);String interfaceName = msg.getInterfaceName();Object service = serviceRegistry.getService(interfaceName);//通过请求处理获取结果Object result = requestHandler.handle(msg, service);//结果写入ChannelFuture future = ctx.writeAndFlush(RpcResponse.success(result));future.addListener(ChannelFutureListener.CLOSE);} finally {ReferenceCountUtil.release(msg);}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {logger.error("处理过程调用时有错误发生:");cause.printStackTrace();ctx.close();}}

2.NettyClientHandler

/*** Netty客户端侧处理器* @author bdy*/
public class NettyClientHandler extends SimpleChannelInboundHandler<RpcResponse> {private static final Logger logger = LoggerFactory.getLogger(NettyClientHandler.class);@Overrideprotected void channelRead0(ChannelHandlerContext ctx, RpcResponse msg) throws Exception {try {logger.info(String.format("客户端接收到消息: %s", msg));AttributeKey<RpcResponse> key = AttributeKey.valueOf("rpcResponse");ctx.channel().attr(key).set(msg);ctx.channel().close();} finally {ReferenceCountUtil.release(msg);}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {logger.error("过程调用时有错误发生:");cause.printStackTrace();ctx.close();}
}

六、测试

1.NettyTestServer

/*** 测试用Netty服务提供者（服务端）* @author bdy*/
public class NettyTestServer {public static void main(String[] args) {HelloService helloService = new HelloServiceImpl();//创建注册表ServiceRegistry registry = new DefaultServiceRegistry();//注册服务registry.register(helloService);NettyServer server = new NettyServer();//开启服务server.start(9998);}}

2.NettyTestClient

/*** 测试用Netty消费者* @author bdy*/
public class NettyTestClient {public static void main(String[] args) {//生成一个client实例RpcClient client = new NettyClient("127.0.0.1", 9998);//生成client的动态代理实例RpcClientProxy rpcClientProxy = new RpcClientProxy(client);//生成helloserver的动态代理实例HelloService helloService = rpcClientProxy.getProxy(HelloService.class);HelloObject object = new HelloObject(12, "This is a message");//执行动态代理String res = helloService.hello(object);System.out.println(res);}}

七、知识点

1.NIO

介绍：

Java NIO（New IO）也有人称之为 java non-blocking IO，也有人称为是new io，是从Java1.4版本开始引入的一个新的IO API，可以替代标准的Java IO API。

NIO与原来的IO有同样的作用和目的，但是使用的方式完全不同，NIO支持面向缓冲区的、基于通道的IO操作。NIO将以更加高效的方式进行文件的读写操作。NIO可以理解为非阻塞IO,传统的IO的read和write只能阻塞执行，线程在读写IO期间不能干其他事情，比如调用socket.read()时，如果服务器一直没有数据传输过来，线程就一直阻塞，而NIO中可以配置socket为非阻塞模式。

NIO 相关类都被放在 java.nio 包及子包下，并且对原 java.io 包中的很多类进行改写。

NIO 有三大核心部分：Channel( 通道) ，Buffer( 缓冲区), Selector( 选择器)

Java NIO 的非阻塞模式，使一个线程从某通道发送请求或者读取数据，但是它仅能得到目前可用的数据，如果目前没有数据可用时，就什么都不会获取，而不是保持线程阻塞，所以直至数据变的可以读取之前，该线程可以继续做其他的事情。非阻塞写也是如此，一个线程请求写入一些数据到某通道，但不需要等待它完全写入，这个线程同时可以去做别的事情。

通俗理解：NIO 是可以做到用一个线程来处理多个操作的。假设有 1000 个请求过来,根据实际情况，可以分配20 或者 80个线程来处理。不像之前的阻塞 IO 那样，非得分配 1000 个。

NIO和BIO比较
三大核心（Channel( 通道) ，Buffer( 缓冲区), Selector( 选择器)）

Buffer缓冲区
缓冲区本质上是一块可以写入数据，然后可以从中读取数据的内存。这块内存被包装成NIO Buffer对象，并提供了一组方法，用来方便的访问该块内存。相比较直接对数组的操作，Buffer API更加容易操作和管理。
Channel（通道）
Java NIO的通道类似流，但又有些不同：既可以从通道中读取数据，又可以写数据到通道。但流的（input或output)读写通常是单向的。通道可以非阻塞读取和写入通道，通道可以支持读取或写入缓冲区，也支持异步地读写。
Selector选择器
Selector是一个Java NIO组件，可以能够检查一个或多个 NIO 通道，并确定哪些通道已经准备好进行读取或写入。这样，一个单独的线程可以管理多个channel，从而管理多个网络连接，提高效率

2.netty实现过程
netty基础入门
netty实现过程

服务端注册服务
使用registry注册想要提供的服务
开启netty服务端
客户端发起请求
客户端开启netty，发起request，并生成动态代理对象
服务端处理请求并返回
服务端接受到request，处理请求，生成response，发送给客户端
客户端处理
客户端处理response 并生成结果

3.关于无参构造和有参构造

在HelloObject、RPCRequest、RPCResponse中都需要带有无参构造
原因：jackson的反序列化需要无参构造函数
补充：在未实现有参构造时，JVM会自动生成无参构造

在HelloObject、RPCRequest、RPCResponse中都需要带有无参构造
HelloObject 使用netty自带的编码解码器，需要实现有参构造
RPCRequese在反射调用时需要有参构造

//                          netty编码与解码器实现
//                          pipeline.addLast(new ObjectDecoder(1024*1024, ClassResolvers.cacheDisabled((this.getClass().getClassLoader()))));
//                          pipeline.addLast(new ObjectEncoder());

RpcRequest rpcRequest = new RpcRequest(method.getDeclaringClass().getName(),method.getName(), args, method.getParameterTypes());

4.注册中心的set集合是干嘛的？

暂时没用

5.动态代理的基本实现过程

1.new一个实现类对象；
2.获得实现类对象的代理对象：
2.1 调用Proxy.newProxyInstance来获得一个动态的代理对象，其接收三个参数，三个参数所代表的含义分别是：
①一个ClassLoader对象，定义了由哪个ClassLoader对象来对生成的代理对象进行加载
②一个Interface对象的数组，表示的是我将要给我需要代理的对象提供一组什么接口，如果我提供了一组接口给它，那么这个代理对象就宣称实现了该接口(多态)，这样我就能调用这组接口中的方法了
③ 一个InvocationHandler的实现类对象，表示的是当我这个动态代理对象在调用方法的时候，会关联到哪一个
InvocationHandler的实现类对象上
3.获得代理对象的类对象（这里我们可以打印一下代理类的类对象的类名）
4.获得代理类的所有方法（通过暴力反射getDeclaredMethods（）获得）（将所有获得的方法遍历，输出所有的方法名）
5.通过代理对象调用实现类的方法（并赋值），触发我们的重点步骤：InvocationHandler接口的实现类中的invoked（）方法，从而执行实现类的方法（sout输出结果即可）（这一步就是所谓的无侵入式编码规则）
5.1调用InvocationHandler接口具体步骤如下：
在触发了实现类的方法后，首先需要在InvocationHandler接口中传入三个参数，分别是
① proxy:　 - 指代我们所代理的那个真实对象
② method:　- 指代的是我们所要调用真实对象的某个方法的Method对象（）
③ args:　　- 指代的是调用真实对象某个方法时接受的参数
之后会执行接口独有的invoked（）方法，传入两个参数，分别是真实的实现类对象和传入的参数args，最后返回方法。

总结

RPC-BDY(3)-Netty实现相关推荐

【RPC】---- 基于Netty实现的RPC
基于Netty实现的RPC 一.Netty服务端和客户端 1.服务端server 1.1 NettyServer 1.2 NettyServerHandler 2.客户端client 2.1 Nett ...
分布式理论、架构设计（自定义RPC一 NIO NETTY）
分布式理论.架构设计自定义RPC 第一部分-RPC框架设计 1. Socket回顾与I/0模型 1.1 Socket网络编程回顾 1.1.1 Socket概述 1.1.2 Socket整体流程 1.1 ...
RPC系列之Netty实现自定义RPC框架
进行这个章节之前,需要去看一下RMI的实现哈,如果了解过的童鞋可以直接跳过,如果没有或者不知道RMI的童鞋,移驾到下面的链接看完之后再回来继续看这篇 RPC系列之入门_阿小冰的博客-CSDN博客RPC ...
谈谈如何使用Netty开发实现高性能的RPC服务器
RPC(Remote Procedure Call Protocol)远程过程调用协议,它是一种通过网络,从远程计算机程序上请求服务,而不必了解底层网络技术的协议.说的再直白一点,就是客户端在不必知道 ...
《Netty实战-写一个RPC应用》
一.服务端 1.1.定义rpc服务代理注解用于标识需要作为远程调用的接口,应用于类上 @Target(ElementType.TYPE) @Retention(RetentionPolicy.RUN ...
轻量级分布式 RPC 框架
RPC,即 Remote Procedure Call(远程过程调用),说得通俗一点就是:调用远程计算机上的服务,就像调用本地服务一样. RPC 可基于 HTTP 或 TCP 协议,Web Servi ...
Netty 用起来够猛！
Netty 作为当前流行的 NIO 框架,在游戏.大数据通讯,云计算.物联网等领域都有广泛的应用,大家熟知的 Dubbo,底层用的就是 Netty.尤其在高并发.高性能 RPC 方面,Netty 更是 ...
消息中间件—RocketMQ的RPC通信（二
作者:胡宗棠来源:匠心独运的博客在(一)篇中主要介绍了RocketMQ的协议格式,消息编解码,通信方式(同步/异步/单向).消息发送/接收以及异步回调的主要通信流程.而本篇将主要对RocketMQ ...
RPC与其实现方式概念笔记
一,消息队列服务一般用于设计多系统之间的信息传输,一般这种传输不需要对方对数据做出回应.它最常见的方式是构建异步的生产者-消费者模式.我们在系统开发中,有些业务并不需要及时返回结果,我们可以把这些操作 ...
底层框架_你有必要了解一下Flink底层RPC使用的框架和原理
1. 前言对于Flink中各个组件(JobMaster.TaskManager.Dispatcher等),其底层RPC框架基于Akka实现,本文着重分析Flink中的Rpc框架实现机制及梳理其通信流 ...

RPC-BDY(3)-Netty实现