Netty入门

Netty概述

Netty 是一个 Java 的开源框架,它的底层内部核心就是 NIO。Netty 提供了异步的、事件驱动的网络应用程序框架和工具,可以用来快速开发高性能、高可靠性的网络服务器和客户端工具。

NIO基础

首先我们讲一下 NIO 的三大组件,分别是 ChannelBufferSelector

Channel & Buffer

Channel 有一点类似于 stream,它就是读写数据的 双向通道,可以从 channel 将数据读入 buffer,也可以将 buffer 的数据写入 channel,而之前的 stream 要么是输入,要么是输出,channel 比 stream 更为底层。

graph LR
channel --> buffer
buffer --> channel

常见的 Channel

  • FileChannel
  • DatagramChannel
  • SocketChannel
  • ServerSocketChannel

常见的 buffer

  • ByteBuffer
    • MappedByteBuffer
    • DirectByteBuffer
    • HeapByteBuffer
  • ShortBuffer
  • IntBuffer
  • LongBuffer
  • FloatBuffer
  • DoubleBuffer
  • CharBuffer

多路复用器Selector

多路复用器 Selector 提供了选择已经就绪的任务的能力。之前我们讲的 Channel 它们会注册到 Selector 上面。然后 Selector 会不停的轮询注册在它上面的 Channel,如果某个 Channel 上面发生读或者写事件,这个 Channel 就处于了就绪状态,会被 Selector 轮询出来,然后通过 SelectionKey(一个特定的通道对象和一个特定的选择器对象之间的注册关系) 可以获取到就绪的 Channel 集合,进行后续的 IO 操作。

一个多路复用器 Selector 可以同时轮询多个 Channel,它可以使用更少的线程来处理通道信息。原理图如下:

嗨客网

Netty三大组件总结

结合上诉三大组件,大致的流程图如下:

嗨客网

  1. Channel 调用 register 方法,注册到 Selector 上面,会被分配 SelectionKey。它将 Channel 和 Selector 之间建立了关系。
  2. Selector 上面会有一个死循环,监听其上面注册到 channel 是否有读或者写事件。Selector 会根据 SelectedKey 获取里面相应的 Channel,并且处理里面的数据。
  3. 处理完成后,将 SelectedKey 删除,监听等待下次事件的到来。

读取文件案例

有一普通文本文件 data.txt,内容为

1234567890abcd

使用 FileChannel 来读取文件内容,代码如下:

@Slf4j
public class ChannelDemo1 {
    public static void main(String[] args) {

        try{
            RandomAccessFile randomAccessFile = new RandomAccessFile("data.txt","rw");
            FileChannel channel = randomAccessFile.getChannel();
            ByteBuffer buffer = ByteBuffer.allocate(10);
            do{
                // 从channel写数据到buffer
                int len = channel.read(buffer);
                System.out.println("读取的字节数:" + len);
                if (len == -1) {
                    break;
                }
                // 切换 buffer 读模式
                buffer.flip();
                while(buffer.hasRemaining()) {
                    System.out.println((char)buffer.get());
                }
                // 切换 buffer 写模式
                buffer.clear();
            }while (true);
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

运行结果如下:

ByteBuffer 结构

ByteBuffer 有以下重要属性

  • capacity
  • position
  • limit

在这里插入图片描述

写模式下,position 是写入位置,limit 等于容量,下图表示写入了 4 个字节后的状态

在这里插入图片描述

flip 动作发生后,position 切换为读取位置,limit 切换为读取限制

在这里插入图片描述

读取 4 个字节后,状态

在这里插入图片描述

clear 动作发生后,状态

在这里插入图片描述

compact 方法,是把未读完的部分向前压缩,然后切换至写模式

在这里插入图片描述

ByteBuffer API

写入数据

写入数据两种方法:

  • int writeBytes = channel.read(buf);
  • buffer.put((byte)127);

读取数据

读取数据两种方法:

  • int writeBytes = channel.write(buf);

  • byte b = buf.get();

    ps:get 方法会让 position 读指针向后走,如果想重复读取数据,可以采用以下方法:

    • 可以调用 rewind 方法将 position 重新置为 0。
    • 调用 get(int i) 方法获取索引 i 的内容,它不会移动读指针。

mark 和 reset

mark 是在读取时,做一个标记,即使 position 改变,只要调用 reset 就能回到 mark 的位置

注意

rewind 和 flip 都会清除 mark 位置

黏包/半包案例

网络上有多条数据发送给服务端,数据之间使用 \n 进行分隔
但由于某种原因这些数据在接收时,被进行了重新组合,例如原始数据有3条为

  • Hello,world\n
  • I'm zhangsan\n
  • How are you?\n

变成了下面的两个 byteBuffer (黏包,半包)

  • Hello,world\nI'm zhangsan\nHo
  • w are you?\n

创建工具类,代码如下:

package com.netty;

import io.netty.util.internal.StringUtil;

import java.nio.ByteBuffer;

import static io.netty.util.internal.MathUtil.isOutOfBounds;
import static io.netty.util.internal.StringUtil.NEWLINE;

public class ByteBufferUtil {
    private static final char[] BYTE2CHAR = new char[256];
    private static final char[] HEXDUMP_TABLE = new char[256 * 4];
    private static final String[] HEXPADDING = new String[16];
    private static final String[] HEXDUMP_ROWPREFIXES = new String[65536 >>> 4];
    private static final String[] BYTE2HEX = new String[256];
    private static final String[] BYTEPADDING = new String[16];

    static {
        final char[] DIGITS = "0123456789abcdef".toCharArray();
        for (int i = 0; i < 256; i++) {
            HEXDUMP_TABLE[i << 1] = DIGITS[i >>> 4 & 0x0F];
            HEXDUMP_TABLE[(i << 1) + 1] = DIGITS[i & 0x0F];
        }

        int i;

        // Generate the lookup table for hex dump paddings
        for (i = 0; i < HEXPADDING.length; i++) {
            int padding = HEXPADDING.length - i;
            StringBuilder buf = new StringBuilder(padding * 3);
            for (int j = 0; j < padding; j++) {
                buf.append("   ");
            }
            HEXPADDING[i] = buf.toString();
        }

        // Generate the lookup table for the start-offset header in each row (up to 64KiB).
        for (i = 0; i < HEXDUMP_ROWPREFIXES.length; i++) {
            StringBuilder buf = new StringBuilder(12);
            buf.append(NEWLINE);
            buf.append(Long.toHexString(i << 4 & 0xFFFFFFFFL | 0x100000000L));
            buf.setCharAt(buf.length() - 9, '|');
            buf.append('|');
            HEXDUMP_ROWPREFIXES[i] = buf.toString();
        }

        // Generate the lookup table for byte-to-hex-dump conversion
        for (i = 0; i < BYTE2HEX.length; i++) {
            BYTE2HEX[i] = ' ' + StringUtil.byteToHexStringPadded(i);
        }

        // Generate the lookup table for byte dump paddings
        for (i = 0; i < BYTEPADDING.length; i++) {
            int padding = BYTEPADDING.length - i;
            StringBuilder buf = new StringBuilder(padding);
            for (int j = 0; j < padding; j++) {
                buf.append(' ');
            }
            BYTEPADDING[i] = buf.toString();
        }

        // Generate the lookup table for byte-to-char conversion
        for (i = 0; i < BYTE2CHAR.length; i++) {
            if (i <= 0x1f || i >= 0x7f) {
                BYTE2CHAR[i] = '.';
            } else {
                BYTE2CHAR[i] = (char) i;
            }
        }
    }

    /**
     * 打印所有内容
     * @param buffer
     */
    public static void debugAll(ByteBuffer buffer) {
        int oldlimit = buffer.limit();
        buffer.limit(buffer.capacity());
        StringBuilder origin = new StringBuilder(256);
        appendPrettyHexDump(origin, buffer, 0, buffer.capacity());
        System.out.println("+--------+-------------------- all ------------------------+----------------+");
        System.out.printf("position: [%d], limit: [%d]\n", buffer.position(), oldlimit);
        System.out.println(origin);
        buffer.limit(oldlimit);
    }

    /**
     * 打印可读取内容
     * @param buffer
     */
    public static void debugRead(ByteBuffer buffer) {
        StringBuilder builder = new StringBuilder(256);
        appendPrettyHexDump(builder, buffer, buffer.position(), buffer.limit() - buffer.position());
        System.out.println("+--------+-------------------- read -----------------------+----------------+");
        System.out.printf("position: [%d], limit: [%d]\n", buffer.position(), buffer.limit());
        System.out.println(builder);
    }

    public static void main(String[] args) {
        ByteBuffer buffer = ByteBuffer.allocate(10);
        buffer.put(new byte[]{97, 98, 99, 100});
        debugAll(buffer);
    }

    private static void appendPrettyHexDump(StringBuilder dump, ByteBuffer buf, int offset, int length) {
        if (isOutOfBounds(offset, length, buf.capacity())) {
            throw new IndexOutOfBoundsException(
                    "expected: " + "0 <= offset(" + offset + ") <= offset + length(" + length
                            + ") <= " + "buf.capacity(" + buf.capacity() + ')');
        }
        if (length == 0) {
            return;
        }
        dump.append(
                "         +-------------------------------------------------+" +
                        NEWLINE + "         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |" +
                        NEWLINE + "+--------+-------------------------------------------------+----------------+");

        final int startIndex = offset;
        final int fullRows = length >>> 4;
        final int remainder = length & 0xF;

        // Dump the rows which have 16 bytes.
        for (int row = 0; row < fullRows; row++) {
            int rowStartIndex = (row << 4) + startIndex;

            // Per-row prefix.
            appendHexDumpRowPrefix(dump, row, rowStartIndex);

            // Hex dump
            int rowEndIndex = rowStartIndex + 16;
            for (int j = rowStartIndex; j < rowEndIndex; j++) {
                dump.append(BYTE2HEX[getUnsignedByte(buf, j)]);
            }
            dump.append(" |");

            // ASCII dump
            for (int j = rowStartIndex; j < rowEndIndex; j++) {
                dump.append(BYTE2CHAR[getUnsignedByte(buf, j)]);
            }
            dump.append('|');
        }

        // Dump the last row which has less than 16 bytes.
        if (remainder != 0) {
            int rowStartIndex = (fullRows << 4) + startIndex;
            appendHexDumpRowPrefix(dump, fullRows, rowStartIndex);

            // Hex dump
            int rowEndIndex = rowStartIndex + remainder;
            for (int j = rowStartIndex; j < rowEndIndex; j++) {
                dump.append(BYTE2HEX[getUnsignedByte(buf, j)]);
            }
            dump.append(HEXPADDING[remainder]);
            dump.append(" |");

            // Ascii dump
            for (int j = rowStartIndex; j < rowEndIndex; j++) {
                dump.append(BYTE2CHAR[getUnsignedByte(buf, j)]);
            }
            dump.append(BYTEPADDING[remainder]);
            dump.append('|');
        }

        dump.append(NEWLINE +
                "+--------+-------------------------------------------------+----------------+");
    }

    private static void appendHexDumpRowPrefix(StringBuilder dump, int row, int rowStartIndex) {
        if (row < HEXDUMP_ROWPREFIXES.length) {
            dump.append(HEXDUMP_ROWPREFIXES[row]);
        } else {
            dump.append(NEWLINE);
            dump.append(Long.toHexString(rowStartIndex & 0xFFFFFFFFL | 0x100000000L));
            dump.setCharAt(dump.length() - 9, '|');
            dump.append('|');
        }
    }

    public static short getUnsignedByte(ByteBuffer buffer, int index) {
        return (short) (buffer.get(index) & 0xFF);
    }
}

首先利用工具类,看下两个数据的存储结构,如下:

在这里插入图片描述

在这里插入图片描述

现在要求你编写程序,将错乱的数据恢复成原始的按 \n 分隔的数据,代码如下:

package com.netty;

import lombok.extern.slf4j.Slf4j;

import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

@Slf4j
public class ChannelDemo2 {
    public static void main(String[] args) {
        ByteBuffer source = ByteBuffer.allocate(32);
        //                     11            24
        source.put("Hello,world\nI'm zhangsan\nHo".getBytes());
        split(source);

        source.put("w are you?\n".getBytes());
        split(source);
    }

    private static void split(ByteBuffer source) {
        // 切换 buffer 读模式
        source.flip();
        int oldLimit = source.limit();
        for (int i = 0; i < oldLimit; i++) {
            if (source.get(i) == '\n') {
                System.out.println(i);
                // 初始一个新的目标buffer容器
                ByteBuffer target = ByteBuffer.allocate(i + 1 - source.position());
                // 0 ~ limit
                source.limit(i + 1);
                target.put(source); // 从source 读,向 target 写
                ByteBufferUtil.debugAll(target);
                source.limit(oldLimit);
            }
        }
        source.compact();
    }
}

运行如下:

在这里插入图片描述

网络编程

阻塞 I/O

模型

阻塞 I/O 的模型示意图如下:

在这里插入图片描述

我们可以看到,整个流程是一个串行的过程,从应用程序发起数据请求的时候,就会一直等待数据返回。应用程序在等待的过程中,不能做其他的事情。所以该流程被称为阻塞 I/O 流程。

优点

  • 进程阻塞挂起的时候,不会消耗 CPU 资源,可以及时的响应每个操作。
  • 实现难度比较低,开发接手实现比较容易。
  • 适合并发量比较小的网络应用开发。

缺点

  • 不适合并发量比较大的应用,因为每次请求都会阻塞进程。
  • 需要为每个 I/O 请求分配单独的进程或者线程,系统开销比较大。

案例

服务端代码:

package com.netty.server;

import com.netty.ByteBufferUtil;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;

public class ServerDemo1 {
    public static void main(String[] args) throws Exception{
        // 使用 nio 来理解阻塞模式, 单线程
        // 0. ByteBuffer
        ByteBuffer buffer = ByteBuffer.allocate(16);
        // 1. 创建了服务器
        ServerSocketChannel ssc = ServerSocketChannel.open();

        // 2. 绑定监听端口
        ssc.bind(new InetSocketAddress(8089));

        // 3. 连接集合
        List<SocketChannel> channels = new ArrayList<>();
        while (true) {
            // 4. accept 建立与客户端连接, SocketChannel 用来与客户端之间通信
            System.out.println("connecting...");
            SocketChannel sc = ssc.accept(); // 阻塞方法,线程停止运行
            System.out.println("connected... " + sc);
            channels.add(sc);
            for (SocketChannel channel : channels) {
                // 5. 接收客户端发送的数据
                System.out.println("before read... " + channel);
                channel.read(buffer); // 阻塞方法,线程停止运行
                buffer.flip();
                ByteBufferUtil.debugAll(buffer);
                buffer.clear();
                System.out.println("after read... " + channel);
            }
        }

    }
}

客户端代码

package com.netty.client;

import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

public class ClientDemo1 {
    public static void main(String[] args) throws Exception {
        SocketChannel sc = SocketChannel.open();
        sc.connect(new InetSocketAddress("localhost", 8089));
        sc.write(Charset.defaultCharset().encode("hello"));
        System.out.println("waiting...");

    }
}

运行结果:

在这里插入图片描述

非阻塞I/O描述

模型

非阻塞 I/O 模型示意图如下:

04 none block io.png

当用户进程发出 recvform 请求数据的时候,如果内核中的数据还没有准备好,它并不会阻塞等待,而是立马返回一个 error,客户端可以知道内核数据没有准备好。

这个时候,客户端可以做一些其他的事情,然后再发起一个 recvform 请求。重复上面的过程。这个过程通常被称之为轮询。轮询检查内核数据是否准备好,直到数据准备好,再拷贝数据到进程,进行数据处理。

ps:需要注意,拷贝数据整个过程,进程仍然是属于阻塞的状态。

优点

  • 实现难度相对较低。

缺点

  • 进程轮询,消耗 CPU 资源。
  • 适合并发量比较小且不需要及时响应的网络应用开发。

案例

服务端代码:

package com.netty.server;

public class ServerDemo2 {
    public static void main(String[] args) throws Exception {
        // 使用 nio 来理解非阻塞模式, 单线程
        // 0. ByteBuffer
        ByteBuffer buffer = ByteBuffer.allocate(16);
        // 1. 创建了服务器
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false); // 非阻塞模式
        // 2. 绑定监听端口
        ssc.bind(new InetSocketAddress(8090));
        // 3. 连接集合
        List<SocketChannel> channels = new ArrayList<>();
        while (true) {
            // 4. accept 建立与客户端连接, SocketChannel 用来与客户端之间通信
            SocketChannel sc = ssc.accept(); // 非阻塞,线程还会继续运行,如果没有连接建立,但sc是null
            System.out.println("connecting... " + sc);
            if (sc != null) {
                System.out.println("connected... " + sc);
                sc.configureBlocking(false); // 非阻塞模式
                channels.add(sc);
            }
            for (SocketChannel channel : channels) {
                // 5. 接收客户端发送的数据
                int read = channel.read(buffer);// 非阻塞,线程仍然会继续运行,如果没有读到数据,read 返回 0
                System.out.println("before read... ... " + sc);
                if (read > 0) {
                    buffer.flip();
                    ByteBufferUtil.debugAll(buffer);
                    buffer.clear();
                    System.out.println("after read... " + channel);
                }
            }
        }

    }
}

客户端代码

package com.netty.client;

import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

public class ClientDemo2 {
    public static void main(String[] args) throws Exception {
        SocketChannel sc = SocketChannel.open();
        sc.connect(new InetSocketAddress("localhost", 8090));
        sc.write(Charset.defaultCharset().encode("hello"));
        System.out.println("waiting...");
    }
}

运行结果:

在这里插入图片描述

Selector

graph TD
subgraph selector 版
thread --> selector
selector --> c1(channel)
selector --> c2(channel)
selector --> c3(channel)
end

优点

  • 一个线程配合 selector 就可以监控多个 channel 的事件,事件发生线程才去处理。避免非阻塞模式下所做无用功
  • 让这个线程能够被充分利用
  • 节约了线程的数量
  • 减少了线程上下文切换

创建

Selector selector = Selector.open();

绑定 Channel 事件

也称之为注册事件,绑定的事件 selector 才会关心

channel.configureBlocking(false);
SelectionKey key = channel.register(selector, 绑定事件);
  • channel 必须工作在非阻塞模式
  • FileChannel 没有非阻塞模式,因此不能配合 selector 一起使用
  • 绑定的事件类型可以有
    • connect - 客户端连接成功时触发
    • accept - 服务器端成功接受连接时触发
    • read - 数据可读入时触发,有因为接收能力弱,数据暂不能读入的情况
    • write - 数据可写出时触发,有因为发送能力弱,数据暂不能写出的情况

监听 Channel 事件

可以通过下面三种方法来监听是否有事件发生,方法的返回值代表有多少 channel 发生了事件

方法1,阻塞直到绑定事件发生

int count = selector.select();

方法2,阻塞直到绑定事件发生,或是超时(时间单位为 ms)

int count = selector.select(long timeout);

方法3,不会阻塞,也就是不管有没有事件,立刻返回,自己根据返回值检查是否有事件

int count = selector.selectNow();

案例

服务器端

@Slf4j
public class ChannelDemo6 {
    public static void main(String[] args) {
        try (ServerSocketChannel channel = ServerSocketChannel.open()) {
            channel.bind(new InetSocketAddress(8080));
            System.out.println(channel);
            Selector selector = Selector.open();
            channel.configureBlocking(false);
            // 将服务端通道注册到选择器上,并指定注册监听的事件为OP_ACCEPT
            channel.register(selector, SelectionKey.OP_ACCEPT);

            while (true) {
                int count = selector.select();
                log.debug("select count: {}", count);
                // 获取所有事件
                Set<SelectionKey> keys = selector.selectedKeys();

                // 遍历所有事件,逐一处理
                Iterator<SelectionKey> iter = keys.iterator();
                while (iter.hasNext()) {
                    SelectionKey key = iter.next();
                    // 判断事件是否是客户端连接事件
                    if (key.isAcceptable()) {
                        ServerSocketChannel c = (ServerSocketChannel) key.channel();
                        // 必须处理
                        SocketChannel sc = c.accept();
                        sc.configureBlocking(false);
                        sc.register(selector, SelectionKey.OP_READ);
                        log.debug("连接已建立: {}", sc);
                    } 
                    // 判断是否是客户端读就绪事件SelectionKey.isReadable()
                    else if (key.isReadable()) {
                        SocketChannel sc = (SocketChannel) key.channel();
                        ByteBuffer buffer = ByteBuffer.allocate(128);
                        int read = sc.read(buffer);
                        if(read == -1) {
                            key.cancel();
                            sc.close();
                        } else {
                            buffer.flip();
                            debug(buffer);
                        }
                    }
                    // 处理完毕,必须将事件移除
                    iter.remove();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

ps:事件发生后,要么处理,要么取消(cancel),不能什么都不做,否则下次该事件仍会触发,这是因为 nio 底层使用的是水平触发

客户端

public class Client {
    public static void main(String[] args) {
        try (Socket socket = new Socket("localhost", 8080)) {
            System.out.println(socket);
            socket.getOutputStream().write("world".getBytes());
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

Netty 入门

Netty概述

我们的文章明明是介绍 Netty 的,为什么前面花了很大的篇幅去介绍 NIO 呢?因为 Netty 是一个 Java 的开源框架,它的底层内部核心就是 NIO。Netty 提供了异步的、事件驱动的网络应用程序框架和工具,可以用来快速开发高性能、高可靠性的网络服务器和客户端工具。

Netty应用

Netty 在 Java 网络应用框架中的地位就好比:Spring 框架在 JavaEE 开发中的地位

以下的框架都使用了 Netty,因为它们有网络通信需求!

  • Cassandra - nosql 数据库
  • Spark - 大数据分布式计算框架
  • Hadoop - 大数据分布式存储框架
  • RocketMQ - ali 开源的消息队列
  • ElasticSearch - 搜索引擎
  • gRPC - rpc 框架
  • Dubbo - rpc 框架
  • Spring 5.x - flux api 完全抛弃了 tomcat ,使用 netty 作为服务器端
  • Zookeeper - 分布式协调框架

Netty优点

  1. 框架设计比较优雅,底层的模型切换比较随意,适应不同的网络协议要求。
  2. 提供了很多标准的协议,安全和编解码的支持。
  3. JDK 自带的 NIO 需要了解太多的概念,编程比较复杂。而 Netty 对 NIO 进行了很好的封装,把该踩的坑都踩完了,对 JDK 自身有的 Bug 进行了修复。
  4. Netty 的社区比较活跃,在很多开源框架中使用,已经有了很广泛的验证,健壮性也比较好。

Netty Hello World

开发一个简单的服务器端和客户端

  • 客户端向服务器端发送 hello, world
  • 服务器仅接收,不返回

服务器端

// 0.启动器,负责组装 netty 组件,启动服务器
new ServerBootstrap()
    // 1创建 NioEventLoopGroup
    .group(new NioEventLoopGroup()) 
     // 2 选择服务 Scoket 实现类,其中 NioServerSocketChannel 表示基于 NIO 的服务器端实现
    .channel(NioServerSocketChannel.class)
    // 3负责处理连接 worker(child) 负责处理读写。
    .childHandler(new ChannelInitializer<NioSocketChannel>() { 
        protected void initChannel(NioSocketChannel ch) {
           // 5 解码 ByteBuf => String(将ByteBuf转化为字符串)。
            ch.pipeline().addLast(new StringDecoder());
             // 6 SocketChannel 的业务处理器。
            ch.pipeline().addLast(new SimpleChannelInboundHandler<String>() {
                @Override
                protected void channelRead0(ChannelHandlerContext ctx, String msg) {
                    System.out.println(msg);
                }
            });
        }
    })
   // 7 绑定监听端口。
    .bind(8080); 

客户端

new Bootstrap()
    // 1 创建 NioEventLoopGroup,同 Server
    .group(new NioEventLoopGroup()) 
    // 2 选择客户 Socket 实现类。NIO 的客户端实现
    .channel(NioSocketChannel.class) 
    .handler(new ChannelInitializer<Channel>() {
        @Override
        protected void initChannel(Channel ch) {
            // 4 建立连接后的要处理逻辑。
            ch.pipeline().addLast(new StringEncoder()); 
        }
    })
    // 5 指定要连接的服务器和端口
    .connect("127.0.0.1", 8080) 
    .sync()
    .channel()
    // 6 写入消息并清空缓冲区
    .writeAndFlush(new Date() + ": hello world!"); 

流程图总结

在这里插入图片描述

TCP的粘包拆包

概念

TCP 是一个 “流” 协议,所谓的流,就是没有界限的一个长串的二进制数据。TCP 的底层并不了解上层业务的数据的具体意义,它会根据 TCP 缓冲区的实际情况进行包的划分,所以按照人的思维在业务上传递的数据是一个完整的数据,有可能会被 TCP 拆分成多个包进行发送,也有可能把多个小的包封装成一个大的数据包发送。将多个包合并成一个包我们就叫做粘包,把一个包分成多个包我们叫做拆包。

概念解析

其实 TCP 粘包拆包和我们现实生活中的问题很像,比如你将一个大的物件从 A 地点运送到 B 地点。如果车比较小,你每次只能将物件上面的物件拆分出来,拆成一个一个小的物件,通过车多次运输。这个就是我们理解的拆包。还有一种场景就是你要把两个比较小的物件从 A 地点运送到 B 地点。其实一趟就可以将两个物件运送过去,不需要分两次,所以就将两个物件一起运输过去。这个就是我们所谓的粘包。

在这里插入图片描述

  1. 服务器端分两次读取到了客户端发送端数据 Data1 和 Data2,数据都是完整的,没有 TCP 拆包和 TCP 粘包
  2. 服务端只接受了一次数据,但是将 Data2 和 Data1 这两个数据都获取到了,这种场景称为 TCP 粘包。
  3. 服务端分两次接受到了客户端发来的数据,第一次收到了 Data1 的完整数据和 Data2 的部分数据,第二次收到了 Data2 的剩余的数据。将 Data2 分成两批传输,这就叫 TCP 拆包。当然,我们看到 Data1 和 Data2 的部分数据一起,这边的逻辑是 TCP 粘包。
  4. 如果服务端 TCP 接受数据的缓冲区比较小,那么就将 Data1 和 Data2 都划分成小的数据块传送到服务端,这就会将数据进行多次拆分,这个部分就叫做拆包。

Netty编解码

那么上诉的 TCP 粘包/拆包,在Netty 是怎么处理呢?这时候Netty 提供了编解码技术来处理。

那么在前文中,我们的 NIO 案例中,也是处理过 TCP 粘包/拆包这个问题的,JAVA 也本身也是自带序列化和反序列化的功能。但是它的本身缺陷太多了。它没有办法跨语言并且序列化后流太大,序列化性能太低。

所以 Netty 虽然是 Java 语言编写的高性能通信框架,但是它没有使用 Java 自带的序列化机制,自己提供了编解码接口。

解决方案

TCP 以流的方式进行数据传输,上层的应用协议为了对消息进行区分,我们了解了 TCP 的粘包和拆包机制,它主要有以下四种解决方案:

  1. 消息长度固定,累计读取到长度总和为定长 LEN 的报文后,就认为读取到了一个完整的消息,将计数器置位,重新读取下一个数据报。
  2. 将回车换行符作为消息结束符。
  3. 将特殊的分隔符作为消息的结束标志,回车换行符是一个钟特殊的结束分隔符。
  4. 通过在消息头中定长度字段来标识消息的总长度。

Netty 针对上面 4 种方式,每种方式都提供了相应的处理方法,

  • 消息定长度:FixedLengthFrameDecoder
  • 回车换行符作为结束标识:LineBasedFrameDecoder
  • 特殊分割符:DelimiterBasedFrameDecoder
  • 通用解码器:LengthFieldBasedFrameDecoder

Netty服务端

具体时序图如下:

在这里插入图片描述

Netty客户端

在这里插入图片描述

本文参考资料如下:

黑马程序员Netty全套教程,全网最全Netty深入浅出教程

嗨客网Netty教程

Logo

为开发者提供学习成长、分享交流、生态实践、资源工具等服务,帮助开发者快速成长。

更多推荐