HTTPS、WebSocket的添加使用和大型数据写入以及几种常见的序列化)
一、基于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 {
// 做点什么
}
}
}