【Netty】ChannelHandler和codec

时间  2019-11-06
标签 Netty channelhandler codec 栏目 Netty 繁體版
原文   原文链接

1、前言java

　　前面学习了Netty的codec框架，下面接着学习ChannelHandler与codec之间的关联。web

2、ChannelHandler和codec数据库

　　Netty为不一样的协议提供了处理器和编解码器，能够开箱即用，这些工具支持SSL / TLS和WebSocket，以及经过数据压缩使得HTTP有更好的性能。缓存

　　2.1 使用SSL/TLS保护Netty应用程序安全

　　因为数据隐私很是重要，而SSL和TLS等加密协议用于处理数据隐私，这些协议在其余协议之上以实现数据安全性。安全网站会使用这种协议，同时，不基于HTTP的应用程序，如安全SMTP（SMTPS）邮件服务甚相当系数据库系统也会用到这些协议。
服务器

　　为支持SSL/TLS协议，Java提供了javax.net.ssl包，其SSLContext和SSLEngine类使得实现解密和加密变得至关直接容易。而Netty使用SslHandler的ChannelHandler实现，其内部是使用SSLEngine来进行实际工做。websocket

　　下图展现了经过SslHandler的数据流。
网络

　　

　　下面代码展现如何经过ChannelInitializer将SslHandler添加至ChannelPipeline中。　框架

public class SslChannelInitializer extends ChannelInitializer<Channel>{
    private final SslContext context;
    private final boolean startTls;
    
    public SslChannelInitializer(SslContext context,
        boolean startTls) {
        this.context = context;
        this.startTls = startTls;
    }
    @Override
    protected void initChannel(Channel ch) throws Exception {
        SSLEngine engine = context.newEngine(ch.alloc());
        ch.pipeline().addFirst("ssl",
        　　new SslHandler(engine, startTls));
    }
}

　　在大多数状况下，SslHandler都是ChannelPipeline中的第一个ChannelHandler，只有当全部其余ChannelHandler将其逻辑应用于数据以后，才能进行加密操做。异步

　　2.2 构建Netty的HTTP/HTTPS应用程序

　　HTTP/HTTPS是最多见的协议套件之一，许多Web Service API都是基于HTTP/HTTPS的。

　　1. HTTP的decoder、encoder、codec

　　HTTP基于请求/响应模式：客户端发送HTTP请求至服务端，服务端响应HTTP请求至客户端。Netty提供各类编码器和解码器以简化使用此协议。

　　下图展现了HTTP请求的组成部分。

　　

　　而下图展现了HTTP响应的组成部分。

　　

　　下面代码的HttpPipelineInitializer能够为应用提供HTTP支持，仅仅只须要往ChannelPipeline中添加正确的ChannelHandler便可。　

public class HttpPipelineInitializer extends ChannelInitializer<Channel> {
    private final boolean client;
    public HttpPipelineInitializer(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("encoder", new HttpRequestEncoder());
        } else {
            pipeline.addLast("decoder", new HttpRequestDecoder());
            pipeline.addLast("encoder", new HttpResponseEncoder());
        }
    }
}

　　2. HTTP消息聚合

　　当管道中的处理器被初始化后，就能够处理不一样的消息，但因为HTTP请求和响应由多部分组成，须要将这些部分汇聚成完整的结果，Netty提供了一个合成器，其能够消息的不一样部分组合成FullHttpRequest和FullHttpResponse。因为消息在完整以前须要被缓存，所以其会带来小小的开销，可是不用担忧消息的碎片化处理。

　　下面代码展现了如何自动化装配消息。

public class HttpAggregatorInitializer extends ChannelInitializer<Channel> {
    private final boolean isClient;
    public HttpAggregatorInitializer(boolean isClient) {
        this.isClient = isClient;
    }
    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        if (isClient) {
            pipeline.addLast("codec", new HttpClientCodec());
        } else {
            pipeline.addLast("codec", new HttpServerCodec());
        }
        pipeline.addLast("aggregator",
            new HttpObjectAggregator(512 * 1024));
    }
}

　　3. HTTP压缩

　　使用HTTP时，建议采用压缩来尽量减小传输数据的大小。Netty为压缩和解压缩提供了ChannelHandler的实现，其支持gzip和deflate编码。

　　以下代码展现了如何自动压缩HTTP消息。　　

public class HttpCompressionInitializer extends ChannelInitializer<Channel> {
    private final boolean isClient;
    public HttpCompressionInitializer(boolean isClient) {
        this.isClient = isClient;
    }
    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        if (isClient) {
            pipeline.addLast("codec", new HttpClientCodec());
            pipeline.addLast("decompressor",
            new HttpContentDecompressor());
        } else {
            pipeline.addLast("codec", new HttpServerCodec());
            pipeline.addLast("compressor",
                new HttpContentCompressor());
        }
    }
}

　　4. 使用HTTPS

　　当混合了SslHandler时就可使用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());
        pipeline.addFirst("ssl", new SslHandler(engine));
        if (isClient) {
            pipeline.addLast("codec", new HttpClientCodec());
        } else {
            pipeline.addLast("codec", new HttpServerCodec());
        }
    }
}

　　5. WebSocket

　　Netty为基于HTTP的应用程序提供了支持，WebSocket解决了长期存在的问题：若是底层使用HTTP协议，是一系列请求-响应交互，那么如何实时发布信息。AJAX提供了一些改进，可是数据流的驱动仍然来自客户端的请求。而WebSocket规范及其实现表明了一种更有效的解决方案，其为双向流量提供了一个单一的TCP链接，其在客户端和服务器之间提供真正的双向数据交换，而且能够处理任何类型的数据。

　　下图展现了WebSocket协议，开始数据通讯为纯HTTP，以后升级到双向WebSocket。

　　

　　为了应用中添加WebSocket支持，能够在管道中添加WebSocketFrames，其包含以下类型。

　　

　　下面代码展现了如何使用WebSocketServerProtocolHandler。　　

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
        public void channelRead0(ChannelHandlerContext ctx,
            TextWebSocketFrame msg) throws Exception {
            // Handle text frame
        }
}

public static final class BinaryFrameHandler extends
    SimpleChannelInboundHandler<BinaryWebSocketFrame> {
    @Override
    public void channelRead0(ChannelHandlerContext ctx,
        BinaryWebSocketFrame msg) throws Exception {
        // Handle binary frame
    }
}

public static final class ContinuationFrameHandler extends
    SimpleChannelInboundHandler<ContinuationWebSocketFrame> {
    @Override
    public void channelRead0(ChannelHandlerContext ctx,
        ContinuationWebSocketFrame msg) throws Exception {
            // Handle continuation frame
        }
    }
}

　　而为了保证WebSocket的安全性，能够将SslHandler做为管道中的第一个处理器。

　　2.3 空闲链接和超时

　　检测空闲链接和超时对于及时释放资源相当重要，Netty提供了几个处理器进行处理，如IdleStateHandler、ReadTimeoutHandler、WriteTimeoutHandler等。

　　首先看看IdleStateHandler的缘由，若是60S内没有接受或发送数据，如何接收通知，可使用向远程对等体发送心跳消息来获取通知; 若是没有响应，则链接被关闭。

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 ChannelStateHandlerAdapter {
        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 {
                super.userEventTriggered(ctx, evt);
            }
        }
    }
}

　　2.4 解码限定的和基于长度的协议

　　1. 限定的协议

　　限定消息协议使用指定的字符来标记称为帧的消息或消息段的开始或结束位置，如如SMTP，POP3，IMAP和Telnet均使用该协议。以下的两个解码器用于处理限定的和基于长度的协议。

　　

　　下图显示了如何以行尾序列\ r \ n（回车+换行）分隔符来处理帧。

　　

　　以下代码展现了如何使用LineBasedFrameDecoder。　

public class LineBasedHandlerInitializer extends ChannelInitializer<Channel> {
    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new LineBasedFrameDecoder(64 * 1024));
        pipeline.addLast(new FrameHandler());
    }
    public static final class FrameHandler
        extends SimpleChannelInboundHandler<ByteBuf> {
        @Override
        public void channelRead0(ChannelHandlerContext ctx,
            ByteBuf msg) throws Exception {
            // Do something with the data extracted from the frame
        }
    }
}

　　DelimiterBasedFrameDecoder类能够自定义字符来处理数据。

　　2. 基于长度的协议

　　基于长度的协议经过在帧的头部对其长度进行编码来定义帧，而不是经过用特殊的分隔符标记其结尾，以下两个类用于处理该协议。

　　

　　下图展现了FixedLengthFrameDecoder的操做，每帧由8个字节构成。

　　

　　对于可变长度的帧，可使用LengthFieldBasedFrameDecoder进行处理，其从头部字段肯定帧长度，并从数据流中提取指定的字节数。

　　下图展现了长度字段在头部的偏移量0而且具备2字节。

　　

　　2.5 写大数据

　　因为网络缘由，异步框架中的写入大量数据将存在问题，由于写入操做是非阻塞的，在写入完成时或者全部的数据都没有被写出就会通知ChannelFuture，此时，若不中止写入数据，可能会形成内存溢出，所以，在写入大量数据时，须要处理与远程对等体链接缓慢可能致使内存释放延迟的状况。FileRegion支持零拷贝文件传输的通道。

　　下面是FileRegion的使用示例。　　

FileInputStream in = new FileInputStream(file);
FileRegion region = new DefaultFileRegion(
    in.getChannel(), 0, file.length());
channel.writeAndFlush(region).addListener(
    new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture future)
        throws Exception {
        if (!future.isSuccess()) {
            Throwable cause = future.cause();
            // Do something
        }
    }
});

　　此示例仅适用于文件内容的直接传输，不包括应用程序对数据的任何处理，若须要将数据从文件系统复制到用户内存，可使用ChunkedWriteHandler，它支持异步编写大型数据流，而不会致使高内存消耗。

　　其核心是ChunkedInput<B>接口，B代码readChunk的返回类型，对于ChunkedInput<B>，有以下的四种实现。

　　

　　以下是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();
        pipeline.addLast(new SslHandler(sslCtx.createEngine());
        pipeline.addLast(new ChunkedWriteHandler());
        pipeline.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)));
        }
    }
}

　　2.6 序列化数据

　　JDK提供了ObjectOutputStream和ObjectInputStream对POJO的原始数据类型和图形进行序列化和反序列化，但其并不高效。

　　1. JDK序列化

　　当应用程序必须与使用ObjectOutputStream和ObjectInputStream与对等体进行交互时，首先应该考虑兼容性，此时JDK的序列化是正确的选择。下图展现了Netty与JDK进行互操做的序列化类。

　　

　　2. JBOSS序列化

　　JBOSS比JDK序列化快三倍，而且更为紧凑，Netty使用以下两个编解码器来支持JBOSS。

　　

　　以下代码展现了如何使用JBOSS进行序列化操做。　

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();
        pipeline.addLast(new MarshallingDecoder(unmarshallerProvider));
        pipeline.addLast(new MarshallingEncoder(marshallerProvider));
        pipeline.addLast(new ObjectHandler());
    }
    public static final class ObjectHandler
        extends SimpleChannelInboundHandler<Serializable> {
        @Override
        public void channelRead0(
            ChannelHandlerContext channelHandlerContext,
            Serializable serializable) throws Exception {
            // Do something
        }
    }
}

　　3. Protobuf序列化

　　Netty还可以使用协议缓冲区进行序列化操做，其由谷歌开发，协议缓冲区以紧凑和高效的方式对结构化数据进行编码和解码，下图显示了Netty为支持protobuf而实现的ChannelHandler。

　　

　　下面示例展现了如何使用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();
        pipeline.addLast(new ProtobufVarint32FrameDecoder());
        pipeline.addLast(new ProtobufEncoder());
        pipeline.addLast(new ProtobufDecoder(lite));
        pipeline.addLast(new ObjectHandler());
    }
    public static final class ObjectHandler
        extends SimpleChannelInboundHandler<Object> {
        @Override
        public void channelRead0(ChannelHandlerContext ctx, Object msg)
            throws Exception {
            // Do something with the object
        }
    }
}

3、总结

　　本篇博文讲解了如何保证Netty应用程序的安全性，以及如何处理不一样协议的数据，以及编解码器和ChannelHandler之间的关系，以及如何把编解码器添加至管道中进行处理。也谢谢各位园友的观看~

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程