Netty粘包/拆包（二）

news/2024/7/15 16:43:41 标签: Netty

利用DelimiterBasedFrameDecoder解决TCP粘包问题

服务端：

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
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.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;

public class TimeServer3 {

    public void bind(int port) throws Exception {
        // NIO线程组
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChildChannelHandler());

            // 绑定端口，同步等待成功
            ChannelFuture f = b.bind(port).sync();
            // 等待服务端监听端口关闭
            f.channel().closeFuture().sync();
        } finally {
            // 优雅退出，释放线程池资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {
        @Override
        protected void initChannel(SocketChannel arg0) throws Exception {
            arg0.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,
                    Unpooled.copiedBuffer("$_".getBytes())));
            arg0.pipeline().addLast(new StringDecoder());
            arg0.pipeline().addLast(new TimeServerHandler3());
        }
    }

    public static void main(String[] args) throws Exception {

        int port=8082;
        new TimeServer3().bind(port);
    }
}

这里通过"$_"作为分隔符。

arg0.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,
        Unpooled.copiedBuffer("$_".getBytes())));

DelimiterBasedFrameDecoder有多个构造方法，这里传递两个参数：第一个1024表示单条消息最大长度。当达到该长度后任然没有查到分隔符，就抛出TooLongFrameException异常，防止由于异常码流缺失分隔符导致内存溢出，第二个参数就是分隔符缓冲对象。

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.Date;

public class TimeServerHandler3 extends ChannelInboundHandlerAdapter {

    private int counter;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        System.out.println("服务器启动..");
        String body = (String) msg;
        System.out.println("服务器接收数据:" + body + " 次数:" + (++counter));
        String currentTime = body+ new Date(
                System.currentTimeMillis()).toString();
        currentTime = currentTime + "!_";
        ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
        ctx.writeAndFlush(resp);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        System.out.println("服务器读取完毕..");
        ctx.flush();//刷新后才将数据发出到SocketChannel
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        System.out.println("服务器异常..");
        ctx.close();
    }
}

由于设置了DelimiterBasedFrameDecoder过滤掉了分隔符，所以返回给客户端时需要在请求消息尾部拼接分隔符“!_”。

客户端：

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;

public class TimeClient3 {

    public void connect(int port, String host) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();

            b.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,
                                    Unpooled.copiedBuffer("!_".getBytes())));
                            ch.pipeline().addLast(new StringDecoder());
                            ch.pipeline().addLast(new TimeClientHandler3());
                        }
                    });

            // 发起异步连接操作
            ChannelFuture f = b.connect(host, port).sync();
            // 等待客户端连接关闭
            f.channel().closeFuture().sync();
        } finally {
            // 优雅退出，释放NIO线程组
            group.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        int port = 8082;
        new TimeClient3().connect(port, "127.0.0.1");
    }
}

与服务端类似，分别将DelimiterBasedFrameDecoder和StringDecoder添加到客户端ChannelPipeline中，最后添加客户端I/O时间处理类TimeClientHandler3。

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.logging.Logger;

public class TimeClientHandler3 extends ChannelInboundHandlerAdapter {

    private static final Logger logger = Logger
            .getLogger(TimeClientHandler3.class.getName());

    private int counter;

    private byte[] req;

    public TimeClientHandler3() {
        req = ("Hello Netty……$_").getBytes();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx){
        //与服务端建立连接后
        System.out.println("客户端 channelActive..");
        ByteBuf message=null;
        for (int i=0;i<100;i++){
            message=Unpooled.buffer(req.length);
            message.writeBytes(req);
            ctx.writeAndFlush(message);
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        System.out.println("客户端 channelRead..");
        //服务端返回消息后
        String body = (String) msg;
        System.out.println("当前时间 :" + body + " 次数:" + (++counter));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
            throws Exception {
        System.out.println("客户端异常..");
        // 释放资源
        logger.warning("Unexpected exception from downstream:"
                + cause.getMessage());
        ctx.close();
    }

}

利用FixedLengthFrameDecoder解决TCP粘包问题

服务端：

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
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.FixedLengthFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;

public class TimeServer4 {

    public void bind(int port) throws Exception {
        // NIO线程组
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 100)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChildChannelHandler());

            // 绑定端口，同步等待成功
            ChannelFuture f = b.bind(port).sync();
            // 等待服务端监听端口关闭
            f.channel().closeFuture().sync();
        } finally {
            // 优雅退出，释放线程池资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {
        @Override
        protected void initChannel(SocketChannel arg0) throws Exception {
            arg0.pipeline().addLast(new FixedLengthFrameDecoder(20));
            arg0.pipeline().addLast(new StringDecoder());
            arg0.pipeline().addLast(new TimeServerHandler4());
        }
    }

    public static void main(String[] args) throws Exception {

        int port=8082;
        new TimeServer4().bind(port);
    }
}

此处将 FixedLengthFrameDecoder设置为20个字符，因此只会接受20个字符。

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.Date;

public class TimeServerHandler4 extends ChannelInboundHandlerAdapter {

    private int counter;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        System.out.println("服务器启动..");
        String body = (String) msg;
        System.out.println("服务器接收数据:" + body + " 次数:" + (++counter));
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        System.out.println("服务器异常..");
        ctx.close();
    }
}

通过cmd进行发送数据：

总结：

DelimiterBasedFrameDecoder用于对使用分隔符结尾的消息进行自动解码，FixedLengthFrameDecoder对于固定长度的消息进行自动解码。

 


http://www.niftyadmin.cn/n/1457109.html

相关文章

Netty编解码（Protobuf实现）

Netty编解码（Protobuf实现）

Java序列化的缺点 Java序列化从JDK1.1版本就已经提供了&#xff0c;它不需要添加额外类库&#xff0c;只需要实现java.io.Serializable并生成序列ID即可。但是在远程服务调用&#xff08;RPC&#xff09;时&#xff0c;很少直接使用Java序列化进行消息的编解码和传输&#xff…
阅读更多...
设计模式（五）（模板方法模式、迭代器模式、组合模式）

设计模式（五）（模板方法模式、迭代器模式、组合模式）

十&#xff1a;模板方法模式 模板方法定义了一个算法的步骤&#xff0c;并允许子类为一个或多个步骤提供实现。 模板方法模式在一个方法中定义一个算法骨架&#xff0c;而将一些步骤延迟到子类中。模板方法使得子类可以在不改变算法结构的情况下&#xff0c;重新定义算法中的…
阅读更多...
Netty编解码（MessagePack实现）

Netty编解码（MessagePack实现）

MessagePack是一个高效的二进制序列化框架&#xff0c;它像JSON一样支持不同语言间的数据交换&#xff0c;但是它性能更快&#xff0c;序列化后码流更小。 MessagePack特点&#xff1a; 1.编解码高效&#xff0c;性能高&#xff1b; 2.序列化后的码流小&#xff1b; 3.支持…
阅读更多...
Netty协议开发（HTTP）

Netty协议开发（HTTP）

HTTP是一个属于应用层的面向对象协议&#xff0c;由于其简捷、快速的方式&#xff0c;适用于分布式超媒体信息系统。 HTTP协议的URL http://host[":"port][abs_path] http表示通过HTTP协议来定位网络资源&#xff1b;host表示合法的Internet主机域名或者IP地址&am…
阅读更多...
Netty协议开发（WebSocket）

Netty协议开发（WebSocket）

WebSocket是H5开始提供的一种游览器与服务器间进行双全工通信的网络技术&#xff0c;WebSocket通信协议与2011年被IETF定义为标准RFC6455&#xff0c;WebSocket API被W3c定义为标准。 其特点如下&#xff1a; 单一的TCP连接&#xff0c;采用双全工模式通信&#xff1b;对代理…
阅读更多...
设计模式（六）（状态模式、代理模式）

设计模式（六）（状态模式、代理模式）

十三&#xff1a;状态模式 状态模式允许对象在内部状态改变它的行为&#xff0c;对象看起来好像修改了它的类。 意图&#xff1a;允许对象在内部状态发生改变时改变它的行为&#xff0c;对象看起来好像修改了它的类。 主要解决&#xff1a;对象的行为依赖于它的状态&#xf…
阅读更多...
Netty协议栈开发

Netty协议栈开发

由于现代软件的复杂性&#xff0c;一个大型软件系统往往会被人为地拆分称为多个模块&#xff0c;另外随着移动互联网的兴起&#xff0c;网站的规模越来越大&#xff0c;为了能够支撑业务的发展&#xff0c;需要集群和分布式部署。模块之间的通信就需要进行跨节点通信。 在传统…
阅读更多...
安装m_map以及用matlab绘制高精度海岸线地图

安装m_map以及用matlab绘制高精度海岸线地图

最初目的是为了绘制站位图&#xff0c;matlab自带的geoshow绘制效果不太好&#xff0c;分辨率偏低。在这里整理好分享给大家&#xff0c;也方便自己以后查找~ 进入正题前&#xff0c;先介绍一下安装m_map工具箱 源自&#xff1a;https://blog.csdn.net/achumoyangguang/artic…
阅读更多...
最新文章

Copyright @ 2022~2023