【转载】Netty 学习 (五)-DelimiterBasedFrameDecoder

上一节我们说了 LineBasedframeDecoder 来解决粘包拆包的问题,TCP 以流的方式进行数据传输,上层应用协议为了对消息进行区分,一般采用如下 4 种方式:

  1. 消息长度固定,累计读取到消息长度总和为定长 Len 的报文之后即认为是读取到了一个完整的消息。计数器归位,重新读取。
  2. 将回车换行符作为消息结束符。
  3. 将特殊的分隔符作为消息分隔符,回车换行符是他的一种。
  4. 通过在消息头定义长度字段来标识消息总长度。

LineBasedframeDecoder 属于第二种,今天我们要说的 DelimiterBasedFrameDecoder 和 FixedLengthFrameDecoder 属于第三种和第一种。DelimiterBasedFrameDecoder 用来解决以特殊符号作为消息结束符的粘包问题,FixedLengthFrameDecoder 用来解决定长消息的粘包问题。下面首先来用 DelimiterBasedFrameDecoder 来写一个例子,我们看一下效果然后接着分析用法。

1. DelimiterBasedFrameDecoder 使用

服务端:

public class HelloWordServer {
    private int port;

    public HelloWordServer(int port) {
        this.port = port;
    }

    public void start(){
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workGroup = new NioEventLoopGroup();

        ServerBootstrap server = new ServerBootstrap().group(bossGroup,workGroup)
                                    .channel(NioServerSocketChannel.class)
                                    .childHandler(new ServerChannelInitializer());

        try {
            ChannelFuture future = server.bind(port).sync();
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) {
        HelloWordServer server = new HelloWordServer(7788);
        server.start();
    }
}

服务端 ServerChannelInitializer:

public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();

        ByteBuf delimiter = Unpooled.copiedBuffer("\t".getBytes());
        pipeline.addLast("framer", new DelimiterBasedFrameDecoder(2048,delimiter));    
        // 字符串解码 和 编码
        pipeline.addLast("decoder", new StringDecoder());
        pipeline.addLast("encoder", new StringEncoder());

        // 自己的逻辑 Handler
        pipeline.addLast("handler", new ServerHandler());
    }
}

服务端 handler:

public class ServerHandler extends ChannelInboundHandlerAdapter {
    private int counter;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        String body = (String)msg;
        System.out.println("server receive order :" + body + ";the counter is:" + ++counter);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
    }
}

客户端:

public class HelloWorldClient {
    private  int port;
    private  String address;

    public HelloWorldClient(int port,String address) {
        this.port = port;
        this.address = address;
    }

    public void start(){
        EventLoopGroup group = new NioEventLoopGroup();

        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ClientChannelInitializer());

        try {
            ChannelFuture future = bootstrap.connect(address,port).sync();
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            group.shutdownGracefully();
        }

    }

    public static void main(String[] args) {
        HelloWorldClient client = new HelloWorldClient(7788,"127.0.0.1");
        client.start();
    }
}

客户端 ClientChannelInitializer:

public class ClientChannelInitializer extends  ChannelInitializer<SocketChannel> {

    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();

        /*
         * 这个地方的 必须和服务端对应上。否则无法正常解码和编码
         *
         *
         */       
        ByteBuf delimiter = Unpooled.copiedBuffer("\t".getBytes());
        pipeline.addLast("framer", new DelimiterBasedFrameDecoder(2048,delimiter)); 
        pipeline.addLast("decoder", new StringDecoder());
        pipeline.addLast("encoder", new StringEncoder());

        // 客户端的逻辑
        pipeline.addLast("handler", new ClientHandler());
    }
}

客户端 handler:

public class ClientHandler extends ChannelInboundHandlerAdapter {
    private byte[] req;
    private int counter;

    public ClientHandler() {
         req = ("Unless required by applicable law or agreed to in writing, software\t" +
                "distributed under the License is distributed on an \"AS IS\"BASIS,\t" +
                "WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\t" +
                "See the License for the specific language governing permissions and\t" +
                "limitations under the License.This connector uses the BIO implementation that requires the JSSE\t" +
                "style configuration. When using the APR/native implementation, the\t" +
                "penSSL style configuration is required as described in the APR/native\t" +
                "documentation.An Engine represents the entry point (within Catalina) that processes\t" +
                "every request.  The Engine implementation for Tomcat stand alone\t" +
                "analyzes the HTTP headers included with the request, and passes them\t" +
                "on to the appropriate Host (virtual host)# Unless required by applicable law or agreed to in writing, software\t" +
                "# distributed under the License is distributed on an \"AS IS\"BASIS,\t" +
                "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\t" +
                "# See the License for the specific language governing permissions and\t" +
                "# limitations under the License.# For example, set the org.apache.catalina.util.LifecycleBase logger to log\t" +
                "# each component that extends LifecycleBase changing state:\t" +
                "#org.apache.catalina.util.LifecycleBase.level = FINE\t"
                ).getBytes();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ByteBuf message;

        message = Unpooled.buffer(req.length);
        message.writeBytes(req);
        ctx.writeAndFlush(message);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        String buf = (String)msg;
        System.out.println("Now is :" + buf + "; the counter is :"+ (++counter));
    }

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

输出如下:

server receive order : Unless required by applicable law or agreed to in writing, software;the counter is: 1
server receive order :   distributed under the License is distributed on an "AS IS" BASIS,;the counter is: 2
server receive order :   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.;the counter is: 3
server receive order :   See the License for the specific language governing permissions and;the counter is: 4
server receive order :   limitations under the License.This connector uses the BIO implementation that requires the JSSE;the counter is: 5
server receive order :   style configuration. When using the APR/native implementation, the;the counter is: 6
server receive order :   penSSL style configuration is required as described in the APR/native;the counter is: 7
server receive order :   documentation.An Engine represents the entry point (within Catalina) that processes;the counter is: 8
server receive order :   every request.  The Engine implementation for Tomcat stand alone;the counter is: 9
server receive order :   analyzes the HTTP headers included with the request, and passes them;the counter is: 10
server receive order :   on to the appropriate Host (virtual host)# Unless required by applicable law or agreed to in writing, software;the counter is: 11
server receive order : # distributed under the License is distributed on an "AS IS" BASIS,;the counter is: 12
server receive order : # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.;the counter is: 13
server receive order : # See the License for the specific language governing permissions and;the counter is: 14
server receive order : # limitations under the License.# For example, set the org.apache.catalina.util.LifecycleBase logger to log;the counter is: 15
server receive order : # each component that extends LifecycleBase changing state:;the counter is: 16
server receive order : #org.apache.catalina.util.LifecycleBase.level = FINE;the counter is: 17

启动服务端和客户端,我们能看到服务端接收客户端发过来的消息一共分 17 次接收。那么为什么是 17 次呢?而且我们并没有使用在上一篇中解决拆包和粘包问题的 LineBasedFrameDecoder,并且这次我们的消息每一行的末尾也换成了”\t”。下面就来讲解一下 DelimiterBasedFrameDecoder 的使用。

DelimiterBasedFrameDecoder 是将特殊的字符作为消息的分隔符,本例中用到的是”\t”。而 LineBasedFrameDecoder 是默认将换行符”\n”作为消息分隔符。首先我们注意到在 ServerChannelInitializer 中我们在添加解码器时跟以前有点不一样:

ByteBuf delimiter = Unpooled.copiedBuffer("\t".getBytes());
pipeline.addLast("framer", new DelimiterBasedFrameDecoder(2048, delimiter));

这里我们添加 DelimiterBasedFrameDecoder 解码器并且手动指定消息分隔符为:”\t”。我们可以看一下 DelimiterBasedFrameDecoder 的构造方法:

public DelimiterBasedFrameDecoder(int maxFrameLength, boolean stripDelimiter, ByteBuf delimiter) {
        this(maxFrameLength, stripDelimiter, true, delimiter);
}

maxFrameLength:解码的帧的最大长度

stripDelimiter:解码时是否去掉分隔符

failFast:为 true,当 frame 长度超过 maxFrameLength 时立即报 TooLongFrameException 异常,为 false,读取完整个帧再报异常

delimiter:分隔符

这个时候大家应该明白了为什么服务端分 17 次收到消息。我们在消息的每一行都加了一个”\t”, 自然解码器在度消息时遇到”\t”就会认为这是一条消息的结束。用这种方式我们可以把”\t”换成任何我们自定义的字符对象。换成”\n”也是可以的。

2. FixedLengthFrameDecoder 使用

FixedLengthFrameDecoder 是固定长度解码器,它能够按照指定的长度对消息进行自动解码。使用它也没有什么特别费力的事情,在 ServerChannelInitializer 类中添加:

pipeline.addLast(new FixedLengthFrameDecoder(23));// 参数为一次接受的数据长度

即可,同时也别忘了把刚才使用的 DelimiterBasedFrameDecoder 注释掉啊,不然达不到效果。

原地址 http://blog.csdn.net/a953713428/article/details/68231119