netty之 定长数据流处理

/**
     * 监听处理逻辑
     * @param port 监听端口
     * @return ChannelFuture
     * @throws InterruptedException
     */
    public ChannelFuture doAccept(int port) throws InterruptedException {
        /**
         * childHandler 使bootstrap 独有的方法，是用于提供处理对象的
         * 可以一次性添加若干个处理逻辑，是类似责任链模式处理方式。
         * 增加A，B两个处理逻辑，在处理客户端请求数据的时候，根据A->B 顺序依次处理
         * ChannelInitializer 用于提供处理器的模型对象
         *  其中定义了 initChannel方法，用于初始化处理逻辑责任链的。
         *  可以保证服务端的bootstrap只初始化一次处理器，提供处理逻辑的重用，避免反复创建处理器对象，节约资源
         */
        bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                //  定长数据流的修改在此
                ChannelHandler[] acceptorHandlers = new ChannelHandler[3];
                //定长Handler，通过构造函数的参数设置消息长度，发送的消息长度不足可以用空格补全
                //当使用netty codec中的工具类时，必须要引入netty-codec-http依赖
                acceptorHandlers[0] = new FixedLengthFrameDecoder(3);
                //字符串解码器Handler，会自动处理channelRead方法中的msg参数，讲ByteBuf类型的数据转换成String类型
                acceptorHandlers[1] = new StringDecoder(StandardCharsets.UTF_8);
                acceptorHandlers[2] = new Server4FiexedLengthHandler();
                socketChannel.pipeline().addLast(acceptorHandlers);
            }
        });
        //bind 绑定端口，可以绑定多个端口
        //sync 开启监听，返回ChannelFuture，代表监听成功后的一个对应的未来结果
        //可以使用此对象实现后续的服务器和客户端的交互
        return bootstrap.bind(port).sync();
    }

import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandler.Sharable;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    import java.nio.charset.StandardCharsets;
    @Sharable
    public class Server4FiexedLengthHandler extends ChannelHandlerAdapter {
        /**
        * 用户处理读取数据请求的逻辑
        * @param ctx 包含与客户端建立连接的资源，如： 对应的channel
        * @param msg 读取到的数据，默认类型为ByteBuf.是对ByteBuffer的封装
        * @throws Exception
        */
        @Override
        public void channelRead(ChannelHandlerContext ctx,Object msg) throws Exception {
            String message = msg.toString();
            System.out.println("from client:" + message);
            String serverMsg = "ok ";
            ctx.writeAndFlush(Unpooled.copiedBuffer(serverMsg.getBytes(StandardCharsets.UTF_8)));

        }
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            System.out.println("server exceptionCaught method run ..... ");
            ctx.close();
        }
    }

public ChannelFuture doRequest(String host, int port) throws InterruptedException {
        /**
         * 客户端的bootstrap没有childHandler，只有handler
         *
         * 在客户端必须绑定处理器，也就是必须绑定handler方法。
         * 服务器必须绑定处理器，必须绑定childHandler方法
         */
        this.bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                ChannelHandler[] handlers = new ChannelHandler[3];
                handlers[0] = new FixedLengthFrameDecoder(3);
                //字符串解码器Handler，会自动处理channelRead方法中的msg参数，讲ByteBuf类型的数据转换成String类型
                handlers[1] = new StringDecoder(StandardCharsets.UTF_8);
                handlers[2] = new Client4FixedLengthHandler();

                socketChannel.pipeline().addLast(handlers);
            }
        });
        //connect 连接
        return this.bootstrap.connect(host, port).sync();
    }

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.ReferenceCountUtil;

import java.nio.charset.StandardCharsets;

public class Client4FixedLengthHandler extends ChannelHandlerAdapter {


    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try{
            String message = msg.toString();
            System.out.println("from server : "+message);
        }finally {
            ReferenceCountUtil.release(msg);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("client exceptionCaught method run .....");
        ctx.close();
    }
}