关于netty的多个handler链式模式

1. 老规矩, 引入我们喜闻乐见的maven依赖

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.6.Final</version>
</dependency>

2. 服务端

　　2.1: 服务端引导类:

public class EchoServer {

    private int port;

    private EchoServer(int port) {
        this.port = port;
    }

    private void start() throws Exception {
        System.out.println("Echo Server Start");
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(group)
                    .channel(NioServerSocketChannel.class)
                    .localAddress(new InetSocketAddress(port))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new EchoOutboundHandler1());
                            ch.pipeline().addLast(new EchoOutboundHandler2());
                            
                            ch.pipeline().addLast(new EchoInboundHandler1());
                            ch.pipeline().addLast(new EchoInboundHandler2());
                        }
                    });
            ChannelFuture f = b.bind().sync();
            System.out.println("Server Start Listen At: " + port);
            f.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        int port;
        if (args.length > 0) {
            port = Integer.parseInt(args[0]);
        } else {
            port = 8080;
        }
        new EchoServer(port).start();
    }
}

　　2.2 EchoOutboundHandler1

public class EchoInboundHandler1 extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 读取msg中的数据
        ByteBuf result = (ByteBuf) msg;
        byte[] bytes = new byte[result.readableBytes()];
        result.readBytes(bytes);
        String resultStr = new String(bytes);
        System.out.println("Server Received: " + resultStr + ": Inbound 1 Is OK");
        
        // 处理完msg中的数据后往msg中重新存放新的数据继续传递
        result.writeBytes(resultStr.getBytes());
        ctx.fireChannelRead(result);
    }
    
}

　　2.3 EchoInboundHandler2

public class EchoInboundHandler2 extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 读取msg中的数据
        ByteBuf result = (ByteBuf) msg;
        byte[] bytes = new byte[result.readableBytes()];
        result.readBytes(bytes);
        String resultStr = new String(bytes);
        System.out.println("Server Received: " + resultStr + ": Inbound 2 Is OK");

        // 处理完msg中的数据后往msg中重新存放新的数据继续传递
        result.writeBytes(resultStr.getBytes());
        ctx.write(result);
    }

}

　　2.4 EchoOutboundHandler2 

public class EchoOutboundHandler2 extends ChannelOutboundHandlerAdapter {

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        // 读取msg中的数据
        ByteBuf result = (ByteBuf) msg;
        byte[] bytes = new byte[result.readableBytes()];
        result.readBytes(bytes);
        String resultStr = new String(bytes);
        System.out.println("Server Received: " + resultStr + ": Outbound 2 Is OK");
        
        // 处理完msg中的数据后往msg中重新存放新的数据继续传递
        result.writeBytes(resultStr.getBytes());
        ctx.write(result);
    }
}

　　2.5 EchoOutboundHandler1 

public class EchoOutboundHandler1 extends ChannelOutboundHandlerAdapter {

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        // 读取msg中的数据
        ByteBuf result = (ByteBuf) msg;
        byte[] bytes = new byte[result.readableBytes()];
        result.readBytes(bytes);
        String resultStr = new String(bytes);
        System.out.println("Server Received: " + resultStr + ": Outbound 1 Is OK");

        // 处理完msg中的数据后往msg中重新存放新的数据继续传递
        result.writeBytes(resultStr.getBytes());
        ctx.write(result);
        ctx.flush();
    }
}

3. 客户端

public class EchoClient {
    
    private String host;
    private int port;

    private EchoClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    private void start() throws Exception {
        System.out.println("Echo Client Start");
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioSocketChannel.class)
                    .remoteAddress(new InetSocketAddress(host, port))
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new EchoClientHandler());
                        }
                    });
            ChannelFuture f = b.connect().sync();
            System.out.println("Server Client Listen IP: [" + host + ":" + port + "]");
            f.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        String host = "127.0.0.1";
        int port = 8080;
        int len = 2;
        if (args.length == len) {
            host = args[0];
            port = Integer.parseInt(args[1]);
        }
        new EchoClient(host, port).start();
    }

}

public class EchoClientHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        ctx.writeAndFlush(Unpooled.copiedBuffer("Netty Rocks!", CharsetUtil.UTF_8));
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 读取msg中的数据
        ByteBuf result = (ByteBuf) msg;
        byte[] bytes = new byte[result.readableBytes()];
        result.readBytes(bytes);
        String resultStr = new String(bytes);
        System.out.println("Echo Client Received Is OK: " + resultStr);
        ctx.close();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

4. 结果:

　　服务端:

　　客户端:

5. 注意事项:

5.1. ChannelInboundHandler之间的传递, 通过调用 ctx.fireChannelRead(msg) 实现; 调用ctx.write(msg) 将传递到ChannelOutboundHandler

5.2. ctx.write()方法执行后, 需要调用flush()方法才能令它立即执行

5.3. ChannelOutboundHandler 在注册的时候需要放在最后一个ChannelInboundHandler之前, 否则将无法传递到ChannelOutboundHandler