预置的ChannelHandler和编解码器（二）HTTPS、WebSocket的添加使用和大型数据写入以及几种常见的序列化

一、基于Netty的HTTPS程序
- 1.2 使用HTTPS
- 2.3 WebSocket
二、空闲连接和超时
三、解码基于分隔符的协议和基于长度的协议
- 3.1 基于分割符的协议
- 3.2 基于长度的协议
四、写大型数据
五、序列化数据
- 5.1 JDK序列化
- 5.2 使用JBoss Marshalling进行序列化
- 5.3 使用Protocol Buffers序列化

一、基于Netty的HTTPS程序

1.2 使用HTTPS

我们接着上一篇继续进行，启用 HTTPS 只需要将 SslHandler 添加到 ChannelPipeline 的ChannelHandler 组合中。

下面代码展示了这一过程：

import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslHandler;

import javax.net.ssl.SSLEngine;

/**
 * Author: lhd
 * Data: 2023/6/11
 * Annotate: HTTPS 使用
 */
public class HttpsCodecInitializer extends ChannelInitializer<Channel> {
    private final SslContext context;
    private final boolean isClient;
    public HttpsCodecInitializer(SslContext context, boolean isClient) {
        this.context = context;
        this.isClient = isClient;
    }
    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        SSLEngine engine = context.newEngine(ch.alloc());
        //将 SslHandler 添加到ChannelPipeline 中以使用 HTTPS
        pipeline.addFirst("ssl", new SslHandler(engine));
        if (isClient) {
            //如果是客户端，则添加 HttpClientCodec
            pipeline.addLast("codec", new HttpClientCodec());
        } else {
            //如果是服务器，则添加 HttpServerCodec
            pipeline.addLast("codec", new HttpServerCodec());
        }
    }
}

2.3 WebSocket

WebSocket有什么特点？
WebSocket解决了一个长期存在的问题：既然底层的协议（HTTP）是一个请求/响应模式的交互序列，那么如何实时地发布信息呢？AJAX提供了一定程度上的改善，但是数据流仍然是由客户端所发送的请求驱动的。而
WebSocket为网页和远程服务器之间的双向通信提供了一种替代HTTP轮询的方案。”，在客户端和服务器之间提供了真正的双向数据交换。

WebSocket协议图示：
在这里插入图片描述

要想向应用程序中添加对于 WebSocket 的支持，需要将适当的客户端或者服务器WebSocket ChannelHandler 添加到 ChannelPipeline 中。这个类将处理由 WebSocket 定义的称为帧的特殊消息类型。像WebSocketFrame 数据类型表展示的这样，WebSocketFrame 可以被归类为数据帧或者控制帧。

WebSocketFrame 数据类型表：

名称	描述
BinaryWebSocketFrame	数据帧：二进制数据
TextWebSocketFrame	数据帧：文本数据
ContinuationWebSocketFrame	数据帧：属于上一个 BinaryWebSocketFrame 或者 TextWebSocketFrame 的文本的或者二进制数据
CloseWebSocketFrame	控制帧：一个 CLOSE 请求、关闭的状态码以及关闭的原因
PingWebSocketFrame	控制帧：请求一个 PongWebSocketFrame
PongWebSocketFrame	控制帧：对 PingWebSocketFrame 请求的响应

如何在服务器支持WebSocket呢？看下面的示例：

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;

/**
 * Author: lhd
 * Data: 2023/6/11
 * Annotate: WebSocket支持
 */
public class WebSocketServerInitializer extends ChannelInitializer<Channel> {

    @Override
    protected void initChannel(Channel ch) throws Exception {
        ch.pipeline().addLast(
                new HttpServerCodec(),
                //为握手提供聚合的 HttpRequest
                new HttpObjectAggregator(65536),
                //如果被请求的端点是"/websocket"，则升级握手
                new WebSocketServerProtocolHandler("/websocket"),
                //TextFrameHandler 处理 TextWebSocketFrame
                new TextFrameHandler(),
                //BinaryFrameHandler 处理 BinaryWebSocketFrame
                new BinaryFrameHandler(),
                //ContinuationFrameHandler 处理 ContinuationWebSocketFrame
                new ContinuationFrameHandler());
    }
    public static final class TextFrameHandler extends
            SimpleChannelInboundHandler<TextWebSocketFrame> {
        @Override
        public void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
            //处理文本框
        }
    }
    public static final class BinaryFrameHandler extends SimpleChannelInboundHandler<BinaryWebSocketFrame> {
        @Override
        public void channelRead0(ChannelHandlerContext ctx, BinaryWebSocketFrame msg) throws Exception {
          // 处理二进制帧
        }
    }
    public static final class ContinuationFrameHandler extends
            SimpleChannelInboundHandler<ContinuationWebSocketFrame> {
        @Override
        public void channelRead0(ChannelHandlerContext ctx, ContinuationWebSocketFrame msg) throws Exception {
          //逻辑
        }
    }
}

保护 WebSocket
要想为 WebSocket 添加安全性，只需要将 SslHandler 作为第一个 ChannelHandler 添加到 ChannelPipeline 中。

二、空闲连接和超时

我们说了 Netty 通过专门的编解码器和处理器对 HTTP 的变型 HTTPS 和 WebSocket的支持。只要能有效地管理网络资源，这些技术就可以使得应用程序更加高效、易用和安全。下面我们说说连接管理。

检测空闲连接以及超时对于及时释放资源来说是至关重要的。由于这是一项常见的任务，Netty 特地为它提供了几个 ChannelHandler 实现。

用于空闲连接以及超时的 ChannelHandler：

名称	描述
IdleStateHandler	当连接空闲时间太长时，将会触发一个 IdleStateEvent 事件。然后，可以通过在 ChannelInboundHandler 中重写 userEventTriggered()方法来处理该 IdleStateEvent 事件
ReadTimeoutHandler	如果在指定的时间间隔内没有收到任何的入站数据，则抛出一个 ReadTimeoutException 并关闭对应的Channel。可以通过重写ChannelHandler 中的 exceptionCaught()方法来检测该 ReadTimeoutException
WriteTimeoutHandler	如果在指定的时间间隔内没有任何出站数据写入，则抛出一个 WriteTimeoutException 并关闭对应的 Channel 。可以通过重写ChannelHandler 的 exceptionCaught()方法检测该 WriteTimeoutException

以上三个ChannelHandler的实现，在实践中使用最多的应该是IdleStateHandler。下面展示一下使用发送心跳消息到远程节点的方法时，如果在 60 秒之内没有接收或者发送任何的数据，我们将如何得到通知；如果没有响应，则连接会被关闭。

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.CharsetUtil;

import java.util.concurrent.TimeUnit;

/**
 * Author: lhd
 * Data: 2023/6/11
 * Annotate: 心跳发送
 */
public class IdleStateHandlerInitializer extends ChannelInitializer<Channel> {
    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        //IdleStateHandler 将在被触发时发送一个 IdleStateEvent 事件
        pipeline.addLast(new IdleStateHandler(0, 0, 60, TimeUnit.SECONDS));
        //将一个HeartbeatHandler添加到 ChannelPipeline中
        pipeline.addLast(new HeartbeatHandler());
    }

    //实现 userEventTriggered()方法以发送心跳消息
    public static final class HeartbeatHandler extends ChannelInboundHandlerAdapter {

        //发送到远程节点的心跳消息
        private static final ByteBuf HEARTBEAT_SEQUENCE =
                Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("HEARTBEAT", CharsetUtil.ISO_8859_1));
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            
            //发送心跳消息，并在发送失败时关闭该连接
            if (evt instanceof IdleStateEvent) {
                ctx.writeAndFlush(HEARTBEAT_SEQUENCE.duplicate())
                        .addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            } else {
                //不是 IdleStateEvent 事件，所以将它传递给下一个 ChannelInboundHandler
                super.userEventTriggered(ctx, evt);
            }
        }
    }
}

三、解码基于分隔符的协议和基于长度的协议

3.1 基于分割符的协议

什么是基于分隔符的协议？
基于分隔符的（delimited）消息协议使用定义的字符来标记的消息或者消息段（通常被称为帧）的开头或者结尾。由RFC文档正式定义的许多协议（如SMTP、POP3、IMAP以及Telnet）都是这样的。此外，还有一些自定义的类似协议。

下图展示了基于分割符的协议：
在这里插入图片描述

但无论使用什么样的协议，下表列出的解码器都能帮助我们定义可以提取由任意标记（token）序列分隔的帧的自定义解码器。

名称	描述
DelimiterBasedFrameDecoder	使用任何由用户提供的分隔符来提取帧的通用解码器
LineBasedFrameDecoder	提取由行尾符（\n 或者\r\n）分隔的帧的解码器。这个解码器比 DelimiterBasedFrameDecoder 更快

我们示例一下了当出现帧由行尾序列\r\n（回车符+换行符）分隔时是如何处理的：

import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.handler.codec.LineBasedFrameDecoder;

/**
 * Author: lhd
 * Data: 2023/6/11
 * Annotate: 处理当帧由行尾序列\r\n（回车符+换行符）分隔时的数据
 */
public class LineBasedHandlerInitializer extends ChannelInitializer<Channel> {
    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        //该 LineBasedFrameDecoder将提取的帧转发给下一个 ChannelInboundHandler
        pipeline.addLast(new LineBasedFrameDecoder(64 * 1024));
        //添加 FrameHandler以接收帧
        pipeline.addLast(new FrameHandler());
    }
    public static final class FrameHandler extends SimpleChannelInboundHandler<ByteBuf> {
        
        //传入了单个帧的内容
        @Override
        public void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
             // 对从帧中提取的数据执行操作
        }
    }
}

但如果接收数据不是以上示例的符号结尾咋办？
可以以类似的方式使用 DelimiterBasedFrameDecoder，只需要将特定的分隔符序列指定到其构造函数即可。

3.2 基于长度的协议

什么是基于长度的协议？
基于长度的协议通过将它的长度编码到帧的头部来定义帧，而不是使用特殊的分隔符来标记它的结束。

下图展示了基于长度的协议（帧长度为8字节）：
在这里插入图片描述

下表列出了可以用于处理这种协议的两种解码器：

名称	描述
FixedLengthFrameDecoder	提取在调用构造函数时指定的定长帧
LengthFieldBasedFrameDecoder	根据编码进帧头部中的长度值提取帧；该字段的偏移量以及长度在构造函数中指定

LengthFieldBasedFrameDecoder 提供了几个构造函数来支持各种各样的头部配置情况。我们展示下如何使用其 3 个构造参数分别为 maxFrameLength、lengthFieldOffset 和 lengthFieldLength 的构造函数：

import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;

/**
 * Author: lhd
 * Data: 2023/6/11
 * Annotate: 使用 LengthFieldBasedFrameDecoder 解码器基于长度的协议
 */
public class LengthBasedInitializer extends ChannelInitializer<Channel> {

    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        //使用 LengthFieldBasedFrameDecoder 解码将帧长度编码到帧起始的前 8 个字节中的消息
        pipeline.addLast(new LengthFieldBasedFrameDecoder(64 * 1024, 0, 8));
        //添加 FrameHandler以处理每个帧
        pipeline.addLast(new FrameHandler());
    }
    public static final class FrameHandler extends SimpleChannelInboundHandler<ByteBuf> {

        //处理帧的数据
        @Override
        public void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
            // 做点什么
        }
    }
}

四、写大型数据

因为网络饱和的可能性，如何在异步框架中高效地写大块的数据是一个特殊的问题。由于写操作是非阻塞的，所以即使没有写出所有的数据，写操作也会在完成时返回并通知 ChannelFuture。当这种情况发生时，如果仍然不停地写入，就有内存耗尽的风险。

我们在写大型数据时，需要准备好处理到远程节点的连接是慢速连接的情况，这种情况会导致内存释放的延迟。让我们考虑下将一个文件内容写出到网络的情况。

我们之前提到 NIO 的零拷贝特性，这种特性消除了将文件的内容从文件系统移动到网络栈的复制过程。所有的这一切都发生在 Netty 的核心中，所以应用程序所有需要做的就是使用一个 FileRegion 接口的实现，其在Netty 的 API 文档中的定义是：“通过支持零拷贝的文件传输的 Channel 来发送的文件区域。”

我们演示一下展示了如何通过从FileInputStream创建一个DefaultFileRegion，并将其写，从而利用零拷贝特性来传输一个文件的内容。

          //创建一个FileInputStream
            FileInputStream in = new FileInputStream(file);
            //以该文件的完整长度创建一个新的 DefaultFileRegion
            FileRegion region = new DefaultFileRegion(in.getChannel(), 0, file.length());
            //发送该 DefaultFileRegion，并注册一个ChannelFutureListener
            channel.writeAndFlush(region).addListener(
                    new ChannelFutureListener() {
                        @Override
                        public void operationComplete(ChannelFuture future) throws Exception {
                            //处理失败
                            if (!future.isSuccess()) {
                                Throwable cause = future.cause();
                                // 做点什么
                            }
                        }
                    });

上面的代码只适用于文件内容的直接传输，不包括应用程序对数据的任何处理。在需要将数据从文件系统复制到用户内存中时，可以使用ChunkedWriteHandler，它支持异步写大型数据流，而又不会导致大量的内存消耗。

一部写大数据可以考虑：ChunkedWriteHandler，它的的关键是 interface ChunkedInput <> （这个里面是B，不知道咋回事把B写进<>，整段话的格式就会改变😂），<>中的类型参数是 readChunk()方法返回的类型，当 Channel 的状态变为活动的时，WriteStreamHandler 将会逐块地把来自文件中的数据作为 ChunkedStream 写入。数据在传输之前将会由 SslHandler 加密。

我们来看一下ChunkedInput 的实现：

名称	描述
ChunkedFile	从文件中逐块获取数据，当你的平台不支持零拷贝或者你需要转换数据时使用
ChunkedNioFile	和 ChunkedFile 类似，只是它使用了 FileChannel
ChunkedStream	从 InputStream 中逐块传输内容
ChunkedNioStream	从 ReadableByteChannel 中逐块传输内容

大部分情况下，我们都是在使用ChunkedStream 来进行数据传输，下面我们看一下使用 ChunkedStream 传输文件内容：

import io.netty.channel.*;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.stream.ChunkedStream;
import io.netty.handler.stream.ChunkedWriteHandler;

import java.io.File;
import java.io.FileInputStream;

/**
 * Author: lhd
 * Data: 2023/6/11
 * Annotate:使用 ChunkedStream 传输文件内容
 */
public class ChunkedWriteHandlerInitializer extends ChannelInitializer<Channel> {
    private final File file;
    private final SslContext sslCtx;

    public ChunkedWriteHandlerInitializer(File file, SslContext sslCtx) {
        this.file = file;
        this.sslCtx = sslCtx;
    }
    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        // SslHandler 添加到ChannelPipeline 中
        pipeline.addLast(new SslHandler(sslCtx.newEngine(ch.alloc())));
        //添加 ChunkedWriteHandler以处理作为ChunkedInput传入的数据
        pipeline.addLast(new ChunkedWriteHandler());
        //一旦连接建立，WriteStreamHandler就开始写文件数据
        pipeline.addLast(new WriteStreamHandler());
    }
    public final class WriteStreamHandler extends ChannelInboundHandlerAdapter {
        
        //当连接建立时，channelActive()方法将使用ChunkedInput写文件数据
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            super.channelActive(ctx);
            ctx.writeAndFlush(new ChunkedStream(new FileInputStream(file)));
        }
    }
}

逐块输入
要使用你自己的 ChunkedInput 实现，在 ChannelPipeline 中安装一个ChunkedWriteHandler去做

五、序列化数据

前面我们说了如何通过使用零拷贝特性来高效地传输文件，以及如何通过使用ChunkedWriteHandler 来写大型数据而又不必冒着导致 OutOfMemoryError 的风险。下来我们说究几种序列化 POJO 的方法。

5.1 JDK序列化

JDK 提供了 ObjectOutputStream 和 ObjectInputStream，用于通过网络对 POJO 的基本数据类型和图进行序列化和反序列化。该 API 并不复杂，而且可以被应用于任何实现了java.io.Serializable接口的对象。但是它的性能也不是非常高效的。

如果你的应用程序必须要和使用了ObjectOutputStream和ObjectInputStream的远程节点交互，并且兼容性也是你最关心的，那么JDK序列化将是正确的选择。

这种情况下我们就要看看Netty提供的用于和JDK进行互操作的序列化类：

名称	描述
CompatibleObjectDecoder	和使用 JDK 序列化的非基于 Netty 的远程节点进行互操作的解码器
CompatibleObjectEncoder	和使用 JDK 序列化的非基于 Netty 的远程节点进行互操作的编码器
ObjectDecoder	构建于 JDK 序列化之上的使用自定义的序列化来解码的解码器；当没有其他的外部依赖时，它提供了速度上的改进。否则其他的序列化实现更加可取
ObjectEncoder	构建于 JDK 序列化之上的使用自定义的序列化来编码的编码器；当没有其他的外部依赖时，它提供了速度上的改进。否则其他的序列化实现更加可取

5.2 使用JBoss Marshalling进行序列化

如果你可以自由地使用外部依赖，那么JBoss Marshalling将是个理想的选择：它比JDK序列化最多快 3 倍，而且也更加紧凑。

JBoss Marshalling 是一种可选的序列化 API，它修复了在 JDK 序列化 API 中所发现的许多问题，同时保留了与 java.io.Serializable及其相关类的兼容性，并添加了几个新的可调优参数以及额外的特性，所有的这些都是可以通过工厂配置（如外部序列化器、类/实例查找表、类解析以及对象替换等）实现可插拔的。

Netty 提供了下表所示的两组解码器/编码器对为 Boss Marshalling 提供了支持。第一组兼容只使用 JDK 序列化的远程节点。第二组提供了最大的性能，适用于和使用 JBoss Marshalling 的远程节点一起使用。

JBoss Marshalling 编解码器：

名称	描述
CompatibleMarshallingDecoder；CompatibleMarshallingEncoder	与只使用 JDK 序列化的远程节点兼容
MarshallingDecoder；MarshallingEncoder	适用于使用 JBoss Marshalling 的节点。这些类必须一起使用

我们展示一下如何使用 MarshallingDecoder 和 MarshallingEncoder，很简单的，配置一下ChannelPipeline就ok。

import io.netty.channel.*;
import io.netty.handler.codec.marshalling.MarshallerProvider;
import io.netty.handler.codec.marshalling.MarshallingDecoder;
import io.netty.handler.codec.marshalling.MarshallingEncoder;
import io.netty.handler.codec.marshalling.UnmarshallerProvider;

import java.io.Serializable;

/**
 * Author: lhd
 * Data: 2023/6/11
 * Annotate:  使用 JBoss Marshalling
 */
public class MarshallingInitializer extends ChannelInitializer<Channel> {
    private final MarshallerProvider marshallerProvider;
    private final UnmarshallerProvider unmarshallerProvider;
    public MarshallingInitializer(UnmarshallerProvider unmarshallerProvider, MarshallerProvider marshallerProvider) {
        this.marshallerProvider = marshallerProvider;
        this.unmarshallerProvider = unmarshallerProvider;
    }
    @Override
    protected void initChannel(Channel channel) throws Exception {
        ChannelPipeline pipeline = channel.pipeline();
        //添加 MarshallingDecoder 以将 ByteBuf 转换为 POJO
        pipeline.addLast(new MarshallingDecoder(unmarshallerProvider));
        //添加 MarshallingEncoder 以将POJO转换为 ByteBuf
        pipeline.addLast(new MarshallingEncoder(marshallerProvider));
        //添加 ObjectHandler，以处理普通的实现了Serializable 接口的 POJO
        pipeline.addLast(new ObjectHandler());
    }
    public static final class ObjectHandler extends SimpleChannelInboundHandler<Serializable> {

        @Override
        public void channelRead0(ChannelHandlerContext channelHandlerContext, Serializable serializable) throws Exception {
          // 做点什么
        }
    }
}

5.3 使用Protocol Buffers序列化

Netty序列化的最后一个解决方案是利用Protocol Buffers 的编解码器，它是一种由Google公司开发的、现在已经开源的数据交换格式。

Protocol Buffers 以一种紧凑而高效的方式对结构化的数据进行编码以及解码。它具有许多的编程语言绑定，使得它很适合跨语言的项目

Protobuf 编解码器：

名称	描述
ProtobufDecoder	使用 protobuf 对消息进行解码
ProtobufEncoder	使用 protobuf 对消息进行编码
ProtobufVarint32FrameDecoder	根据消息中的 Google Protocol Buffers 的“Base 128 Varints”a整型长度字段值动态地分割所接收到的 ByteBuf
ProtobufVarint32LengthFieldPrepender	向 ByteBuf 前追加一个 Google Protocal Buffers 的“Base128 Varints”整型的长度字段值

同样的使用 protobuf 只不过是将正确的 ChannelHandler 添加到 ChannelPipeline 中：

import io.netty.channel.*;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;

/**
 * Author: lhd
 * Data: 2023/6/11
 * Annotate:使用 protobuf
 */
public class ProtoBufInitializer extends ChannelInitializer<Channel> {
    
    private final MessageLite lite;
    
    public ProtoBufInitializer(MessageLite lite) {
        this.lite = lite;
    }
    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        //添加 ProtobufVarint32FrameDecoder以分隔帧
        pipeline.addLast(new ProtobufVarint32FrameDecoder());
        //添加 ProtobufEncoder以处理消息的编码
        pipeline.addLast(new ProtobufEncoder());
        //添加 ProtobufDecoder以解码消息
        pipeline.addLast(new ProtobufDecoder(lite));
        //添加 ObjectHandler 以处理解码消息
        pipeline.addLast(new ObjectHandler());
    }
    public static final class ObjectHandler extends SimpleChannelInboundHandler<Object> {
        @Override
        public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
          // 做点什么
        }
    }
}