5.1 DelimiterBasedFrameDecoder应用开发
下面我们来完成一个演示程序使用DelimiterBasedFrameDecoder应用进行开发
5.1.1 DelimiterBasedFrameDecoder服务端开发
首先我们创建分隔符缓冲对象，本例中我们使用"$_"作为分隔符，然后创建DelimiterBasedFrameDecoder对象加入到ChannelPipeline中去
public class EchoServer {
public void bind(int port) throws InterruptedException {
// 配置服务端的NIO线程组
// 主线程组, 用于接受客户端的连接，但是不做任何具体业务处理，像老板一样，负责接待客户，不具体服务客户
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
// 工作线程组, 老板线程组会把任务丢给他，让手下线程组去做任务，服务客户
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap(); // (2)
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class) // (3)
.option(ChannelOption.SO_BACKLOG,100).handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() { // (4)
@Override
public void initChannel(SocketChannel ch){
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(); // (7)
System.out.println("服务器启动完成，监听端口： " + port );
// 等待服务端监听端口关闭
f.channel().closeFuture().sync();
}finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws InterruptedException {
int port = 8081;
if (args != null && args.length > 0){
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e){
}
}
new EchoServer().bind(port);
}
}
public class EchoServerHandler extends ChannelInboundHandlerAdapter{
int counter = 0;
/**
* 由于使用了DelimiterBasedFrameDecoder解码器，channelHandler接收到的msg就是一个完整的消息包
* 然后StringDecoder将ByteBuf解码成了字符串对象
* 所以这里的msg就是解码后的字符串对象
* 我们在返回消息给客户端的时候加上"$_"分隔符 供客户端解码
* @param ctx
* @param msg
* @throws Exception
*/
@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) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
5.1.2 DelimiterBasedFrameDecoder客户端开发
与服务端类似，分别将DelimiterBasedFrameDecoder和StringDecoder添加到客户端ChannelPipeline中
public class EchoClient {
public void connect(int port, String host) throws InterruptedException {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.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 = bootstrap.connect(host, port).sync();
// 等待客户端链路关闭
f.channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
}
public static void main(String[] args) throws InterruptedException {
int port = 8081;
if (args != null && args.length > 0){
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
}
}
new EchoClient().connect(port,"127.0.0.1");
}
}
public class EchoClientHandler extends ChannelInboundHandlerAdapter{
private int counter;
private static final String ECHO_REQ = "Hi, Lilinfeng. Welcome to Netty.$_";
public EchoClientHandler() {
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
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) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
5.1.3 运行DelimiterBasedFrameDecoder服务端和客户端
服务端成功接受了十条消息，客户端也成功接受了返回的10条消息，结果符合预期。
This is 1 times receive server : [Hi, Lilinfeng. Welcome to Netty.]
This is 2 times receive server : [Hi, Lilinfeng. Welcome to Netty.]
This is 3 times receive server : [Hi, Lilinfeng. Welcome to Netty.]
This is 4 times receive server : [Hi, Lilinfeng. Welcome to Netty.]
This is 5 times receive server : [Hi, Lilinfeng. Welcome to Netty.]
This is 6 times receive server : [Hi, Lilinfeng. Welcome to Netty.]
This is 7 times receive server : [Hi, Lilinfeng. Welcome to Netty.]
This is 8 times receive server : [Hi, Lilinfeng. Welcome to Netty.]
This is 9 times receive server : [Hi, Lilinfeng. Welcome to Netty.]
This is 10 times receive server : [Hi, Lilinfeng. Welcome to Netty.]
This is 1 times receive client : [Hi, Lilinfeng. Welcome to Netty.]
This is 2 times receive client : [Hi, Lilinfeng. Welcome to Netty.]
This is 3 times receive client : [Hi, Lilinfeng. Welcome to Netty.]
This is 4 times receive client : [Hi, Lilinfeng. Welcome to Netty.]
This is 5 times receive client : [Hi, Lilinfeng. Welcome to Netty.]
This is 6 times receive client : [Hi, Lilinfeng. Welcome to Netty.]
This is 7 times receive client : [Hi, Lilinfeng. Welcome to Netty.]
This is 8 times receive client : [Hi, Lilinfeng. Welcome to Netty.]
This is 9 times receive client : [Hi, Lilinfeng. Welcome to Netty.]
This is 10 times receive client : [Hi, Lilinfeng. Welcome to Netty.]
5.2 FixedLengthFrameDecoder 应用开发
FixedLengthFrameDecoder是固定长度解码器，它能够按照指定的长度对消息进行自动解码。
5.2.1 FixedLengthFrameDecoder 服务端开发
类似的，在ChannelPipeline中新增FixedLengthFrameDecoder解码器，长度设置为20
public class EchoServer {
public void bind(int port) throws InterruptedException {
// 配置服务端的NIO线程组
// 主线程组, 用于接受客户端的连接，但是不做任何具体业务处理，像老板一样，负责接待客户，不具体服务客户
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
// 工作线程组, 老板线程组会把任务丢给他，让手下线程组去做任务，服务客户
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap(); // (2)
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class) // (3)
.option(ChannelOption.SO_BACKLOG,100).handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() { // (4)
@Override
public void initChannel(SocketChannel ch){
ch.pipeline().addLast(new FixedLengthFrameDecoder(20));
ch.pipeline().addLast(new StringDecoder());
ch.pipeline().addLast(new EchoServerHandler());
}
});
// 绑定端口，开始接收进来的连接
ChannelFuture f = b.bind(port).sync(); // (7)
System.out.println("服务器启动完成，监听端口： " + port );
// 等待服务端监听端口关闭
f.channel().closeFuture().sync();
}finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws InterruptedException {
int port = 8081;
if (args != null && args.length > 0){
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e){
}
}
new EchoServer().bind(port);
}
}
public class EchoServerHandler extends ChannelInboundHandlerAdapter{
int counter = 0;
/**
* 由于使用了FixedLengthFrameDecoder解码器，无论一次接受了多少数据包，都会按照设置的定长解码，
* @param ctx
* @param msg
* @throws Exception
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("This is " + ++counter + " times receive client : ["+msg+"]");
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
5.2.2 利用telnet命令测试
如下所示，每次服务端读取数据为20字节定长
//终端工具命令
YaleWeideMacBook-Pro:~ yale$ telnet localhost 8081
Trying ::1...
Connected to localhost.
Escape character is '^]'.
Lilinfeng welcome to Netty at Nanjing
//server 端接收打印
INFO: [id: 0x4a43fb90, L:/0:0:0:0:0:0:0:0:8081] READ COMPLETE
This is 1 times receive client : [Lilinfeng welcome to]
5.3 总结
DelimiterBasedFrameDecoder 用于对使用分隔符结尾的消息进行自动解码FixedLengthFrameDecoder用于对固定长度的消息进行自动解码。除了有上述两种解码器，再结合其他解码器，如字符串解码器等，可以轻松地完成对很多消息的自动解码，不用再考虑粘包拆包的读半包问题。
网友评论