Java Netty Codecs 程序

服务端定义了一个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