学习前需要掌握基本的java网络编程，可参考这篇博客

Netty 简介

是什么？

Netty 是一个基于 NIO (Non-blocking I/O，非阻塞I/O)的 client-server(客户端服务器)框架，使用它可以快速简单地开发网络应用程序。它极大地简化并简化了 TCP 和 UDP 套接字服务器等网络编程,并且性能以及安全性等很多方面甚至都要更好。支持多种协议如 FTP，SMTP，HTTP 以及各种二进制和基于文本的传统协议。
我们平常经常接触的 Dubbo、RocketMQ、Elasticsearch、gRPC 等等都用到了 Netty。

特点：

1.统一的 API，支持多种传输类型，阻塞和非阻塞的。2.简单而强大的线程模型。3.自带编解码器解决 TCP 粘包/拆包问题。4.自带各种协议栈。5.真正的无连接数据包套接字支持。6.比直接使用 Java 核心 API 有更高的吞吐量、更低的延迟、更低的资源消耗和更少的内存复制。7.安全性不错，有完整的 SSL/TLS 以及 StartTLS 支持。8.社区活跃9.成熟稳定，经历了大型项目的使用和考验，而且很多开源项目都使用到了 Netty 比如我们经常接触的 Dubbo、RocketMQ 等等。
......

能做什么？

作为 RPC 框架的网络通信工具 、实现一个自己的 HTTP 服务器、实现一个即时通讯系统、消息推送系统等等

Netty 使用 kryo 序列化传输对象案例

我们首先定义两个对象，这两个对象是客户端与服务端进行交互的实体类。 客户端将 RpcRequest 类型的对象发送到服务端，服务端进行相应的处理之后将得到结果 RpcResponse 对象返回给客户端。

注意 ：Kryo不支持没有无参构造函数的对象进行反序列化，因此如果某个对象希望使用Kryo来进行序列化操作的话，需要有相应的无参构造函数才可以。

RpcRequest.java :客户端请求实体类

@AllArgsConstructor//lombok注解
@Getter
@NoArgsConstructor
@Builder
@ToString
public class RpcRequest {private String interfaceName;private String methodName;
}

RpcResponse.java ：服务端响应实体类

@AllArgsConstructor
@Getter
@NoArgsConstructor
@Builder
@ToString
public class RpcResponse {private String message;
}

客户端代码

客户端中主要有一个用于向服务端发送消息的sendMessage()方法，客户端向服务器发送一个 RpcRequest 对象，然后等待并获取一个 RpcResponse 对象。这是典型的请求-响应模型，也是 RPC 的基本特征。

RpcRequest 对象中包含了要调用的接口名和方法名，这些信息会被发送到服务器，服务器根据这些信息找到对应的方法并执行，然后将结果返回给客户端。这就是所谓的远程过程调用。

客户端使用 Kryo 序列化库将 RpcRequest 对象序列化为字节流，然后通过网络发送到服务器；服务器接收到字节流后，再将其反序列化为 RpcRequest 对象。这是 RPC 中的常见做法，因为网络通信只能传输字节流。

public class NettyClient {private static final Logger logger = LoggerFactory.getLogger(NettyClient.class);// 创建日志记录器private final String host;// 服务器的主机名private final int port;// 服务器的端口号private static final Bootstrap b;//Bootstrap用于Netty客户端程序的启动和配置。public NettyClient(String host, int port) {this.host = host;this.port = port;}// 初始化相关资源比如 EventLoopGroup, Bootstrapstatic {EventLoopGroup eventLoopGroup = new NioEventLoopGroup();// 创建处理 I/O 操作的多线程事件循环组b = new Bootstrap();// 创建 Bootstrap 实例，用于配置和启动 Netty 客户端KryoSerializer kryoSerializer = new KryoSerializer();// 创建 Kryo 序列化工具实例// 配置 Bootstrapb.group(eventLoopGroup)// 设置事件循环组.channel(NioSocketChannel.class)// 设置用于创建 Channel 的类.handler(new LoggingHandler(LogLevel.INFO))// 添加日志处理器// 连接的超时时间，超过这个时间还是建立不上的话则代表连接失败//  如果 15 秒之内没有发送数据给服务端的话，就发送一次心跳请求.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000).handler(new ChannelInitializer<SocketChannel>() { // 添加 Channel 初始化器@Overrideprotected void initChannel(SocketChannel ch) {// 添加 Channel 初始化器// 自定义序列化编解码器// ByteBuf -> RpcResponsech.pipeline().addLast(new NettyKryoDecoder(kryoSerializer, RpcResponse.class));// RpcRequest -> ByteBufch.pipeline().addLast(new NettyKryoEncoder(kryoSerializer, RpcRequest.class));// 添加自定义的 ChannelHandlerch.pipeline().addLast(new NettyClientHandler());}});}/*** 发送消息到服务端** @param rpcRequest 消息体* @return 服务端返回的数据*/public RpcResponse sendMessage(RpcRequest rpcRequest) {try {ChannelFuture f = b.connect(host, port).sync();// 连接到服务器，并获取 ChannelFuture 对象logger.info("client connect  {}", host + ":" + port);// 记录连接信息Channel futureChannel = f.channel();// 获取 Channel 这个对象代表了和服务器的连接。logger.info("send message");// 记录发送消息的信息if (futureChannel != null) {futureChannel.writeAndFlush(rpcRequest).addListener(future -> {// 向服务器发送消息，并添加监听器处理发送结果if (future.isSuccess()) {// 如果消息发送成功，记录发送的消息logger.info("client send message: [{}]", rpcRequest.toString());} else {// 如果消息发送失败，记录失败的原因logger.error("Send failed:", future.cause());}});futureChannel.closeFuture().sync();// 等待 Channel 关闭AttributeKey<RpcResponse> key = AttributeKey.valueOf("rpcResponse");// 从Channel的属性中获取服务器返回的RpcResponse对象return futureChannel.attr(key).get();}} catch (InterruptedException e) {logger.error("occur exception when connect server:", e);}return null;// 如果无法获取服务器返回的数据，返回 null}public static void main(String[] args) {RpcRequest rpcRequest = RpcRequest.builder().interfaceName("interface").methodName("hello").build();NettyClient nettyClient = new NettyClient("127.0.0.1", 8889);// 创建 NettyClient 对象，设置服务器的 IP 地址和端口号for (int i = 0; i < 3; i++) {// 向服务器发送 3 次相同的请求nettyClient.sendMessage(rpcRequest);}RpcResponse rpcResponse = nettyClient.sendMessage(rpcRequest);// 再次向服务器发送请求，并获取服务器返回的数据System.out.println(rpcResponse.toString());}
}

sendMessage() 方法分析：

1. 首先初始化了一个 Bootstrap
2. 通过Bootstrap 对象连接服务端
3. 通过 Channel 向服务端发送消息RpcRequest
4. 发送成功后，阻塞等待 ，直到Channel 关闭
5. 拿到服务端返回的结果RpcResponse

代码中用到了自定义 ChannelHandler 处理服务端消息，其代码如下：

public class NettyClientHandler extends ChannelInboundHandlerAdapter {private static final Logger logger = LoggerFactory.getLogger(NettyClientHandler.class);// 创建日志记录器@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {// 当从服务器接收到一条消息时被调用try {RpcResponse rpcResponse = (RpcResponse) msg;// 将接收到的消息转换为 RpcResponselogger.info("client receive msg: [{}]", rpcResponse.toString());// 声明一个 AttributeKey 对象 在Netty中每个Channel都可以有一些与之关联的属性 这些属性可以通过 AttributeKey来访问。AttributeKey<RpcResponse> key = AttributeKey.valueOf("rpcResponse");// 将服务端的返回结果保存到 AttributeMap 上，AttributeMap 可以看作是一个Channel的共享数据源// AttributeMap的key是AttributeKey，value是Attributectx.channel().attr(key).set(rpcResponse);ctx.channel().close();// 关闭 Channel} finally {ReferenceCountUtil.release(msg);// 释放接收到的消息}}// 当出现 Throwable 对象才会被调用，即当 Netty 由于 IO 错误或者处理器在处理事件时抛出的异常时@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {logger.error("client caught exception", cause);ctx.close();}
}

NettyClientHandler 用于读取服务端发送过来的 RpcResponse 消息对象，并将 RpcResponse 消息对象保存到 AttributeMap 上， AttributeMap 可以看作是一个Channel 的共享数据源。这样的话，我们就能通过channel 和key 将数据读取出来。

AttributeKey<RpcResponse> key = AttributeKey.valueOf("rpcResponse");
return futureChannel.attr(key).get();

AttributeMap 是一个接口，类似于 Map 数据结构 。

public interface AttributeMap {<T> Attribute<T> attr(AttributeKey<T> key);<T> boolean hasAttr(AttributeKey<T> key);
}

Channel 实现了 AttributeMap 接口，这样也就表明它存在了AttributeMap 相关的属性。 每个Channel 上的AttributeMap 属于共享数据。AttributeMap 的结构，和Map 很像，我们可以把key看作是AttributeKey ， value 看作是Attribute ，所以我们可以根据AttributeKey 找到对应的Attribute 。

public interface Channel extends AttributeMap, ChannelOutboundInvoker,
Comparable<Channel> {......
}

ChannelHandlerContext对象我们在之前的NettyClientHandler也看到过。ChannelHandlerContext 是 Netty 中的一个重要组件，它代表 ChannelHandler 和 ChannelPipeline 之间的关联。每当有 ChannelHandler 添加到 ChannelPipeline，都会创建 ChannelHandlerContext

ChannelHandlerContext 的主要功能是管理它所关联的 ChannelHandler 和在同一个 ChannelPipeline 中的其他 ChannelHandler 之间的交互。例如，你可以通过 ChannelHandlerContext 来触发各种 I/O 事件和操作

此外，ChannelHandlerContext 还提供了一种方式来访问关联的 ChannelHandler 所在的 ChannelPipeline 的 Channel。这意味着，你可以通过 ChannelHandlerContext 来获取 Channel，并且对 Channel 进行读写操作

服务端代码

NettyServer 主要作用就是开启了一个服务端用于接受客户端的请求并处理。

public class NettyServer {private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);// 创建日志记录器private final int port;// 服务器的端口号private NettyServer(int port) {this.port = port;}private void run() {// 运行服务器// 创建两个处理 I/O 操作的多线程事件循环组EventLoopGroup bossGroup = new NioEventLoopGroup(); // 处理连接请求EventLoopGroup workerGroup = new NioEventLoopGroup();// 处理网络读写KryoSerializer kryoSerializer = new KryoSerializer();// 创建 Kryo 序列化工具实例try {ServerBootstrap b = new ServerBootstrap();//ServerBootstrap用于Netty服务端程序的启动和配置// 配置 ServerBootstrapb.group(bossGroup, workerGroup)// 设置事件循环组.channel(NioServerSocketChannel.class)// 设置用于创建 Channel 的类// TCP默认开启了 Nagle 算法，该算法的作用是尽可能的发送大数据快，减少网络传输。TCP_NODELAY 参数的作用就是控制是否启用 Nagle 算法。.childOption(ChannelOption.TCP_NODELAY, true)// 是否开启 TCP 底层心跳机制.childOption(ChannelOption.SO_KEEPALIVE, true)//表示系统用于临时存放已完成三次握手的请求的队列的最大长度,如果连接建立频繁，服务器处理创建新连接较慢，可以适当调大这个参数.option(ChannelOption.SO_BACKLOG, 128).handler(new LoggingHandler(LogLevel.INFO))// 添加日志处理器.childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) {// 添加 Channel 初始化器// 自定义序列化编解码器// ByteBuf -> RpcRequestch.pipeline().addLast(new NettyKryoDecoder(kryoSerializer, RpcRequest.class));// RpcResponse -> ByteBufch.pipeline().addLast(new NettyKryoEncoder(kryoSerializer, RpcResponse.class));// 添加自定义的 ChannelHandlerch.pipeline().addLast(new NettyServerHandler());}});// 绑定端口，同步等待绑定成功ChannelFuture f = b.bind(port).sync();// 等待服务端监听端口关闭f.channel().closeFuture().sync();} catch (InterruptedException e) {logger.error("occur exception when start server:", e);} finally {// 关闭 EventLoopGroup，释放资源bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}public static void main(String[] args) {new NettyServer(8889).run();}}

NettyServerHandler 用于接收客户端发送过来的消息并返回结果给客户端。

public class NettyServerHandler extends ChannelInboundHandlerAdapter {private static final Logger logger = LoggerFactory.getLogger(NettyServerHandler.class);// 创建日志记录器private static final AtomicInteger atomicInteger = new AtomicInteger(1);// 创建原子整数，用于记录接收到的消息数量@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {// 当从客户端接收到一条消息时被调用try {RpcRequest rpcRequest = (RpcRequest) msg;// 将接收到的消息转换为 RpcRequestlogger.info("server receive msg: [{}] ,times:[{}]", rpcRequest, atomicInteger.getAndIncrement());// 记录接收到的消息和消息数量RpcResponse messageFromServer = RpcResponse.builder().message("message from server").build();// 创建一个 RpcResponse 对象ChannelFuture f = ctx.writeAndFlush(messageFromServer);// 将 RpcResponse 对象写入到 Channel 中，并刷新 Channelf.addListener(ChannelFutureListener.CLOSE);// 添加监听器，在写入操作完成后关闭 Channel} finally {ReferenceCountUtil.release(msg);// 释放接收到的消息}}// 当出现 Throwable 对象才会被调用，即当 Netty 由于 IO 错误或者处理器在处理事件时抛出的异常时@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {logger.error("server catch exception", cause);ctx.close();}
}

其中

ChannelFuture f = b.bind(port).sync();

这行代码的作用是启动服务器并绑定到指定的端口。b.bind(port) 是异步操作，它会立即返回一个 ChannelFuture 对象，表示绑定操作的结果。sync() 方法会阻塞当前线程，直到绑定操作完成。如果绑定成功，服务器就可以开始接收客户端的连接请求了。

f.channel().closeFuture().sync();

这行代码的作用是等待服务器的关闭。closeFuture() 方法会返回一个表示 Channel 关闭的 Future 对象。sync() 方法会阻塞当前线程，直到 Channel 关闭。这样做的目的是让服务器保持运行状态，不会因为主线程退出(比如手动停止)而立即关闭。

编码器

自定义编码器

NettyKryoEncoder 是我们自定义的编码器。它负责处理"出站"消息，将消息格式转换为字节数组然后写入到字节数据的容器 ByteBuf 对象中。

@AllArgsConstructor
public class NettyKryoEncoder extends MessageToByteEncoder<Object> {private final Serializer serializer;// 序列化工具 在服务端和客户端传入进来的是KryoSerializerprivate final Class<?> genericClass;// 需要序列化的类 RpcResponse.class或者RpcRequest.class/*** 将对象转换为字节码然后写入到 ByteBuf 对象中*/@Overrideprotected void encode(ChannelHandlerContext channelHandlerContext, Object o, ByteBuf byteBuf) {if (genericClass.isInstance(o)) {// 检查 o 是否是 genericClass 的实例// 1. 将对象转换为bytebyte[] body = serializer.serialize(o);// 2. 读取消息的长度int dataLength = body.length;// 3.写入消息对应的字节数组长度,writerIndex 加 4byteBuf.writeInt(dataLength);//4.将字节数组写入 ByteBuf 对象中byteBuf.writeBytes(body);}}
}

自定义解码器

NettyKryoDecoder 是我们自定义的解码器。它负责处理"入站"消息，它会从ByteBuf 中读取到业务对象对应的字节序列，然后再将字节序列转换为我们的业务对象。

@AllArgsConstructor
@Slf4j
public class NettyKryoDecoder extends ByteToMessageDecoder {private final Serializer serializer;// 序列化工具 在服务端和客户端传入进来的是KryoSerializerprivate final Class<?> genericClass;// 需要序列化的类 RpcResponse.class或者RpcRequest.class/*** Netty传输的消息长度也就是对象序列化后对应的字节数组的大小，存储在 ByteBuf 头部* 因为字节数组大小用int类型存储，所以这里的值是4*/private static final int BODY_LENGTH = 4;/*** 解码 ByteBuf 对象** @param ctx 解码器关联的 ChannelHandlerContext 对象* @param in  "入站"数据，也就是 ByteBuf 对象* @param out 解码之后的数据对象需要添加到 out 对象里面*/@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {//1.byteBuf中写入的消息长度所占的字节数已经是4了，所以 byteBuf 的可读字节必须大于 4，if (in.readableBytes() >= BODY_LENGTH) {//2.标记当前readIndex的位置，以便后面重置readIndex 的时候使用in.markReaderIndex();//3.读取消息的长度//注意： 消息长度是encode的时候我们自己写入的，参见 NettyKryoEncoder 的encode方法int dataLength = in.readInt();//4.遇到不合理的情况直接 returnif (dataLength < 0 || in.readableBytes() < 0) {log.error("data length or byteBuf readableBytes is not valid");return;}//5.如果可读字节数小于消息长度的话，说明是不完整的消息，重置readIndexif (in.readableBytes() < dataLength) {in.resetReaderIndex();return;}// 6.走到这里说明没什么问题了，可以序列化了byte[] body = new byte[dataLength];in.readBytes(body);// 将bytes数组转换为我们需要的对象Object obj = serializer.deserialize(body, genericClass);out.add(obj);log.info("successful decode ByteBuf to Object");}}
}

在代码注释里写到"5.如果可读字节数小于消息长度的话，说明是不完整的消息，重置readIndex"
这段代码我感觉重置readIndex其实没有什么意义，因为下面就是return了。readIndex是ByteBuf的属性，不同的ByteBuf也不是共享同一个readIndex
在网络传输中，由于各种原因，可能会出现粘包和拆包的情况。粘包是指多个包被一起发送，而拆包是指一个包被拆分成多个部分发送。这就可能导致在读取数据时，一次读取的数据实际上包含了多个消息，或者一个消息的部分数据。
这里的代码逻辑能够解决沾包的问题，至于拆包问题，即一个消息被拆分到多个 ByteBuf 中，这通常需要在更高的层次（例如应用层——比如这里的NettyClient或者NettyClientHandler）来处理。一种常见的做法是使用一个缓冲区（buffer）来存储不完整的消息，然后在接收到新的 ByteBuf 时，将新的数据添加到缓冲区中，直到收到完整的消息。

自定义序列化接口

Serializer 接口主要有两个方法一个用于序列化，一个用户反序列化。

public interface Serializer {/*** 序列化** @param obj 要序列化的对象* @return 字节数组*/byte[] serialize(Object obj);/*** 反序列化** @param bytes 序列化后的字节数组* @param clazz 类* @param <T>* @return 反序列化的对象*/<T> T deserialize(byte[] bytes, Class<T> clazz);
}

实现序列化接口

自定义 kryo 序列化实现类。

public class KryoSerializer implements Serializer {/*** 由于 Kryo 不是线程安全的。每个线程都应该有自己的 Kryo，Input 和 Output 实例。* 所以，使用 ThreadLocal 存放 Kryo 对象*/private static final ThreadLocal<Kryo> kryoThreadLocal = ThreadLocal.withInitial(() -> {Kryo kryo = new Kryo();// 创建 Kryo 实例kryo.register(RpcResponse.class);// 注册需要序列化和反序列化的类kryo.register(RpcRequest.class);kryo.setReferences(true);//默认值为true,是否关闭注册行为,关闭之后可能存在序列化问题，一般推荐设置为 truekryo.setRegistrationRequired(false);//默认值为false,是否关闭循环引用，可以提高性能，但是一般不推荐设置为 truereturn kryo;});@Overridepublic byte[] serialize(Object obj) {try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();Output output = new Output(byteArrayOutputStream)) {Kryo kryo = kryoThreadLocal.get();// 从 ThreadLocal 中获取 Kryo 实例// Object->byte:将对象序列化为byte数组kryo.writeObject(output, obj);kryoThreadLocal.remove();// 从 ThreadLocal 中移除 Kryo 实例return output.toBytes();// 返回字节数组} catch (Exception e) {throw new SerializeException("序列化失败");}}@Overridepublic <T> T deserialize(byte[] bytes, Class<T> clazz) {try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);Input input = new Input(byteArrayInputStream)) {Kryo kryo = kryoThreadLocal.get();// 从 ThreadLocal 中获取 Kryo 实例// byte->Object:从byte数组中反序列化出对对象Object o = kryo.readObject(input, clazz);kryoThreadLocal.remove();// 从 ThreadLocal 中移除 Kryo 实例return clazz.cast(o);} catch (Exception e) {throw new SerializeException("反序列化失败");}}
}

函数里每次序列化或者反序列之后都会从 ThreadLocal 中移除 Kryo 实例，为什么？


ThreadLocal 是一种线程封闭技术，可以为每个线程提供一个独立的变量副本。但是，ThreadLocal 有一个特性，那就是它不会自动清理线程结束后的数据。如果不手动清理，那么这些数据将一直存在于 ThreadLocal 中，占用内存，这可能会导致内存泄漏。


所以，为了避免内存泄漏，我们在每次使用完 Kryo 实例后，都应该调用 kryoThreadLocal.remove(); 来清理数据。而我们在创建ThreadLocal时提供了一个 InitialValue->ThreadLocal.withInitialwithInitial(…)，所以，当我们从 ThreadLocal 中获取 Kryo 实例时，如果当前线程的 Kryo 实例不存在（例如第一次获取，或者已经被移除），ThreadLocal 就会自动来创建一个新的 Kryo 实例。

自定义序列化异常类 SerializeException 如下：

public class SerializeException extends RuntimeException {public SerializeException(String message) {super(message);}
}

启动服务端后启动客户端，可在各自的控制台看到输出，案例完毕。