5.1 DelimiterBasedFrameDecoder應用開發(fā)
下面我們來完成一個演示程序使用DelimiterBasedFrameDecoder應用進行開發(fā)
5.1.1 DelimiterBasedFrameDecoder服務端開發(fā)
首先我們創(chuàng)建分隔符緩沖對象，本例中我們使用"$_"作為分隔符，然后創(chuàng)建DelimiterBasedFrameDecoder對象加入到ChannelPipeline中去
public class EchoServer {
public void bind(int port) throws InterruptedException {
// 配置服務端的NIO線程組
// 主線程組, 用于接受客戶端的連接，但是不做任何具體業(yè)務處理，像老板一樣，負責接待客戶，不具體服務客戶
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("服務器啟動完成，監(jiān)聽端口： " + port );
// 等待服務端監(jiān)聽端口關閉
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客戶端開發(fā)
與服務端類似，分別將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 應用開發(fā)
FixedLengthFrameDecoder是固定長度解碼器，它能夠按照指定的長度對消息進行自動解碼。
5.2.1 FixedLengthFrameDecoder 服務端開發(fā)
類似的，在ChannelPipeline中新增FixedLengthFrameDecoder解碼器，長度設置為20
public class EchoServer {
public void bind(int port) throws InterruptedException {
// 配置服務端的NIO線程組
// 主線程組, 用于接受客戶端的連接，但是不做任何具體業(yè)務處理，像老板一樣，負責接待客戶，不具體服務客戶
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("服務器啟動完成，監(jiān)聽端口： " + port );
// 等待服務端監(jiān)聽端口關閉
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字節(jié)定長
//終端工具命令
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用于對固定長度的消息進行自動解碼。除了有上述兩種解碼器，再結合其他解碼器，如字符串解碼器等，可以輕松地完成對很多消息的自動解碼，不用再考慮粘包拆包的讀半包問題。
