文章目录提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档
- 基本说明
- 编码解码器
- 解码器-ByteToMessageDecoder
- Netty的handler链的调用机制
- 解码器-ReplayingDecoder
- 其它解码器
基本说明
-
netty的组件设计：Netty的主要组件有Channel、EventLoop、ChannelFuture、 ChannelHandler、ChannelPipe等
-
ChannelHandler充当了处理入站和出站数据的应用程序逻辑的容器。例如，实 现ChannelInboundHandler接口（或ChannelInboundHandlerAdapter），你就 可以接收入站事件和数据，这些数据会被业务逻辑处理。当要给客户端发送响 应时，也可以从ChannelInboundHandler冲刷数据。业务逻辑通常写在一个或 者多个ChannelInboundHandler中。ChannelOutboundHandler原理一样，只不 过它是用来处理出站数据的
-
ChannelPipeline提供了ChannelHandler链的容器。以客户端应用程序为例，如 果事件的运动方向是从客户端到服务端的，那么我们称这些事件为出站的，即 客户端发送给服务端的数据会通过pipeline中的一系列 ChannelOutboundHandler，并被这些Handler处理，反之则称为入站的
- 当Netty发送或者接受一个消息的时候，就将会发生一次数据转换。入站消息会 被解码：从字节转换为另一种格式（比如java对象）；如果是出站消息，它会 被编码成字节。
- Netty提供一系列实用的编解码器，他们都实现了ChannelInboundHadnler或者 ChannelOutboundHandler接口。在这些类中，channelRead方法已经被重写了。 以入站为例，对于每个从入站Channel读取的消息，这个方法会被调用。随后， 它将调用由解码器所提供的decode()方法进行解码，并将已经解码的字节转发 给ChannelPipeline中的下一个ChannelInboundHandler。
实例要求:
- 使用自定义的编码器和解码器来 说明Netty的handler 调用机制
客户端发送long -> 服务器
服务端发送long -> 客户端 - 案例演示
首先讲一下各个类的意思：
从上到下：
解码代码、客户端代码、客户端处理器、客户端通道初始化类、编码代码、服务器代码、服务器端处理器、服务器端通道初始化类。
下面从上至下附上完整代码：
解码代码
public class MyByteToLongDecoder extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List
客户端代码
public class Myclient {
public static void main(String[] args) throws Exception {
NioEventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new MyclientInitializer());
ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6000).sync();
channelFuture.channel().closeFuture().sync();
}finally {
group.shutdownGracefully();
}
}
}
客户端处理器
public class MyclientHandler extends SimpleChannelInboundHandler{ @Override protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception { } @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("MyclientHandler 发送数据"); ctx.writeAndFlush(12345L); } }
客户端通道初始化类
public class MyclientInitializer extends ChannelInitializer{ @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); //加入一个出战的handler，对数据进行编码 pipeline.addLast(new MyLongToByteEncoder()); //再加入一个handler，处理业务 pipeline.addLast(new MyclientHandler()); } }
编码代码
public class MyLongToByteEncoder extends MessageToByteEncoder{ @Override protected void encode(ChannelHandlerContext ctx, Long msg, ByteBuf out) throws Exception { System.out.println("MyLongToByteEncoder encode 被调用"); System.out.println("msg="+msg); out.writeLong(msg); } }
服务器代码
public class MyServer {
public static void main(String[] args) throws Exception {
NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
NioEventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new MyServerInitializer());
ChannelFuture channelFuture = serverBootstrap.bind(6000).sync();
System.out.println("server is ready");
channelFuture.channel().closeFuture().sync();
}finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
服务器端处理器
public class MyServerHandler extends SimpleChannelInboundHandler{ @Override protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception { System.out.println("从客户端"+ctx.channel().remoteAddress()+"读取到long"+msg); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); ctx.close(); } }
服务器端通道初始化类
public class MyServerInitializer extends ChannelInitializer{ @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); //入栈的handler进行解码 MyByteToLongDecoder pipeline.addLast(new MyByteToLongDecoder()); pipeline.addLast(new MyServerHandler()); } }
测试：先启动服务器再启动客户端
如果客户端发送的是String呢
public class MyclientHandler extends SimpleChannelInboundHandler{ @Override protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception { } @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("MyclientHandler 发送数据"); ctx.writeAndFlush(Unpooled.copiedBuffer("abcdabcdabcdabcd", CharsetUtil.UTF_8)); } }
)
现象：
1、客户端没有调用编码器
2、服务器端解码调用了两次
分析：
- “abcdabcdabcdabcd” 是 16个字节
- 该处理器的前一个handler 是 MyLongToByteEncoder
- MyLongToByteEncoder 父类 MessageToByteEncoder
- 父类 MessageToByteEncoder
ublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
ByteBuf buf = null;
try {
if (acceptOutboundMessage(msg)) { //判断当前msg 是不是应该处理的类型，如果是就处理，不是就跳过encode
@SuppressWarnings("unchecked")
I cast = (I) msg;
buf = allocateBuffer(ctx, cast, preferDirect);
try {
encode(ctx, cast, buf);
} finally {
ReferenceCountUtil.release(cast);
}
if (buf.isReadable()) {
ctx.write(buf, promise);
} else {
buf.release();
ctx.write(Unpooled.EMPTY_BUFFER, promise);
}
buf = null;
} else {
ctx.write(msg, promise);
}
}
我i们客户端的编码处理器只能处理Long类型
因此我们编写 Encoder 是要注意传入的数据类型和处理的数据类型一致
之前的代码是客户端给服务器发送消息，下面修改下代码，补充服务器给客户端回复消息
服务器端处理器
public class MyServerHandler extends SimpleChannelInboundHandler{ @Override protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception { System.out.println("从客户端" + ctx.channel().remoteAddress() + " 读取到long " + msg); //给客户端发送一个long ctx.writeAndFlush(98765L); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); ctx.close(); } }
服务器端通道初始化对象
public class MyServerInitializer extends ChannelInitializer{ @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline();//一会下断点 //入站的handler进行解码 MyByteToLongDecoder //pipeline.addLast(new MyByteToLongDecoder()); pipeline.addLast(new MyByteToLongDecoder2()); //出站的handler进行编码 pipeline.addLast(new MyLongToByteEncoder()); //自定义的handler 处理业务逻辑 pipeline.addLast(new MyServerHandler()); System.out.println("xx"); } }
客户端处理器：
public class MyclientHandler extends SimpleChannelInboundHandler{ @Override protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception { System.out.println("服务器的ip="+ctx.channel().remoteAddress()); System.out.println("收到服务器消息="+msg); } @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("MyclientHandler 发送数据"); // ctx.writeAndFlush(Unpooled.copiedBuffer("abcdabcdabcdabcd", CharsetUtil.UTF_8)); ctx.writeAndFlush(123456L); } }
客户端通道初始化类
public class MyclientInitializer extends ChannelInitializer{ @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); //加入一个出战的handler，对数据进行编码 pipeline.addLast(new MyLongToByteEncoder()); //客户端入站的编码器 pipeline.addLast(new MyByteToLongDecoder()); //再加入一个handler，处理业务 pipeline.addLast(new MyclientHandler()); } }
这样就形成了双向的handler链调用机制
public class MyByteToLongDecoder2 extends ReplayingDecoder其它解码器{ @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List
