Netty学习笔记 10.2 Netty的handler链的调用机制

实例要求:
使用自定义的编码器和解码器来说明Netty的handler 调用机制
客户端发送long -> 服务器
服务端发送long -> 客户端

案例演示

3. 结论
   不论解码器handler 还是 编码器handler 即接收的消息类型必须与待处理的消息类型一致，否则该handler不会被执行

在解码器 进行数据解码时，需要判断 缓存区(ByteBuf)的数据是否足够 ，否则接收到的结果会期望结果可能不一致

MyByteToLongDecoder

package com.my.netty.inboundhandlerandoutboundhandler;import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;import java.util.List;public class MyByteToLongDecoder extends ByteToMessageDecoder {/**** decode 会根据接收的数据，被调用多次, 直到确定没有新的元素被添加到list* , 或者是ByteBuf 没有更多的可读字节为止* 如果list out 不为空，就会将list的内容传递给下一个 channelinboundhandler处理, 该处理器的方法也会被调用多次** @param ctx 上下文对象* @param in 入站的 ByteBuf* @param out List 集合，将解码后的数据传给下一个handler* @throws Exception*/@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {System.out.println("MyByteToLongDecoder 被调用");//因为 long 8个字节, 需要判断有8个字节，才能读取一个longif(in.readableBytes() >= 8) {out.add(in.readLong());}}
}

MyLongToByteEncoder

package com.my.netty.inboundhandlerandoutboundhandler;import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;public class MyLongToByteEncoder extends MessageToByteEncoder<Long> {//编码方法@Overrideprotected void encode(ChannelHandlerContext ctx, Long msg, ByteBuf out) throws Exception {System.out.println("MyLongToByteEncoder encode 被调用");System.out.println("msg=" + msg);out.writeLong(msg);}
}

MyClient

package com.my.netty.inboundhandlerandoutboundhandler;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;public class MyClient {public static void main(String[] args)  throws  Exception{EventLoopGroup group = new NioEventLoopGroup();try {Bootstrap bootstrap = new Bootstrap();bootstrap.group(group).channel(NioSocketChannel.class).handler(new MyClientInitializer()); //自定义一个初始化类ChannelFuture channelFuture = bootstrap.connect("localhost", 7000).sync();channelFuture.channel().closeFuture().sync();}finally {group.shutdownGracefully();}}
}

MyClient

package com.my.netty.inboundhandlerandoutboundhandler;import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;import java.nio.charset.Charset;public class MyClientHandler  extends SimpleChannelInboundHandler<Long> {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {System.out.println("服务器的ip=" + ctx.channel().remoteAddress());System.out.println("收到服务器消息=" + msg);}//重写channelActive 发送数据@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println("MyClientHandler 发送数据");//ctx.writeAndFlush(Unpooled.copiedBuffer(""))ctx.writeAndFlush(123456L); //发送的是一个long//分析//1. "abcdabcdabcdabcd" 是 16个字节//2. 该处理器的前一个handler 是  MyLongToByteEncoder//3. MyLongToByteEncoder 父类  MessageToByteEncoder//4. 父类  MessageToByteEncoder/*public 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);}}4. 因此我们编写 Encoder 是要注意传入的数据类型和处理的数据类型一致*/// ctx.writeAndFlush(Unpooled.copiedBuffer("abcdabcdabcdabcd",CharsetUtil.UTF_8));}
}

MyClientInitializer

package com.my.netty.inboundhandlerandoutboundhandler;import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;public class MyClientInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();//加入一个出站的handler 对数据进行一个编码pipeline.addLast(new MyLongToByteEncoder());//这时一个入站的解码器(入站handler )//pipeline.addLast(new MyByteToLongDecoder());pipeline.addLast(new MyByteToLongDecoder2());//加入一个自定义的handler ， 处理业务pipeline.addLast(new MyClientHandler());}
}

MyServer

package com.my.netty.inboundhandlerandoutboundhandler;import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;public class MyServer {public static void main(String[] args) throws Exception{EventLoopGroup bossGroup = new NioEventLoopGroup(1);EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class).childHandler(new MyServerInitializer()); //自定义一个初始化类ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();channelFuture.channel().closeFuture().sync();}finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}

MyServerHandler

package com.my.netty.inboundhandlerandoutboundhandler;import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;public class MyServerHandler extends SimpleChannelInboundHandler<Long> {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {System.out.println("从客户端" + ctx.channel().remoteAddress() + " 读取到long " + msg);//给客户端发送一个longctx.writeAndFlush(98765L);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.close();}
}

MyServerInitializer

package com.my.netty.inboundhandlerandoutboundhandler;import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;public class MyServerInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected 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");}
}