服务端定义了一个Handler和三个Decoder。Handler接收客户端的信息,然后传递给decoder过滤处理。
1.服务端
package com.learn.netty.codecs;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
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.string.StringDecoder;
import java.net.InetSocketAddress;
public class Server {
public static void main(String[] args) throws Exception {
ServerBootstrap bootstrap = new ServerBootstrap();
EventLoopGroup group = new NioEventLoopGroup();
try {
bootstrap.group(group).channel(NioServerSocketChannel.class)
.localAddress(new InetSocketAddress(8888))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addFirst(new ReadHandler())
.addLast(new FixedLengthDecoder())
.addLast(new LoggingDecoder())
.addLast(new LastDecoder());
}
});
ChannelFuture future = bootstrap.bind().sync();
future.channel().closeFuture().sync();
} finally {
group.shutdownGracefully().sync();
}
}
}
2.服务端Handler
package com.learn.netty.codecs;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
public class ReadHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = (ByteBuf) msg;
System.out.println("Received: " + buf.readableBytes());
ctx.fireChannelRead(buf);
}
}
3.服务端 FixedLengthDecoder
客户端传递的是数字,Java中每个int类型4字节,读取转成字符串,然后传递到下一个Decoder。
package com.learn.netty.codecs;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.util.CharsetUtil;
import java.util.List;
public class FixedLengthDecoder extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
System.out.println("Fixed: " + byteBuf.readableBytes());
StringBuilder sb = new StringBuilder();
while (byteBuf.readableBytes() >= 4) {
// 读取才能向后传递
sb.append(byteBuf.readInt());
}
list.add(Unpooled.copiedBuffer(sb, CharsetUtil.UTF_8));
}
}
4.服务端 LoggingDecoder
将所有信息记录下来,然后传递接收的数字并追加一个字符串。
package com.learn.netty.codecs;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.util.CharsetUtil;
import java.nio.charset.Charset;
import java.util.List;
public class LoggingDecoder extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
System.out.println("Logging: " + byteBuf.readableBytes());
String num = "";
while (byteBuf.isReadable()) {
// 读取才能向后传递
num = byteBuf.readCharSequence(byteBuf.readableBytes(), Charset.defaultCharset()).toString();
System.out.println(num);
}
list.add(Unpooled.copyInt(Integer.parseInt(num)));
ByteBuf bf = Unpooled.copiedBuffer("Netty", CharsetUtil.UTF_8);
list.add(bf);
}
}
5.服务端 LastDecoder
读取接收的数字和字符串。
package com.learn.netty.codecs;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import java.nio.charset.Charset;
import java.util.List;
public class LastDecoder extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
// Decoder没有读取会被调用多次
// System.out.println("Last readable: " + byteBuf.readableBytes());
// System.out.println(byteBuf.toString(Charset.defaultCharset()));
System.out.println("Last readable: " + byteBuf.readableBytes());
if (byteBuf.readableBytes() == 4) {
int num = byteBuf.readInt();
System.out.println("num: " + num);
} else {
ByteBuf buf = byteBuf.readBytes(byteBuf.readableBytes());
System.out.println(buf.toString(Charset.defaultCharset()));
}
}
}
6.客户端
向服务端传递 6 个 int。
package com.learn.netty.codecs;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import java.net.InetSocketAddress;
public class Client {
public static void main(String[] args) throws Exception {
Bootstrap bootstrap = new Bootstrap();
EventLoopGroup group = new NioEventLoopGroup();
try {
bootstrap.group(group);
bootstrap.channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addFirst(new ChannelInboundHandlerAdapter(){
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("client active");
ctx.writeAndFlush(Unpooled.copyInt(1));
ctx.writeAndFlush(Unpooled.copyInt(2));
ctx.writeAndFlush(Unpooled.copyInt(3));
ctx.writeAndFlush(Unpooled.copyInt(4));
ctx.writeAndFlush(Unpooled.copyInt(5));
ctx.writeAndFlush(Unpooled.copyInt(6));
}
});
}
}).remoteAddress(new InetSocketAddress("127.0.0.1", 8888));
ChannelFuture future = bootstrap.connect().sync();
future.channel().close().sync();
} finally {
group.shutdownGracefully().sync();
}
}
}
结果:
Received: 24
Fixed: 24
Logging: 6
123456
Last readable: 4
num: 123456
Last readable: 5
Netty