使用Netty编写异步非阻塞Socket tcp通信,和TCP粘包、拆包问题解决

举报
程序员-上善若水 发表于 2022/06/23 23:54:06 2022/06/23
【摘要】 使用Netty编写异步非阻塞Socket tcp通信,和TCP粘包、拆包问题解决 一、Netty应用场景 分布式开源框架中dubbo、Zookeeper,RocketMQ底层rpc通讯使用就是nett...

使用Netty编写异步非阻塞Socket tcp通信,和TCP粘包、拆包问题解决

一、Netty应用场景

  1. 分布式开源框架中dubbo、Zookeeper,RocketMQ底层rpc通讯使用就是netty。
  2. 游戏开发中,底层使用netty通讯。

二、netty的优点

  1. NIO的类库和API繁杂,使用麻烦,你需要熟练掌握Selector、ServerSocketChannel、SocketChannel、ByteBuffer等;
  2. 需要具备其它的额外技能做铺垫,例如熟悉Java多线程编程,因为NIO编程涉及到Reactor模式,你必须对多线程和网路编程非常熟悉,才能编写出高质量的NIO程序;
  3. 可靠性能力补齐,工作量和难度都非常大。例如客户端面临断连重连、网络闪断、半包读写、失败缓存、网络拥塞和异常码流的处理等等,NIO编程的特点是功能开发相对容易,但是可靠性能力补齐工作量和难度都非常大;
  4. JDK NIO的BUG,例如臭名昭著的epoll bug,它会导致Selector空轮询,最终导致CPU 100%。官方声称在JDK1.6版本的update18修复了该问题,但是直到JDK1.7版本该问题仍旧存在,只不过该bug发生概率降低了一些而已,它并没有被根本解决。该BUG以及与该BUG相关的问题单如下:

三、集成

pom

		<dependency>
			<groupId>io.netty</groupId>
			<artifactId>netty-all</artifactId>
			<version>5.0.0.Alpha2</version>
		</dependency>

  
 
  • 1
  • 2
  • 3
  • 4
  • 5

如果在android上集成,请去网上下载 netty.jar,引入到环境中。注意版本问题。

实现服务端

1、回调

首先建立回调类,客户端发来的信息,以及其他一些日志信息,都会反馈到这个Handler类中,下面开启服务时,会指向这个类。

class ServerHandler extends ChannelHandlerAdapter {
    /**
     * 当通道被调用,执行该方法
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 接收数据
        String value = (String) msg;
        System.out.println("Server msg:" + value);
        // 回复给客户端 “您好!”
        String res = "好的...";
        ctx.writeAndFlush(Unpooled.copiedBuffer(res.getBytes()));
    }
}

  
 
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14

2、开启服务

public class NettyServer {
    public static void main(String[] args) throws InterruptedException {
        // 1.创建2个线程,一个负责接收客户端连接, 一个负责进行 传输数据
        NioEventLoopGroup pGroup = new NioEventLoopGroup();
        NioEventLoopGroup cGroup = new NioEventLoopGroup();
        // 2. 创建服务器辅助类
        ServerBootstrap b = new ServerBootstrap();
        b.group(pGroup, cGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 1024)
                // 3.设置缓冲区与发送区大小
                .option(ChannelOption.SO_SNDBUF, 32 * 1024).option(ChannelOption.SO_RCVBUF, 32 * 1024)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel sc) throws Exception {
                        sc.pipeline().addLast(new StringDecoder());
                        sc.pipeline().addLast(new ServerHandler());
                    }
                });
        ChannelFuture cf = b.bind(8080).sync();
        System.out.println("服务器端已经启动....");
        cf.channel().closeFuture().sync();
        pGroup.shutdownGracefully();
        cGroup.shutdownGracefully();
    }
}

  
 
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24

此时在浏览器输入localhost:9090,看控制台打印信息,如果出现下面日志,说明服务端开启成功并可用。
在这里插入图片描述

客户端的实现

和服务端实现类似,只不过将其中一些对象换成客户端的对象。

回调

class ClientHandler extends ChannelHandlerAdapter {

    /**
     * 当通道被调用,执行该方法
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 接收数据
        String value = (String) msg;
        System.out.println("client msg:" + value);
    }
}

  
 
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12

连接服务器,并发送信息

public class NettyClient {

    public static void main(String[] args) throws InterruptedException {
        System.out.println("客户端已经启动....");
        // 创建负责接收客户端连接
        NioEventLoopGroup pGroup = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        b.group(pGroup).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel sc) throws Exception {
                sc.pipeline().addLast(new StringDecoder());
                sc.pipeline().addLast(new ClientHandler());
            }
        });
        ChannelFuture cf = b.connect("127.0.0.1", 8080).sync();
        cf.channel().writeAndFlush(Unpooled.wrappedBuffer("aaaa".getBytes()));
        cf.channel().writeAndFlush(Unpooled.wrappedBuffer("bbbb".getBytes()));
        // 等待客户端端口号关闭
        cf.channel().closeFuture().sync();
        pGroup.shutdownGracefully();
    }
}

  
 
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22

TCP粘包、拆包问题解决方案

粘包/拆包

一个完整的业务可能会被TCP拆分成多个包进行发送,也有可能把多个小的包封装成一个大的数据包发送,这个就是TCP的拆包和封包问题。
下面可以看一张图,是客户端向服务端发送包

  1. 第一种情况,Data1和Data2都分开发送到了Server端,没有产生粘包和拆包的情况。
  2. 第二种情况,Data1和Data2数据粘在了一起,打成了一个大的包发送到Server端,这个情况就是粘包。
  3. 第三种情况,Data2被分离成Data2_1和Data2_2,并且Data2_1在Data1之前到达了服务端,这种情况就产生了拆包。
    由于网络的复杂性,可能数据会被分离成N多个复杂的拆包/粘包的情况,所以在做TCP服务器的时候就需要首先解决拆包/

解决办法

消息定长,报文大小固定长度,不够空格补全,发送和接收方遵循相同的约定,这样即使粘包了通过接收方编程实现获取定长报文也能区分。

sc.pipeline().addLast(new FixedLengthFrameDecoder(10));

  
 
  • 1

包尾添加特殊分隔符,例如每条报文结束都添加回车换行符(例如FTP协议)或者指定特殊字符作为报文分隔符,接收方通过特殊分隔符切分报文区分。

ByteBuf buf = Unpooled.copiedBuffer("_mayi".getBytes());
sc.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, buf));

  
 
  • 1
  • 2

文章来源: blog.csdn.net,作者:小毕超,版权归原作者所有,如需转载,请联系作者。

原文链接:blog.csdn.net/qq_43692950/article/details/107475997

【版权声明】本文为华为云社区用户转载文章,如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。