Netty中分隔符解碼器代碼示例與分析
阿新 • • 發佈：2018-02-12
rac ride 通用 否則 eventloop connect href throw java [toc]
Netty中分隔符解碼器代碼示例與分析
通過特別解碼器的使用，可以解決Netty中TCP的粘包問題，上一篇《Netty中LineBasedFrameDecoder解碼器使用與分析：解決TCP粘包問題》通過行解碼器的使用來解決TCP粘包問題，這意味著Netty的服務端和客戶端在每次發送請求消息前，都需要在消息的尾部拼接換行符。
除了使用行解碼器外，Netty還提供了通用的分隔符解碼器，即可以自定義消息的分隔符，那就是DelimiterBasedFrameDecoder分隔符解碼器。
下面的示例代碼來自於《Netty權威指南》第5章，這個例子模擬的是echo服務，即類似於ping，客戶端向服務端echo一個消息，服務端就會進行回應。
仍然需要註意的是，代碼做了部分修改，也加入了個人的一些註釋和分析。
服務端
EchoServer.java
package cn.xpleaf.netty; 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; 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, 1024) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { // 構建DelimiterBasedFrameDecoder解碼器的分隔符 ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes()); // 添加DelimiterBasedFrameDecoder解碼器到pipeline中 // 1024表示單條消息的最大長度，當達到該長度後仍然沒有查找到分隔符，就拋出TooLongFrameExceptio異常 // 防止由於異常碼流缺失分隔符導致的內存溢出，這是Netty解碼器的可靠性保護 // 第二個參數就是分隔符緩沖對象 ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter)); // 添加StringDecoder解碼器，將ByteBuf解碼成字符串對象 ch.pipeline().addLast(new StringDecoder()); // 添加業務處理handler 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; if(args != null && args.length > 0) { try { port = Integer.valueOf(port); } catch (NumberFormatException e) { // TODO: handle exception } } new EchoServer().bind(port); } }
EchoServerHandler.java
package cn.xpleaf.netty; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; public class EchoServerHandler extends ChannelInboundHandlerAdapter { private int counter = 0; @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { String body = (String) msg; // counter的作用是標記這是第幾次收到客戶端的請求 System.out.println("This is : " + ++counter + " times receive client : [" + body + "]"); // 由於設置DelimiterBasedFrameDecoder過濾掉了分隔符， // 所以返回給客戶端時需要在請求消息尾部拼接分隔符"$_" body += "$_"; ByteBuf echo = Unpooled.copiedBuffer(body.getBytes()); ctx.write(echo); } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ctx.flush(); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // 發生異常，關閉鏈路 ctx.close(); } }
客戶端
EchoClient.java
package cn.xpleaf.netty;
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;
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
protected void initChannel(SocketChannel ch) throws Exception {
// 構建DelimiterBasedFrameDecoder解碼器的分隔符
ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());
// 添加DelimiterBasedFrameDecoder解碼器到pipeline中
// 1024表示單條消息的最大長度，當達到該長度後仍然沒有查找到分隔符，就拋出TooLongFrameExceptio異常
// 防止由於異常碼流缺失分隔符導致的內存溢出，這是Netty解碼器的可靠性保護
// 第二個參數就是分隔符緩沖對象
ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
// 添加StringDecoder解碼器，將ByteBuf解碼成字符串對象
ch.pipeline().addLast(new StringDecoder());
// 添加業務處理handler
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;
if(args != null && args.length > 0) {
try {
port = Integer.valueOf(port);
} catch (NumberFormatException e) {
// 采用默認值
}
}
new EchoClient().connect(port, "localhost");
}
}EchoClientHandler.java
package cn.xpleaf.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
public class EchoClientHandler extends ChannelInboundHandlerAdapter {
private int counter;
static final String ECHO_REQ = "Hi, xpleaf. Welcome to Netty.$_";
public EchoClientHandler() {
}
@Override
public void channelActive(ChannelHandlerContext ctx) {
// 構建ByteBuf對象
ByteBuf req = Unpooled.copiedBuffer(ECHO_REQ.getBytes());
// 每次調用ctx.write()寫入ByteBuf對象時，必須要重新構建一個ByteBuf對象
// 即不能對一個ByteBuf對象多次使用ctx.write()，否則會發生異常
// 向服務端發送10次請求
for(int i = 0; i < 10; i++) {
ctx.writeAndFlush(Unpooled.copiedBuffer(ECHO_REQ.getBytes()));
// ctx.writeAndFlush(req);
}
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String body = (String) msg;
// counter的作用是標記這是第幾次收到服務端的回應
System.out.println("This is : " + ++counter + " times receive server : [" + body + "]");
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
}測試
服務端輸出：
This is : 1 times receive client : [Hi, xpleaf. Welcome to Netty.]
This is : 2 times receive client : [Hi, xpleaf. Welcome to Netty.]
This is : 3 times receive client : [Hi, xpleaf. Welcome to Netty.]
This is : 4 times receive client : [Hi, xpleaf. Welcome to Netty.]
This is : 5 times receive client : [Hi, xpleaf. Welcome to Netty.]
This is : 6 times receive client : [Hi, xpleaf. Welcome to Netty.]
This is : 7 times receive client : [Hi, xpleaf. Welcome to Netty.]
This is : 8 times receive client : [Hi, xpleaf. Welcome to Netty.]
This is : 9 times receive client : [Hi, xpleaf. Welcome to Netty.]
This is : 10 times receive client : [Hi, xpleaf. Welcome to Netty.]客戶端輸出：
This is : 1 times receive server : [Hi, xpleaf. Welcome to Netty.]
This is : 2 times receive server : [Hi, xpleaf. Welcome to Netty.]
This is : 3 times receive server : [Hi, xpleaf. Welcome to Netty.]
This is : 4 times receive server : [Hi, xpleaf. Welcome to Netty.]
This is : 5 times receive server : [Hi, xpleaf. Welcome to Netty.]
This is : 6 times receive server : [Hi, xpleaf. Welcome to Netty.]
This is : 7 times receive server : [Hi, xpleaf. Welcome to Netty.]
This is : 8 times receive server : [Hi, xpleaf. Welcome to Netty.]
This is : 9 times receive server : [Hi, xpleaf. Welcome to Netty.]
This is : 10 times receive server : [Hi, xpleaf. Welcome to Netty.]Netty中分隔符解碼器代碼示例與分析