你了解Netty的编解码器吗？史上最通俗易懂的Netty解码器应用案例带你解开Netty解码器的神秘面纱1. 前言2. DelimiterBasedFrameDecoder解码器应用3. FixedLengthFrameDecoder解码器应用

78 阅读 0 评论 52 点赞

我是靠谱客的博主温婉黄蜂，最近开发中收集的这篇文章主要介绍你了解Netty的编解码器吗？史上最通俗易懂的Netty解码器应用案例带你解开Netty解码器的神秘面纱1. 前言2. DelimiterBasedFrameDecoder解码器应用3. FixedLengthFrameDecoder解码器应用，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

Netty解码器也是非常重要的一个模块, 服务端接收到客户端发送过来的消息, 准确说是字节数组, Netty底层已经将它们读取成ByteBuf了, 但是这些ByteBuf是没有任何含义的，需要我们根据业务来对字节数组进行解码。本文中我们将介绍Netty中常见的两种解码器DelimiterBasedFrameDecoder和FixedLengthFrameDecoder。

Netty解码器

1. 前言
- 1.1 Netty解码器
- 1.2 ByteToMessageDecoder
2. DelimiterBasedFrameDecoder解码器应用
- 2.1 服务端代码
- 2.2 客户端代码
- 2.3 运行结果
3. FixedLengthFrameDecoder解码器应用
- 3.1 服务端代码
- 3.2 telnet客户端测试

1. 前言

TCP以流的方式进行传输数据，上层的应用协议为了对消息进行区分，通常会采用以下4种方式：

消息长度固定：累计读取到长度总和为定长LEN的报文后，就认为读取到了一个完整的消息；然后会将计数器重置，重新开始读取下一个数据报；
回车换行符作为消息结束符：以回车换行符来标记一个数据报j结束；
特殊字符标识：自定义一个特殊字符，来作为数据报结束的标识；
消息头定义：通过在消息头种定义长度字段来标识消息的总长度。

Netty对上述4种应用做了统一的抽象，提供了4种解码器来解决相应的问题。在本文中我们将详细介绍DelimiterBasedFrameDecoder和FixedLengthFrameDecoder两种解码器，两者分别可以完成上述第3和第1种功能。

1.1 Netty解码器

在对这两种解码器介绍之前，我们简单了解一下Netty解码器的工作原理。下面首先给出一个简单的示例：
在这里插入图片描述
上述图片种展示是一个ToIntegerDecoder解码器的工作过程，从字面上我们可以了解，该解码器是将一个字节数组转化为Integer类型数据。decoder 负责将“入站”数据从一种格式转换到另一种格式，Netty的解码器是一种 ChannelInboundHandler 的抽象实现。实践中使用解码器很简单，就是将入站数据转换格式后传递到 ChannelPipeline 中的下一个ChannelInboundHandler 进行处理；这样的处理是很灵活的，我们可以将解码器放在 ChannelPipeline 中，重用逻辑。

Netty 提供了丰富的解码器抽象基类，我们可以很容易的实现这些基类来自定义解码器。主要分两类：

解码字节到消息（ByteToMessageDecoder 和 ReplayingDecoder）
解码消息到消息（MessageToMessageDecoder）

由于常用的几种解码器都是解码字节到消息，那么下面简单了解ByteToMessageDecoder。

1.2 ByteToMessageDecoder

ByteToMessageDecoder 是用于将字节转为消息（或其他字节序列）。你不能确定远端是否会一次发送完一个完整的“信息”,因此这个类会缓存入站的数据,直到准备好了用于处理。ByteToMessageDecoder抽象类中有两个最重要的方法，如下表所示：

方法名称	描述
Decode	需要实现的唯一抽象方法。通过具有输入字节的ByteBuf和添加了已解码消息的List来调用它。反复调用decode（），直到列表返回时为空。然后将List的内容传递到管道中的下一个处理程序。
decodeLast	所提供的默认实现只调用了decode（）。当Channel变为非活动状态时，此方法被调用一次。

下面我们依旧使用上面的ToIntegerDecoder作为示例。假设我们接收一个包含简单整数的字节流,每个都单独处理。在本例中,我们将从入站 ByteBuf 读取每个整数并将其传递给 pipeline 中的下一个ChannelInboundHandler。“解码”字节流成整数我们将扩展ByteToMessageDecoder，实现类为“ToIntegerDecoder”,如下图所示。
在这里插入图片描述
每次从入站的 ByteBuf 读取四个字节，解码成整形，并添加到一个 List （本例是指 Integer）,当不能再添加数据到 list 时，它所包含的内容就会被发送到下个 ChannelInboundHandler

读者如果想要详细了解解码器的源码设计可以阅读博客。

2. DelimiterBasedFrameDecoder解码器应用

DelimiterBasedFrameDecoder解码器是通用的分隔符解码器，可支持多个分隔符，每个分隔符可为一个或多个字符。如果定义了多个分隔符，并且可解码出多个消息帧，则选择产生最小帧长的结果。下面我们使用$_作为分隔符来演示。

2.1 服务端代码

package netty.frame.delimiter;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;

/**
 * created by LMR on 2020/5/20
 */
public class EchoServer {
    public void bind(int port) throws Exception {
	// 配置服务端的NIO线程组
	EventLoopGroup bossGroup = new NioEventLoopGroup();
	EventLoopGroup workerGroup = new NioEventLoopGroup();
	try {
	    ServerBootstrap b = new ServerBootstrap();
	    b.group(bossGroup, workerGroup)
		    .channel(NioServerSocketChannel.class)
		    .option(ChannelOption.SO_BACKLOG, 100)
		    .childHandler(new ChannelInitializer<SocketChannel>() {
			@Override
			public void initChannel(SocketChannel ch)
				throws Exception {
			    ByteBuf delimiter = Unpooled.copiedBuffer("$_"
				    .getBytes());
			    ch.pipeline().addLast(
				    new DelimiterBasedFrameDecoder(1024,
					    delimiter));
			    ch.pipeline().addLast(new StringDecoder());
			    ch.pipeline().addLast(new EchoServerHandler());
			}
		    });

	    // 绑定端口，同步等待成功
	    ChannelFuture f = b.bind(port).sync();

	    // 等待服务端监听端口关闭
	    f.channel().closeFuture().sync();
	} finally {
	    // 优雅退出，释放线程池资源
	    bossGroup.shutdownGracefully();
	    workerGroup.shutdownGracefully();
	}
    }

    public static void main(String[] args) throws Exception {
	int port = 8080;
	new EchoServer().bind(port);
    }
}

在initChannel方法中，我们首先创建分隔符缓冲对象ByteBuf，然后使用该分隔符缓冲对象创建DelimiterBasedFrameDecoder编码器对象，并加入到ChannelPipeline中。其中第二个参数白哦是消息的最大长度，如果超过这个长度还没有找到分隔符，则认为消息出错，报出异常，这是为了防止异常数据导致内存溢出，提高编码器的可靠性，最后还添加了字符串解码器和服务端处理类实例。

package netty.frame.delimiter;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
/**
 * created by LMR on 2020/5/20
 */
public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    int counter = 0;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
	    throws Exception {
	String body = (String) msg;
	System.out.println("This is " + ++counter + " times receive client : ["
		+ body + "]");
	body += "$_";
	ByteBuf echo = Unpooled.copiedBuffer(body.getBytes());
	ctx.writeAndFlush(echo);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
	cause.printStackTrace();
	ctx.close();// 发生异常，关闭链路
    }
}

channelRead方法非常简单，由于我们在ChannelPipeline中添加了多个编码器，那个在这里接收到的消息直接就是完整的消息数据字符串，由于我们使用DelimiterBasedFrameDecoder解码器过滤掉了分隔符，在这里我们重新添加分隔符，以便于客户端识别，再发送给客户端。

2.2 客户端代码

package netty.frame.delimiter;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;

/**
 * created by LMR on 2020/5/20
 */
public class EchoClient {

    public void connect(int port, String host) throws Exception {
	// 配置客户端NIO线程组
	EventLoopGroup group = new NioEventLoopGroup();
	try {
	    Bootstrap b = new Bootstrap();
	    b.group(group).channel(NioSocketChannel.class)
		    .option(ChannelOption.TCP_NODELAY, true)
		    .handler(new ChannelInitializer<SocketChannel>() {
			@Override
			public void initChannel(SocketChannel ch)
				throws Exception {
			    ByteBuf delimiter = Unpooled.copiedBuffer("$_"
				    .getBytes());
			    ch.pipeline().addLast(
				    new DelimiterBasedFrameDecoder(1024,
					    delimiter));
			    ch.pipeline().addLast(new StringDecoder());
			    ch.pipeline().addLast(new EchoClientHandler());
			}
		    });

	    // 发起异步连接操作
	    ChannelFuture f = b.connect(host, port).sync();

	    // 当代客户端链路关闭
	    f.channel().closeFuture().sync();
	} finally {
	    // 优雅退出，释放NIO线程组
	    group.shutdownGracefully();
	}
    }
    
    public static void main(String[] args) throws Exception {
	int port = 8080;
	new EchoClient().connect(port, "127.0.0.1");
    }
}

同样在客户端我们也添加相应的解码器，然后创建客户端处理类对象EchoClientHandler。

package netty.frame.delimiter;

import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * created by LMR on 2020/5/20
 */
public class EchoClientHandler extends ChannelInboundHandlerAdapter {

    private int counter;

    static final String ECHO_REQ = "This is a example by LMRZero.$_";
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
	for (int i = 0; i < 10; i++) {
	    ctx.writeAndFlush(Unpooled.copiedBuffer(ECHO_REQ.getBytes()));
	}
    }
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
	    throws Exception {
	System.out.println("This is " + ++counter + " times receive server : ["
		+ msg + "]");
    }
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
	ctx.flush();
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
	cause.printStackTrace();
	ctx.close();
    }
}

2.3 运行结果

服务端结果：
在这里插入图片描述
客户端结果：

3. FixedLengthFrameDecoder解码器应用

FixedLengthFrameDecoder是按照固定长度frameLength解码出消息帧。在本节中我们使用一个应用实例对其用法进行介绍。

3.1 服务端代码

我们在服务端ChannelPipeline中添加FixedLengthFrameDecoder，设置其长度为20，然后同样添加字符串解码器和服务端处理类实例。

package netty.frame.fixedLen;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.FixedLengthFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;

/**
 * created by LMR on 2020/5/20
 */
public class EchoServer {
    public void bind(int port) throws Exception {
	// 配置服务端的NIO线程组
	EventLoopGroup bossGroup = new NioEventLoopGroup();
	EventLoopGroup workerGroup = new NioEventLoopGroup();
	try {
	    ServerBootstrap b = new ServerBootstrap();
	    b.group(bossGroup, workerGroup)
		    .channel(NioServerSocketChannel.class)
		    .option(ChannelOption.SO_BACKLOG, 100)
		    .childHandler(new ChannelInitializer<SocketChannel>() {
			@Override
			public void initChannel(SocketChannel ch)
				throws Exception {
			    ch.pipeline().addLast(
				    new FixedLengthFrameDecoder(20));
			    ch.pipeline().addLast(new StringDecoder());
			    ch.pipeline().addLast(new EchoServerHandler());
			}
		    });

	    // 绑定端口，同步等待成功
	    ChannelFuture f = b.bind(port).sync();

	    // 等待服务端监听端口关闭
	    f.channel().closeFuture().sync();
	} finally {
	    // 优雅退出，释放线程池资源
	    bossGroup.shutdownGracefully();
	    workerGroup.shutdownGracefully();
	}
    }

    public static void main(String[] args) throws Exception {
	int port = 8080;
	new EchoServer().bind(port);
    }
}

下面看看服务端处理类EchoServerHandler的实现代码：

package netty.frame.fixedLen;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
/**
 * created by LMR on 2020/5/20
 */
public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
	    throws Exception {
	System.out.println("Receive client : [" + msg + "]");
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
	cause.printStackTrace();
	ctx.close();// 发生异常，关闭链路
    }
}

在本例中，我们在服务端接收到消息之后直接进行打印，不进行任何的操作。利用FixedLengthFrameDecoder解码器，无论一次接收到多少数据报，都会按照固定长度来进行解码。下面我们通过telnet命令行来测试服务端能否按照预期进行工作。

3.2 telnet客户端测试

在这里我们通过telnet命令来对服务端进行测试，下面介绍具体步骤：
（1）启动服务端
（2）开启本地回显功能，便于观察
在这里插入图片描述
（3）打开命令行窗口，输入telnet localhost 8080

（3）在命令行窗口输入需要传输的数据内容：

在这里插入图片描述
（4）服务端查看结果

可以看出每次都是收到20个字节的数据。

————————————————————————————————————————
参考博客和书籍：
https://www.w3cschool.cn/essential_netty_in_action/essential_netty_in_action-x7mn28bx.html
https://blog.csdn.net/usagoole/article/details/87389182
https://www.cnblogs.com/yuanrw/p/9866356.html
https://blog.csdn.net/mascf/article/details/60478539
https://www.cnblogs.com/ZhuChangwu/p/11225158.html
《Netty 权威指南》