Netty学习篇②

Channel、ChannelPipeline、ChannelHandlerContent发送数据的不同

// channel往回写数据

Channel channel = ctx.channel();

channel.writeAndFlush(Unpooled.copiedBuffer(new String("123").toCharArray(), CharsetUtil.UTF_8));

// pipeline写回数据

ChannelPipeline pipeline = ctx.pipeline();

pipeline.writeAndFlush(Unpooled.copiedBuffer(new String("123").toCharArray(), CharsetUtil.UTF_8));

// ctx写回数据

ctx.writeAndFlush(Unpooled.copiedBuffer(new String("123").toCharArray(), CharsetUtil.UTF_8));

三种方式的区别

前2种影响channel整个通道的channelHandler的事件
ctx.writeAndFlush只影响当前handler

netty入站出站handler执行顺序问题

总结：

1. InboundHandler顺序执行，OutboundHandler逆序执行

2. InboundHandler通过write方法与OutboundHandler进行交互

3. InboundHandler想要传递消息到OutboundHandler，OutboundHandler需要在InboundHandler之前添加到管道中，否则在InboundHandler中write(msg)的时候OutboundHandler可能还没注册到管道中（前提是用ChannelHandlerContent进行回写数据）

4. InboundHandler和OutBoundHandler角色的不同顺序不同；针对客户端而言，客户端是先发起请求在接收数据，所以是OutboundHandler > InboundHandler；针对服务端则反之

netty异步操作

netty中所有的IO操作都是异步的，意味着任何IO调用都会立即返回，通过ChannelFuture获得IO操作的结果和状态

ChannelFuture

ChannelFuture提供了IO操作的结果和状态

它继承了io.netty.util.concurrent包下的Future

间接继承了并发包下的Future

ChannelFuture的类图

注意：

不要在IO线程内调用future对象的sync/await方法，不能再ChannelHandler中调用sync/await方法，可能导致死锁问题

ChannelPromise

继承ChannelFuture，进一步拓展用于设置IO操作结果

TCP粘包拆包

TCP拆包

所谓拆包就是一个数据包拆分成多个包进行发送，就好比咱们在同一家店购买东西过多会拆成多个包裹进去发货类似，一个完成的数据包可能会被TCP拆分成多个包进行传输发送

TCP粘包

粘包则和拆包相反，将多个小包封装成一个大的数据包，就好比快递站收到包裹不会立马运输派送，会等到一定的量才会运输派送；客户端发送若干的数据包，服务端接收的时候粘合成一个包接收

// 客户端

public class HtNettyClient {

    private String ip;

    private int port;

    HtNettyClient(String ip, int port) {

        this.ip = ip;

        this.port = port;

    }

    public void start() {

        EventLoopGroup workGroup = new NioEventLoopGroup();

        try {

            Bootstrap bootstrap = new Bootstrap();

            bootstrap.group(workGroup)

                    .channel(NioSocketChannel.class)

                    .remoteAddress(new InetSocketAddress(ip, port))

                    .handler(new ChannelInitializer<SocketChannel>() {

                @Override

                protected void initChannel(SocketChannel ch) throws Exception {

                    ch.pipeline().addLast(new HtNettyClientHandler());

                }

            });

            ChannelFuture future = bootstrap.connect().sync();

            future.channel().closeFuture().sync();

        } catch (Exception e) {

        } finally {

            workGroup.shutdownGracefully();

        }

    }

    public static void main(String[] args) {

        HtNettyClient client = new HtNettyClient("127.0.0.1", 9000);

        client.start();

    }

}

// 客户端发送数据handler

public class HtNettyClientHandler extends ChannelInboundHandlerAdapter {

    @Override

    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        // 数据缓冲区

        ByteBuf byteBuf = null;

        byte[] req = ("hetangyuese" +System.getProperty("line.separator")).getBytes();

        for (int i= 0; i<10; i++) {

            byteBuf = Unpooled.buffer(req.length);

            byteBuf.writeBytes(req);

            ctx.writeAndFlush(byteBuf);

        }

 }

// 服务端

public class HtNettyServer {

    private int port;

    HtNettyServer(int port) {

        this.port = port;

    }

    public void start() {

        // 定义一个boss来处理请求

        EventLoopGroup bossGroup = new NioEventLoopGroup(1);

        EventLoopGroup workGroup = new NioEventLoopGroup();

        try {

            ServerBootstrap serverBootstrap = new ServerBootstrap();

            serverBootstrap.group(bossGroup, workGroup)

                    .option(ChannelOption.SO_BACKLOG, 1024)

                    .channel(NioServerSocketChannel.class)

                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override

                        protected void initChannel(SocketChannel ch) throws Exception {

                            ch.pipeline()

//                                    .addLast(new LineBasedFrameDecoder(3))

                                    .addLast(new HtNettyHandler());

                        }

                    });

            // 绑定端口

            ChannelFuture channelFuture = serverBootstrap.bind(port).sync();

            // 阻塞至直至关闭通道

            channelFuture.channel().closeFuture().sync();

        } catch (Exception e) {

            e.printStackTrace();

        } finally {

            bossGroup.shutdownGracefully();

            workGroup.shutdownGracefully();

        }

    }

    public static void main(String[] args) {

        HtNettyServer server = new HtNettyServer(9000);

        server.start();

    }

}

// 服务端handler

public class HtNettyHandler extends ChannelInboundHandlerAdapter {

    private int count;

    @Override

    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        // 如果没有定义解码器的话 默认转byteBuf

        ByteBuf byteBuf = (ByteBuf)msg;

        byte[] params = new byte[byteBuf.readableBytes()];

        byteBuf.readBytes(params);

        String body = new String(params, CharsetUtil.UTF_8);

        System.out.println("收到了请求，请求内容：" + body + ", 收到请求次数：" + ++count);

    }

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

        cause.printStackTrace();

        ctx.close();

    }

}

// 结果

收到了请求，请求内容：hetangyuese

hetangyuese

hetangyuese

hetangyuese

hetangyuese

hetangyuese

hetangyuese

hetangyuese

hetangyuese

hetangyuese

, 收到请求次数：1

粘、拆包

为何TCP会导致粘包和拆包？

TCP每次在发送数据时都是以流的形式进行传输，底层有一个缓冲区来存储发送的字节流，

1.当发送数据小于缓冲区的饱和的大小时，会发生粘包，粘成一个包发送至服务端（每个包之间的间隔时间短，包数据很小）；

2.当发送的数据大于缓冲区的阈值，则会拆分成多个包进行发送

3.服务端没及时读取缓冲区的数据，导致数据堆积，可能导致服务端粘包

4.发送方由于Nagle算法机制

Nagle算法

TCP/IP协议中，无论发送多少数据，总是要在数据前面加上协议头，同时，对方接收到数据，也需要发送ACK表示确认。为了尽可能的利用网络带宽，TCP总是希望尽可能的发送足够大的数据。（一个连接会设置MSS参数，因此，TCP/IP希望每次都能够以MSS尺寸的数据块来发送数据）。Nagle算法就是为了尽可能发送大块数据，避免网络中充斥着许多小数据块。

粘包、拆包解决方法

1. 设置定长消息（根据业务设置足够长）

2. 设置消息的边界（设置分隔符）

3. 使用带消息头的协议，消息头存储消息开始标识及消息的长度信息

4. 发送端关闭Nagle算法

Netty编、解码器

Decoder解码

主要对应的是ChannelInboundHandler，主要作用将字节数组转换成对象消息

Decoder解码常用抽象类

ByteToMessageDecoder

字节码转消息对象时需要检查缓冲区是否有足够的字节

@Override

protected void decode(ChannelHandlerContext ctx, ByteBuf in,List<Object> out) throws Exception {

    // 假使消息为int字节 则需要判断bytebuf字节是否大于等于4

    if (in.readableBytes() >= 4) {

        // 符合的话就add到list对象中（解析对象）

        out.add(in.readInt());

    }

}

ReplayingDecoder

继承了ByteToMessageDecoder，不需要检查缓冲区是否有足够多的数据，速度略慢于ByteToMessageDecoder；

tips: 根据项目的复杂程度合理选择，常用ByteToMessageDecoder

MessageToMessageDecoder

用于从一种消息解码到另外一种消息

解码器的具体实现（主要解决TCP底层的粘包和拆包问题）

DelimiterBasedFrameDecoder

指定消息分隔符的解码器（客户端发送数据所有的数据末尾都需要增加分隔符，否则服务端接收不到）

// 定义最大长度及需要的切割符

public DelimiterBasedFrameDecoder(int maxFrameLength, ByteBuf delimiter) {

    this(maxFrameLength, true, delimiter);

}

// 定义切割符

ByteBuf delimiter = Unpooled.copiedBuffer("自定义分隔符".getBytes());

/**

* maxFrameLength：解析分隔符最大长度

* stripDelimiter：true表示解析数据隐藏分隔符，反之解析数据末尾都会带上分隔符

* failFast： true表示超出maxLength立刻抛出异常，false则会解析完在抛出异常

* delimiter: 分隔符

*/

public DelimiterBasedFrameDecoder(

            int maxFrameLength, boolean stripDelimiter, boolean failFast,

            ByteBuf delimiter) {

    this(maxFrameLength, stripDelimiter, failFast, new ByteBuf[] {

        delimiter.slice(delimiter.readerIndex(), delimiter.readableBytes())});

}

// 客户端请求体

reqStr = "hello!_" +

                "My name is hanleilei !_" +

                "What is your name !_" +

                "How are you? !_"

        ;

// 服务端增加解析 以!_为分隔符

.addLast(new DelimiterBasedFrameDecoder(1024,Unpooled.copiedBuffer("!_".getBytes())))

// 结果

收到了请求，请求内容：hello, 收到请求次数：1

收到了请求，请求内容：My name is hanleilei , 收到请求次数：2

收到了请求，请求内容：What is your name , 收到请求次数：3

收到了请求，请求内容：How are you? , 收到请求次数：4

// 服务端设置stripDelimiter为false

addLast(new DelimiterBasedFrameDecoder(1024, false, Unpooled.copiedBuffer("!_".getBytes())));

//结果

收到了请求，请求内容：hello!_, 收到请求次数：1

收到了请求，请求内容：My name is hanleilei !_, 收到请求次数：2

收到了请求，请求内容：What is your name !_, 收到请求次数：3

收到了请求，请求内容：How are you? !_, 收到请求次数：4

LineBasedFrameDecoder

以换行符为结束标志的解码器

// 如果发送的数据超过了maxLength还未解析到换行符则抛出TooLongFrameException异常

public LineBasedFrameDecoder(final int maxLength) {

	this(maxLength, true, false);

}

// 服务端定义最大长度为3的换行解码器

public void start() {

        // 定义一个boss来处理请求

        EventLoopGroup bossGroup = new NioEventLoopGroup(1);

        EventLoopGroup workGroup = new NioEventLoopGroup();

        try {

            ServerBootstrap serverBootstrap = new ServerBootstrap();

            serverBootstrap.group(bossGroup, workGroup)

                    .option(ChannelOption.SO_BACKLOG, 1024)

                    .channel(NioServerSocketChannel.class)

                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override

                        protected void initChannel(SocketChannel ch) throws Exception {

                            ch.pipeline()

                                    .addLast(new LineBasedFrameDecoder(3))

                                    .addLast(new HtNettyHandler());

                        }

                    });

            // 绑定端口

            ChannelFuture channelFuture = serverBootstrap.bind(port).sync();

            // 阻塞至直至关闭通道

            channelFuture.channel().closeFuture().sync();

        } catch (Exception e) {

            e.printStackTrace();

        } finally {

            bossGroup.shutdownGracefully();

            workGroup.shutdownGracefully();

        }

    }

// 客户端发送

public class HtNettyClientHandler extends ChannelInboundHandlerAdapter {

    @Override

    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        // 客户端注册是发送数据至服务端

        ByteBuf byteBuf = null;

        byte[] req = ("hetangyuese" +System.getProperty("line.separator")).getBytes();

        for (int i= 0; i<10; i++) {

            byteBuf = Unpooled.buffer(req.length);

            byteBuf.writeBytes(req);

            ctx.writeAndFlush(byteBuf);

        }

    }

}

// 异常

io.netty.handler.codec.TooLongFrameException: frame length (11) exceeds the allowed maximum (3)

	at io.netty.handler.codec.LineBasedFrameDecoder.fail(LineBasedFrameDecoder.java:146)

	at io.netty.handler.codec.LineBasedFrameDecoder.fail(LineBasedFrameDecoder.java:142)

	at io.netty.handler.codec.LineBasedFrameDecoder.decode(LineBasedFrameDecoder.java:99)

	at io.netty.handler.codec.LineBasedFrameDecoder.decode(LineBasedFrameDecoder.java:75)

	at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:489)

	at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:428)

	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:265)

	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)

	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)

	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)

	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1359)

	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)

	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)

	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:935)

	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:134)

	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:645)

	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580)

	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497)

	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459)

	at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858)

	at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:138)

	at java.lang.Thread.run(Thread.java:748)

// 修改长度

addLast(new LineBasedFrameDecoder(1024))

// 结果

收到了请求，请求内容：hetangyuese, 收到请求次数：1

收到了请求，请求内容：hetangyuese, 收到请求次数：2

收到了请求，请求内容：hetangyuese, 收到请求次数：3

收到了请求，请求内容：hetangyuese, 收到请求次数：4

收到了请求，请求内容：hetangyuese, 收到请求次数：5

收到了请求，请求内容：hetangyuese, 收到请求次数：6

收到了请求，请求内容：hetangyuese, 收到请求次数：7

收到了请求，请求内容：hetangyuese, 收到请求次数：8

收到了请求，请求内容：hetangyuese, 收到请求次数：9

收到了请求，请求内容：hetangyuese, 收到请求次数：10

FixedLengthFrameDecoder

固定长度的解码器

/**

* frameLength：解析的消息体长度，每次直解析frameLength长度字节消息

*/

public FixedLengthFrameDecoder(int frameLength) {

        if (frameLength <= 0) {

            throw new IllegalArgumentException(

                    "frameLength must be a positive integer: " + frameLength);

        }

        this.frameLength = frameLength;

    }

// 客户端

package com.hetangyuese.netty.client.handler;

import io.netty.buffer.ByteBuf;

import io.netty.buffer.Unpooled;

import io.netty.channel.ChannelHandlerContext;

import io.netty.channel.ChannelInboundHandlerAdapter;

/**

 * @program: netty-root

 * @description: 客户端逻辑类

 * @author: hewen

 * @create: 2019-10-22 17:15

 **/

public class HtNettyClientHandler extends ChannelInboundHandlerAdapter {

    @Override

    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        String reqStr = "";

        // 数据缓冲区

        ByteBuf byteBuf = null;

//        reqStr = "hetangyuese" + System.getProperty("line.separator");

        reqStr = "hello!_" +

                "My name is hanleilei !_" +

                "What is your name !_" +

                "How are you? !_"

        ;

        byte[] req = (reqStr).getBytes();

//        for (int i= 0; i<10; i++) {

        byteBuf = Unpooled.buffer(req.length);

        byteBuf.writeBytes(req);

        ctx.writeAndFlush(byteBuf);

//        }

    }

}

// 服务端解析（直接解析3个字节）

ch.pipeline().addLast(new FixedLengthFrameDecoder(3))

.addLast(new HtNettyHandler());

// 结果

收到了请求，请求内容：hel, 收到请求次数：1

收到了请求，请求内容：lo!, 收到请求次数：2

收到了请求，请求内容：_My, 收到请求次数：3

收到了请求，请求内容： na, 收到请求次数：4

收到了请求，请求内容：me , 收到请求次数：5

收到了请求，请求内容：is , 收到请求次数：6

收到了请求，请求内容：han, 收到请求次数：7

收到了请求，请求内容：lei, 收到请求次数：8

收到了请求，请求内容：lei, 收到请求次数：9

收到了请求，请求内容： !_, 收到请求次数：10

收到了请求，请求内容：Wha, 收到请求次数：11

收到了请求，请求内容：t i, 收到请求次数：12

收到了请求，请求内容：s y, 收到请求次数：13

收到了请求，请求内容：our, 收到请求次数：14

收到了请求，请求内容： na, 收到请求次数：15

收到了请求，请求内容：me , 收到请求次数：16

收到了请求，请求内容：!_H, 收到请求次数：17

收到了请求，请求内容：ow , 收到请求次数：18

收到了请求，请求内容：are, 收到请求次数：19

收到了请求，请求内容： yo, 收到请求次数：20

收到了请求，请求内容：u? , 收到请求次数：21

LengthFieldBasedFrameDecoder

消息包括 ：消息头 + 消息体，基于长度通用的解码器

/**

* maxFrameLength: 消息数据最大长度

*

* lengthFieldOffset：长度字段偏移位，长度字段开始的地方，跳过指定长度字节之后的消 * 息才是消息体字段，一般设置为0，从头部开始

*

* lengthFieldLength：消息头长度字段占的字节数，如果设为2则表示在

* lengthFieldOffset开始往后的2个字节存储消息体长度

*

* lengthAdjustment：调整消息体长度字段，如果消息包括消息头（即长度字段），如果需要* 去掉消息头则需要对应设置为负数（长度字段的字节长度），netty解析要减去对应的数值获取* 消息体

*

* initialBytesToStrip：是否需要剔除消息头，在获取到一个完整的数据包之后，去除长度 * 字节，直接拿到消息体的数据,为0代表不去除消息头

*

* failFast：是否快速失败，true代表在数据长度超出最大长度则立刻抛出异常

*/

public LengthFieldBasedFrameDecoder(

            int maxFrameLength, int lengthFieldOffset, int lengthFieldLength,

            int lengthAdjustment, int initialBytesToStrip, boolean failFast) {

        this(

                ByteOrder.BIG_ENDIAN, maxFrameLength, lengthFieldOffset, lengthFieldLength,

                lengthAdjustment, initialBytesToStrip, failFast);

    }

StringDecoder

文本解码器，将接收到的消息解码成字符串，一般与上述搭配使用，然后在后面加业务的handler

Encoder编码器

主要对应的是ChannelOutboundHandler，将消息对象转换为字节数组

Encoder解码常用的抽象解码类

MessageToByteEncoder

消息转为字节数组，write方法会判断是否支持消息类型，如果不支持则通过context传递到下一个ChannelOutboundHandler,自定义需要重写encode方法

@Override

public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {

        ByteBuf buf = null;

        try {

            // 判断是否支持的消息类型

            if (acceptOutboundMessage(msg)) {

                @SuppressWarnings("unchecked")

                I cast = (I) msg;

                buf = allocateBuffer(ctx, cast, preferDirect);

                try {

                    encode(ctx, cast, buf);

                } finally {

                    ReferenceCountUtil.release(cast);

                }

                if (buf.isReadable()) {

                    ctx.write(buf, promise);

                } else {

                    buf.release();

                    ctx.write(Unpooled.EMPTY_BUFFER, promise);

                }

                buf = null;

            } else {

                // 如果不支持则通过context传递到下一个ChannelOutboundHandler

                ctx.write(msg, promise);

            }

        } catch (EncoderException e) {

            throw e;

        } catch (Throwable e) {

            throw new EncoderException(e);

        } finally {

            if (buf != null) {

                buf.release();

            }

        }

    }

MessageToMessageEncoder：将一种消息编码为另外一种消息

ChannelOption常用的参数

SO_BACKLOG

用来设置服务端可连接的数量，当大量客户端连接时，服务端处理不过来的连接会等待，so_backlog的值可以用来控制等待连接的队列的大小

// 应用

.option(ChannelOption.SO_BACKLOG, 1024)

SO_KEEPALIVE

用来设置TCP连接检测连接状态，如果客户端和服务端长时间（2小时）未有数据传输，则会自动发送一个检测是否连接是否还活动的报文

// 应用

.option(ChannelOption.SO_KEEPALIVE, true)

SO_REUSEADDR

重复使用本地端口，某个端口被占用后设置这个属性则可以共同使用该端口

.option(ChannelOption.SO_REUSEADDR, true)

TCP_NODELAY

发送方由于Nagle算法，会把小数据包包装成更大的数据包一起发送，虽然有效的提高了网络的负载，但是却造成了延时，tcp_nodelay的作用就是禁用Nagle算法，nodelay非延迟即即时传输

.option(ChannelOption.TCP_NODELAY, true)

巴特西

Netty学习篇②

Channel、ChannelPipeline、ChannelHandlerContent发送数据的不同

三种方式的区别

netty入站出站handler执行顺序问题

netty异步操作

ChannelFuture

ChannelPromise

TCP粘包拆包

粘、拆包

Nagle算法

粘包、拆包解决方法

Netty编、解码器

Decoder解码

Decoder解码常用抽象类

解码器的具体实现（主要解决TCP底层的粘包和拆包问题）

Encoder编码器

Encoder解码常用的抽象解码类

ChannelOption常用的参数

最新文章

热门文章

巴特西

Netty学习篇②

Channel、ChannelPipeline、ChannelHandlerContent发送数据的不同

三种方式的区别

netty入站出站handler执行顺序问题

netty异步操作

ChannelFuture

ChannelPromise

TCP粘包拆包

粘、拆包

Nagle算法

粘包、拆包解决方法

Netty编、解码器

Decoder解码

Decoder解码常用抽象类

解码器的具体实现 （主要解决TCP底层的粘包和拆包问题）

Encoder编码器

Encoder解码常用的抽象解码类

ChannelOption常用的参数

最新文章

热门文章

解码器的具体实现（主要解决TCP底层的粘包和拆包问题）