Netty In Action中文版 - 第八章:附带的ChannelHandler和Codec
本章介绍
- 使用SSL/TLS创建安全的Netty程序
- 使用Netty创建HTTP/HTTPS程序
- 处理空闲连接和超时
- 解码分隔符和基于长度的协议
- 写大数据
- 序列化数据
上一章讲解了如何创建自己的编解码器,我们现在可以用上一章的知识来编写自己的编解码器。不过Netty提供了一些标准的ChannelHandler和Codec。Netty提供了很多协议的支持,所以我们不必自己发明轮子。Netty提供的这些实现可以解决我们的大部分需求。本章讲解Netty中使用SSL/TLS编写安全的应用程序,编写HTTP协议服务器,以及使用如WebSocket或Google的SPDY协议来使HTTP服务获得更好的性能;这些都是很常见的应用,本章还会介绍数据压缩,在数据量比较大的时候,压缩数据是很有必要的。
8.1 使用SSL/TLS创建安全的Netty程序
- public class SslChannelInitializer extends ChannelInitializer<Channel> {
- private final SSLContext context;
- private final boolean client;
- private final boolean startTls;
- public SslChannelInitializer(SSLContext context, boolean client, boolean startTls) {
- this.context = context;
- this.client = client;
- this.startTls = startTls;
- }
- @Override
- protected void initChannel(Channel ch) throws Exception {
- SSLEngine engine = context.createSSLEngine();
- engine.setUseClientMode(client);
- ch.pipeline().addFirst("ssl", new SslHandler(engine, startTls));
- }
- }
需要注意一点,SslHandler必须要添加到ChannelPipeline的第一个位置,可能有一些例外,但是最好这样来做。回想一下之前讲解的ChannelHandler,ChannelPipeline就像是一个在处理“入站”数据时先进先出,在处理“出站”数据时后进先出的队列。最先添加的SslHandler会啊在其他Handler处理逻辑数据之前对数据进行加密,从而确保Netty服务端的所有的Handler的变化都是安全的。
- setHandshakeTimeout(long handshakeTimeout, TimeUnit unit),设置握手超时时间,ChannelFuture将得到通知
- setHandshakeTimeoutMillis(long handshakeTimeoutMillis),设置握手超时时间,ChannelFuture将得到通知
- getHandshakeTimeoutMillis(),获取握手超时时间值
- setCloseNotifyTimeout(long closeNotifyTimeout, TimeUnit unit),设置关闭通知超时时间,若超时,ChannelFuture会关闭失败
- setHandshakeTimeoutMillis(long handshakeTimeoutMillis),设置关闭通知超时时间,若超时,ChannelFuture会关闭失败
- getCloseNotifyTimeoutMillis(),获取关闭通知超时时间
- handshakeFuture(),返回完成握手后的ChannelFuture
- close(),发送关闭通知请求关闭和销毁
8.2 使用Netty创建HTTP/HTTPS程序
8.2.1 Netty的HTTP编码器,解码器和编解码器
- HttpRequestEncoder,将HttpRequest或HttpContent编码成ByteBuf
- HttpRequestDecoder,将ByteBuf解码成HttpRequest和HttpContent
- HttpResponseEncoder,将HttpResponse或HttpContent编码成ByteBuf
- HttpResponseDecoder,将ByteBuf解码成HttpResponse和HttpContent
看下面代码:
- public class HttpDecoderEncoderInitializer extends ChannelInitializer<Channel> {
- private final boolean client;
- public HttpDecoderEncoderInitializer(boolean client) {
- this.client = client;
- }
- @Override
- protected void initChannel(Channel ch) throws Exception {
- ChannelPipeline pipeline = ch.pipeline();
- if (client) {
- pipeline.addLast("decoder", new HttpResponseDecoder());
- pipeline.addLast("", new HttpRequestEncoder());
- } else {
- pipeline.addLast("decoder", new HttpRequestDecoder());
- pipeline.addLast("encoder", new HttpResponseEncoder());
- }
- }
- }
如果你需要在ChannelPipeline中有一个解码器和编码器,还分别有一个在客户端和服务器简单的编解码器:HttpClientCodec和HttpServerCodec。
在ChannelPipelien中有解码器和编码器(或编解码器)后就可以操作不同的HttpObject消息了;但是HTTP请求和响应可以有很多消息数据,你需要处理不同的部分,可能也需要聚合这些消息数据,这是很麻烦的。为了解决这个问题,Netty提供了一个聚合器,它将消息部分合并到FullHttpRequest和FullHttpResponse,因此不需要担心接收碎片消息数据。
8.2.2 HTTP消息聚合
- /**
- * 添加聚合http消息的Handler
- *
- * @author c.k
- *
- */
- public class HttpAggregatorInitializer extends ChannelInitializer<Channel> {
- private final boolean client;
- public HttpAggregatorInitializer(boolean client) {
- this.client = client;
- }
- @Override
- protected void initChannel(Channel ch) throws Exception {
- ChannelPipeline pipeline = ch.pipeline();
- if (client) {
- pipeline.addLast("codec", new HttpClientCodec());
- } else {
- pipeline.addLast("codec", new HttpServerCodec());
- }
- pipeline.addLast("aggegator", new HttpObjectAggregator(512 * 1024));
- }
- }
如上面代码,很容使用Netty自动聚合消息。但是请注意,为了防止Dos攻击服务器,需要合理的限制消息的大小。应设置多大取决于实际的需求,当然也得有足够的内存可用。
8.2.3 HTTP压缩
使用HTTP时建议压缩数据以减少传输流量,压缩数据会增加CPU负载,现在的硬件设施都很强大,大多数时候压缩数据时一个好主意。Netty支持“gzip”和“deflate”,为此提供了两个ChannelHandler实现分别用于压缩和解压。看下面代码:
- @Override
- protected void initChannel(Channel ch) throws Exception {
- ChannelPipeline pipeline = ch.pipeline();
- if (client) {
- pipeline.addLast("codec", new HttpClientCodec());
- //添加解压缩Handler
- pipeline.addLast("decompressor", new HttpContentDecompressor());
- } else {
- pipeline.addLast("codec", new HttpServerCodec());
- //添加解压缩Handler
- pipeline.addLast("decompressor", new HttpContentDecompressor());
- }
- pipeline.addLast("aggegator", new HttpObjectAggregator(512 * 1024));
- }
8.2.4 使用HTTPS
网络中传输的重要数据需要加密来保护,使用Netty提供的SslHandler可以很容易实现,看下面代码:
- /**
- * 使用SSL对HTTP消息加密
- *
- * @author c.k
- *
- */
- public class HttpsCodecInitializer extends ChannelInitializer<Channel> {
- private final SSLContext context;
- private final boolean client;
- public HttpsCodecInitializer(SSLContext context, boolean client) {
- this.context = context;
- this.client = client;
- }
- @Override
- protected void initChannel(Channel ch) throws Exception {
- SSLEngine engine = context.createSSLEngine();
- engine.setUseClientMode(client);
- ChannelPipeline pipeline = ch.pipeline();
- pipeline.addFirst("ssl", new SslHandler(engine));
- if (client) {
- pipeline.addLast("codec", new HttpClientCodec());
- } else {
- pipeline.addLast("codec", new HttpServerCodec());
- }
- }
- }
8.2.5 WebSocket
HTTP是不错的协议,但是如果需要实时发布信息怎么做?有个做法就是客户端一直轮询请求服务器,这种方式虽然可以达到目的,但是其缺点很多,也不是优秀的解决方案,为了解决这个问题,便出现了WebSocket。
- BinaryWebSocketFrame,包含二进制数据
- TextWebSocketFrame,包含文本数据
- ContinuationWebSocketFrame,包含二进制数据或文本数据,BinaryWebSocketFrame和TextWebSocketFrame的结合体
- CloseWebSocketFrame,WebSocketFrame代表一个关闭请求,包含关闭状态码和短语
- PingWebSocketFrame,WebSocketFrame要求PongWebSocketFrame发送数据
- PongWebSocketFrame,WebSocketFrame要求PingWebSocketFrame响应
为了简化,我们只看看如何使用WebSocket服务器。客户端使用可以看Netty自带的WebSocket例子。
- /**
- * WebSocket Server,若想使用SSL加密,将SslHandler加载ChannelPipeline的最前面即可
- * @author c.k
- *
- */
- public class WebSocketServerInitializer extends ChannelInitializer<Channel> {
- @Override
- protected void initChannel(Channel ch) throws Exception {
- ch.pipeline().addLast(new HttpServerCodec(),
- new HttpObjectAggregator(65536),
- new WebSocketServerProtocolHandler("/websocket"),
- new TextFrameHandler(),
- new BinaryFrameHandler(),
- new ContinuationFrameHandler());
- }
- public static final class TextFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
- @Override
- protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
- // handler text frame
- }
- }
- public static final class BinaryFrameHandler extends SimpleChannelInboundHandler<BinaryWebSocketFrame>{
- @Override
- protected void channelRead0(ChannelHandlerContext ctx, BinaryWebSocketFrame msg) throws Exception {
- //handler binary frame
- }
- }
- public static final class ContinuationFrameHandler extends SimpleChannelInboundHandler<ContinuationWebSocketFrame>{
- @Override
- protected void channelRead0(ChannelHandlerContext ctx, ContinuationWebSocketFrame msg) throws Exception {
- //handler continuation frame
- }
- }
- }
8.2.6 SPDY
SPDY(读作“SPeeDY”)是Google开发的基于TCP的应用层协议,用以最小化网络延迟,提升网络速度,优化用户的网络使用体验。SPDY并不是一种用于替代HTTP的协议,而是对HTTP协议的增强。新协议的功能包括数据流的多路复用、请求优先级以及HTTP报头压缩。谷歌表示,引入SPDY协议后,在实验室测试中页面加载速度比原先快64%。
- 将页面加载时间减少50%。
- 最大限度地减少部署的复杂性。SPDY使用TCP作为传输层,因此无需改变现有的网络设施。
- 避免网站开发者改动内容。 支持SPDY唯一需要变化的是客户端代理和Web服务器应用程序。
SPDY实现技术:
- 单个TCP连接支持并发的HTTP请求。
- 压缩报头和去掉不必要的头部来减少当前HTTP使用的带宽。
- 定义一个容易实现,在服务器端高效率的协议。通过减少边缘情况、定义易解析的消息格式来减少HTTP的复杂性。
- 强制使用SSL,让SSL协议在现存的网络设施下有更好的安全性和兼容性。
- 允许服务器在需要时发起对客户端的连接并推送数据。
SPDY具体的细节知识及使用可以查阅相关资料,这里不作赘述了。
8.3 处理空闲连接和超时
- IdleStateHandler,当一个通道没有进行读写或运行了一段时间后出发IdleStateEvent
- ReadTimeoutHandler,在指定时间内没有接收到任何数据将抛出ReadTimeoutException
- WriteTimeoutHandler,在指定时间内有写入数据将抛出WriteTimeoutException
最常用的是IdleStateHandler,下面代码显示了如何使用IdleStateHandler,如果60秒内没有接收数据或发送数据,操作将失败,连接将关闭:
- public class IdleStateHandlerInitializer extends ChannelInitializer<Channel> {
- @Override
- protected void initChannel(Channel ch) throws Exception {
- ChannelPipeline pipeline = ch.pipeline();
- pipeline.addLast(new IdleStateHandler(0, 0, 60, TimeUnit.SECONDS));
- pipeline.addLast(new HeartbeatHandler());
- }
- public static final class HeartbeatHandler extends ChannelInboundHandlerAdapter {
- private static final ByteBuf HEARTBEAT_SEQUENCE = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer(
- "HEARTBEAT", CharsetUtil.UTF_8));
- @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 {
- super.userEventTriggered(ctx, evt);
- }
- }
- }
- }
8.4 解码分隔符和基于长度的协议
使用Netty时会遇到需要解码以分隔符和长度为基础的协议,本节讲解Netty如何解码这些协议。
8.4.1 分隔符协议
- DelimiterBasedFrameDecoder,解码器,接收ByteBuf由一个或多个分隔符拆分,如NUL或换行符
- LineBasedFrameDecoder,解码器,接收ByteBuf以分割线结束,如"\n"和"\r\n"
下图显示了使用"\r\n"分隔符的处理:
- /**
- * 处理换行分隔符消息
- * @author c.k
- *
- */
- public class LineBasedHandlerInitializer extends ChannelInitializer<Channel> {
- @Override
- protected void initChannel(Channel ch) throws Exception {
- ch.pipeline().addLast(new LineBasedFrameDecoder(65 * 1204), new FrameHandler());
- }
- public static final class FrameHandler extends SimpleChannelInboundHandler<ByteBuf> {
- @Override
- protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
- // do something with the frame
- }
- }
- }
如果框架的东西除了换行符还有别的分隔符,可以使用DelimiterBasedFrameDecoder,只需要将分隔符传递到构造方法中。如果想实现自己的以分隔符为基础的协议,这些解码器是有用的。例如,现在有个协议,它只处理命令,这些命令由名称和参数形成,名称和参数由一个空格分隔,实现这个需求的代码如下:
- /**
- * 自定义以分隔符为基础的协议
- * @author c.k
- *
- */
- public class CmdHandlerInitializer extends ChannelInitializer<Channel> {
- @Override
- protected void initChannel(Channel ch) throws Exception {
- ch.pipeline().addLast(new CmdDecoder(65 * 1024), new CmdHandler());
- }
- public static final class Cmd {
- private final ByteBuf name;
- private final ByteBuf args;
- public Cmd(ByteBuf name, ByteBuf args) {
- this.name = name;
- this.args = args;
- }
- public ByteBuf getName() {
- return name;
- }
- public ByteBuf getArgs() {
- return args;
- }
- }
- public static final class CmdDecoder extends LineBasedFrameDecoder {
- public CmdDecoder(int maxLength) {
- super(maxLength);
- }
- @Override
- protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {
- ByteBuf frame = (ByteBuf) super.decode(ctx, buffer);
- if (frame == null) {
- return null;
- }
- int index = frame.indexOf(frame.readerIndex(), frame.writerIndex(), (byte) ' ');
- return new Cmd(frame.slice(frame.readerIndex(), index), frame.slice(index + 1, frame.writerIndex()));
- }
- }
- public static final class CmdHandler extends SimpleChannelInboundHandler<Cmd> {
- @Override
- protected void channelRead0(ChannelHandlerContext ctx, Cmd msg) throws Exception {
- // do something with the command
- }
- }
- }
8.4.2 长度为基础的协议
一般经常会碰到以长度为基础的协议,对于这种情况Netty有两个不同的解码器可以帮助我们来解码:
- FixedLengthFrameDecoder
- LengthFieldBasedFrameDecoder
下图显示了FixedLengthFrameDecoder的处理流程:
- public class LengthBasedInitializer extends ChannelInitializer<Channel> {
- @Override
- protected void initChannel(Channel ch) throws Exception {
- ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(65*1024, 0, 8))
- .addLast(new FrameHandler());
- }
- public static final class FrameHandler extends SimpleChannelInboundHandler<ByteBuf>{
- @Override
- protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
- //do something with the frame
- }
- }
- }
8.5 写大数据
写大量的数据的一个有效的方法是使用异步框架,如果内存和网络都处于饱满负荷状态,你需要停止写,否则会报OutOfMemoryError。Netty提供了写文件内容时zero-memory-copy机制,这种方法再将文件内容写到网络堆栈空间时可以获得最大的性能。使用零拷贝写文件的内容时通过DefaultFileRegion、ChannelHandlerContext、ChannelPipeline,看下面代码:
- @Override
- public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
- File file = new File("test.txt");
- FileInputStream fis = new FileInputStream(file);
- FileRegion region = new DefaultFileRegion(fis.getChannel(), 0, file.length());
- Channel channel = ctx.channel();
- channel.writeAndFlush(region).addListener(new ChannelFutureListener() {
- @Override
- public void operationComplete(ChannelFuture future) throws Exception {
- if(!future.isSuccess()){
- Throwable cause = future.cause();
- // do something
- }
- }
- });
- }
如果只想发送文件中指定的数据块应该怎么做呢?Netty提供了ChunkedWriteHandler,允许通过处理ChunkedInput来写大的数据块。下面是ChunkedInput的一些实现类:
- ChunkedFile
- ChunkedNioFile
- ChunkedStream
- ChunkedNioStream
看下面代码:
- public class ChunkedWriteHandlerInitializer extends ChannelInitializer<Channel> {
- private final File file;
- public ChunkedWriteHandlerInitializer(File file) {
- this.file = file;
- }
- @Override
- protected void initChannel(Channel ch) throws Exception {
- ch.pipeline().addLast(new ChunkedWriteHandler())
- .addLast(new WriteStreamHandler());
- }
- public final class WriteStreamHandler extends ChannelInboundHandlerAdapter {
- @Override
- public void channelActive(ChannelHandlerContext ctx) throws Exception {
- super.channelActive(ctx);
- ctx.writeAndFlush(new ChunkedStream(new FileInputStream(file)));
- }
- }
- }
8.6 序列化数据
开发网络程序过程中,很多时候需要传输结构化对象数据POJO,Java中提供了ObjectInputStream和ObjectOutputStream及其他的一些对象序列化接口。Netty中提供基于JDK序列化接口的序列化接口。
8.6.1 普通的JDK序列化
- CompatibleObjectEncoder
- CompactObjectInputStream
- CompactObjectOutputStream
- ObjectEncoder
- ObjectDecoder
- ObjectEncoderOutputStream
- ObjectDecoderInputStream
8.6.2 通过JBoss编组序列化
如果你想使用外部依赖的接口,JBoss编组是个好方法。JBoss Marshalling序列化的速度是JDK的3倍,并且序列化的结构更紧凑,从而使序列化后的数据更小。Netty附带了JBoss编组序列化的实现,这些实现接口放在io.netty.handler.codec.marshalling包下面:
- CompatibleMarshallingEncoder
- CompatibleMarshallingDecoder
- MarshallingEncoder
- MarshallingDecoder
看下面代码:
- /**
- * 使用JBoss Marshalling
- * @author c.k
- *
- */
- public class MarshallingInitializer extends ChannelInitializer<Channel> {
- private final MarshallerProvider marshallerProvider;
- private final UnmarshallerProvider unmarshallerProvider;
- public MarshallingInitializer(MarshallerProvider marshallerProvider, UnmarshallerProvider unmarshallerProvider) {
- this.marshallerProvider = marshallerProvider;
- this.unmarshallerProvider = unmarshallerProvider;
- }
- @Override
- protected void initChannel(Channel ch) throws Exception {
- ch.pipeline().addLast(new MarshallingDecoder(unmarshallerProvider))
- .addLast(new MarshallingEncoder(marshallerProvider))
- .addLast(new ObjectHandler());
- }
- public final class ObjectHandler extends SimpleChannelInboundHandler<Serializable> {
- @Override
- protected void channelRead0(ChannelHandlerContext ctx, Serializable msg) throws Exception {
- // do something
- }
- }
- }
8.6.3 使用ProtoBuf序列化
- ProtobufDecoder
- ProtobufEncoder
- ProtobufVarint32FrameDecoder
- ProtobufVarint32LengthFieldPrepender
看下面代码:
- /**
- * 使用protobuf序列化数据,进行编码解码
- * 注意:使用protobuf需要protobuf-java-2.5.0.jar
- * @author Administrator
- *
- */
- 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 {
- ch.pipeline().addLast(new ProtobufVarint32FrameDecoder())
- .addLast(new ProtobufEncoder())
- .addLast(new ProtobufDecoder(lite))
- .addLast(new ObjectHandler());
- }
- public final class ObjectHandler extends SimpleChannelInboundHandler<Serializable> {
- @Override
- protected void channelRead0(ChannelHandlerContext ctx, Serializable msg) throws Exception {
- // do something
- }
- }
- }