Netty

1 IO

1.1 BIO

1.1.1 介绍

Java BIO就是传统的Java IO编程，相关API都在java.io
BIO (blocking I/O)：同步并阻塞，服务器实现模式为一个连接一个线程，即有连接请求时服务器就需要启动一个线程进行处理，如果这个连接不做任何事情会造成不必要的线程开销
BIO方式适用于连接数目较小且固定的架构，这种方式对服务器资源要求比较高，并发局限于应用中，JDK1.4以前的唯一选择

1.1.2 实例

BIO服务器

import org.apache.log4j.Logger;

import java.io.IOException;
import java.io.InputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author  KHighness
 * @date 2020/9/13 11:07
 * @apiNote : BIO服务器
 */
public class BIOServer {
    private final Logger log = Logger.getLogger(BIOServer.class);

    /**
     *  启动服务器
     */
    public void start() throws IOException {
        /**
         * 线程池机制
         * 思路
         * 1、创建一个线程池
         * 2、如果有多个线程池联机呃，就创建一个线程，与之通讯
         */
        ExecutorService newCachedThreadPool = Executors.newCachedThreadPool();
        // 创建ServerSocket
        ServerSocket serverSocket = new ServerSocket(3333);
        log.info("NIO服务器启动，监听端口：3333");
        while (true) {
            // 监听，等待客户端接受
            final Socket socket = serverSocket.accept();
            log.info("新增客户端连接：" + socket.getInetAddress());
            // 创建一个线程，与之通讯
            newCachedThreadPool.execute(new Runnable() {
                public void run() {
                    handler(socket);
                }
            });
        }
    }

    /**
     * handler，和客户端通讯
     */
    public void handler(Socket socket) {
        try {
            byte[] bytes = new byte[1024];
            // 通过socket获取输入流
            InputStream inputStream = socket.getInputStream();
            // 循环读取客户端发送的数据
            while (true) {
                // 输出线程信息
                log.info("线程信息：『PID = " + Thread.currentThread().getId() + ", Name = " + Thread.currentThread().getName() + "』");
                int read = inputStream.read(bytes);
                if (read != -1) {
                    log.info("客户端" + socket.getRemoteSocketAddress() + "：" + new String(bytes, 0, read));
                } else {
                    break;
                }
            }
        } catch (Exception e) {
            log.info(e.getMessage());
        } finally {
            log.info("关闭与客户端的连接");
            try {
                socket.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        try {
            new BIOServer().start();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

1.1.3 运行

Windows配置：控制面板 -> 程序 -> 启用或者关闭Windows功能，勾选Telnet Client。

启动服务器：

1	2021-06-05 17:14:56.456 KAG INFO [com.kag.bio.BIOServer] - NIO服务器启动，监听端口：3333

开启cmd：

1	telnet 127.0.0.1 6666

同时按下ctrl + ]，发送消息：

1.1.4 不足

1）每个请求都需要创建独立的线程，与对应的客户端进行数据Read，业务处理，数据Write

2）当并发数较大时，需要创建大量线程来处理连接，系统资源占用较大

3）连接建立后，如果当当前线程暂存没有数据可读，则线程就阻塞在Read上，造成线程资源浪费

1.2 NIO

1.2.1 介绍

Java NIO 全称 java non-blocking IO，是指JDK提供的核心API，是同步非阻塞的。
NIO 相关类都被放在 java.nio包及子包下，并且对原 java.io包中的很多类进行改写
NIO 三大核心部分：Channel(通道)、Buffer(缓冲区)、Selector(选择器)
NIO 是面向缓冲区或者面向块编程的。数据读取到一个它稍后处理的缓冲区，需要时可在缓冲区前后移动，这就增加处理过程中的灵活性，使用它可以提供非阻塞的高伸缩性网络

1.2.2 比较

BIO VS NIO

BIO以流的方式处理数据，而NIO以块的方式处理数据，块I/O的效率比流I/O高很多
BIO是阻塞的，NIO则是非阻塞的
BIO基于字节流和字符流进行操作，而NIO基于Channel(通道)和Buffer(缓冲区)进行操作，数据总是从通道读取到缓冲区中，或者从缓冲区写入到通道中。Selectors(选择器)用于监听多个通道的事件（比如：连接请求、数据到达等），因此使用单个线程就可以监听多个客户端通道

1.2.3 组件

三大组件：Selector、Channel和Buffer

关系图

说明

每个channel都会对应一个Buffer
Selector对应一个线程，一个线程对应多个channel
该图反应有三个channel注册到了该Selector程序
程序切换到哪个channel是有事件决定的，Event就是一个重要的概念
Selector会根据不同的事件，在各个通道上切换
Buffer就是一个内存块，底层是有一个数组
数据的读取写入都通过Buffer，这个和BIO不同。BIO要么是输入流，要么是输出流，不能双向，但是NIO的buffer是可以读也可以写，需要flip方法切换
channel是双向的，可以返回底层操作系统的情况，比如Linux，底层的操作系统通道就是双向的

1.2.3.1 Buffer

说明

缓冲区：缓冲区本质上是一个可以读写数据的内存块

常用Buffer子类：

ByteBuffer
ShortBuffer
CharBuffer
IntBuffer
LongBuffer
DoubleBuffer
FloatBuffer

Buffer类定义了所有的缓冲区都具有的四个属性来提供关于其所包含的数据元素

属性	描述
Capacity	容量，级可以容纳的最大数据量；在缓冲区创建时被设定且不能改变
Limit	表示缓冲区的当前重点，不能对缓冲区超过极限的位置进行读写操作，且极限是可以修改的
Position	位置，下一个要被读或写的元素的索引，每次读写缓冲区数据时都会改变的值，为下次读写作准备
Mark	标记

mark <= position <= limit <= capacity

1.2.3.2 Channel

说明

通道：NIO的通道类似于流，但有些区别

通道可以同时进行读写，而流只能读或者只能写
通道可以实现异步读写数据
通道可以从缓冲读数据，也可以写数据到缓冲

常用的Channel类：

FileChannel ━━━━━▶ 文件的数据读写
DatagramChannel ━━━━━▶ UDP的数据读写
ServerSocketChannel ━━━━━▶ TCP的数据读写
SocketChannel ━━━━━▶ TCP的数据读写

FileChannel的主要IO操作

方法	描述
public int read(ByteBuffer dst)	从通道读取数据并放到缓冲区中
public int write(ByteBuffer src)	把缓冲区的数据写到通道中
public long transferFrom(ReadableByteChannel src, long position, long count)	从目标通道中复制数据到当前通道
public long transferTo(long position, long count, WritableByteChannel target)	把数据从当前通道复制给目标通道

1.2.3.3 Selector

说明

选择器：

Java的NIO，用非阻塞的IO方式，可以用一个线程，处理多个的客户端连接，就会使用到Selector
Selector能够检测多个注册的通道上是否有事件发生（多个Channel以事件的方式可以注册到同一个Selector），如果有事件发生，便获取事件然后针对每个事件进行相应的处理。这样就可以只用一个单线程去管理多个通道，也就是管理多个连接和请求
只有在连接真正有读写事件发生时，才会进行读写，就大大地减少了系统开销，并且不必为每个连接都创建一个线程
避免了多线程之间的上下文切换导致的开销

类及相关方法

Class \| Method	描述
public abstract class Selector implements Closeable	抽象类
public static Selector open()	得到一个选择器对象
public int select(long timeout)	监控所有注册的通道，参数用来设置超时时间
public Set selectedKeys()	从内部集合中得到所有的SelectionKe’y

说明

当客户端连接时，会通过ServerSocketChannel得到SocketChannel
将SocketChannel注册到Selector上，注册后返回一个SelectionKey
Selector进行监听select方法，返回有事件发生的通道的个数
进一步得到各个SelectionKey
在通道SelectionKey反向获取SocketChannel，方法channel()
可以得到所有的channel，完成业务处理

1.2.3 编程

将数据写入到本地文件

public static void write(String s) throws IOException {
    // 创建输出流 ——> channel
    FileOutputStream fileOutputStream = new FileOutputStream("D:\\Java\\Test\\K1.txt");

    // 通过 fileOutputStream 获取 对应的 FileChannel
    // 这个 fileChannel 真实类型是  FileChannelImpl
    FileChannel fileChannel = fileOutputStream.getChannel();

    // 创建一个缓冲区 ByteBuffer
    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);

    // 将 s 放入 byteBuffer
    byteBuffer.put(s.getBytes());

    // 对 byteBuffer 进行flip
    byteBuffer.flip();

    // 将 byteBuffer 数据写入到 fileChannel
    fileChannel.write(byteBuffer);
    fileOutputStream.close();
}

从本地文件读取数据

public static void read(String path) throws IOException {
    // 创建文件的输入流
    File file = new File(path);
    FileInputStream fileInputStream = new FileInputStream(file);

    // 通过 fileInputStream 获取对应的 FileChannel ——>  实际类型 FileChannelImpl
    FileChannel fileChannel = fileInputStream.getChannel();

    // 创建一个缓冲区ByteBuffer
    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);

    // 将 通道的数据 读入到 Buffer
    fileChannel.read(byteBuffer);

    // 将byteBuffer的字节数据转成字符串
    System.out.println(new String(byteBuffer.array()));
    fileInputStream.close();
}

拷贝文件

public static void copy(String source, String destination) throws IOException {
    FileInputStream fileInputStream = new FileInputStream(source);
    FileChannel fileChannel1 = fileInputStream.getChannel();

    FileOutputStream fileOutputStream = new FileOutputStream(destination);
    FileChannel fileChannel2 = fileOutputStream.getChannel();

    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);

    while (true) {
        // 复位，重置标志位
        byteBuffer.clear();

        int read = fileChannel1.read(byteBuffer);
        if (read == -1) {// 表示读完
            break;
        }
        // 将buffer中的数据写入到fileChannel02 -- K2.txt
        byteBuffer.flip();
        fileChannel2.write(byteBuffer);
    }
    fileOutputStream.close();
    fileInputStream.close();
}

使用transferFrom完成拷贝

public static void copyImage(String fromPath, String toPath) throws IOException {
    // 创建相关流
    FileInputStream fileInputStream = new FileInputStream(fromPath);
    FileOutputStream fileOutputStream = new FileOutputStream(toPath);

    // 获取Channel
    FileChannel sourceChannel = fileInputStream.getChannel();
    FileChannel destinChannel = fileOutputStream.getChannel();

    // 使用transferFrom完成拷贝
    destinChannel.transferFrom(sourceChannel, 0, sourceChannel.size());

    // 关闭相关流
    sourceChannel.close();
    destinChannel.close();
    fileInputStream.close();
    fileOutputStream.close();
}

buffer数组完成读写操作

// Scattering: 将数据写入到buffer时，可以采用buffer数组，依次写入[分散]
// Gathering: 从buffer读取数据时，可以采用buffer数组，依次读取[聚集]
public static void demo() throws Exception{
    // 使用ServerSocketChannel 和 SocketChannel 网络
    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    InetSocketAddress inetSocketAddress = new InetSocketAddress(7000);

    // 绑定端口到Socket，并启动
    serverSocketChannel.socket().bind(inetSocketAddress);

    // 创建buffer数组
    ByteBuffer[] byteBuffers = new ByteBuffer[2];
    byteBuffers[0] = ByteBuffer.allocate(5);
    byteBuffers[1] = ByteBuffer.allocate(3);

    // 等客户端连接
    SocketChannel socketChannel = serverSocketChannel.accept();
    int messageLength = 8;

    // 循环的读取
    while (true) {
        int byteRead = 0;

        while (byteRead < messageLength) {
            long l = socketChannel.read(byteBuffers);
            byteRead  += l;
            System.out.println("byteRead = " + byteRead);
            Arrays.asList(byteBuffers)
                .stream()
                .map(buffer -> "position = "
                     + buffer.position()
                     + ", limit = "
                     + buffer.limit())
                .forEach(System.out::println);

            // 将所有的buffer反转
            Arrays.asList(byteBuffers).forEach(buffer -> buffer.flip());

            // 将数据读出显示到客户端
            long byteWrite = 0;
            while (byteWrite < messageLength) {
                socketChannel.write(byteBuffers);
                byteWrite += 1;
            }

            // 将所欲的buffer进行clear
            Arrays.asList(byteBuffers).forEach(buffer -> {
                buffer.clear();
            });

            System.out.println(
                "byteRead = " + byteRead
                + ", byteWrite = " + byteWrite
                + ", messageLength = " + messageLength);
        }
    }
}

1.2.4 实例

群聊系统Demo

编码步骤：

当客户端连接时，会通过ServerSocketChannel 得到 SocketChannel
Selector 进行监听 select 方法, 返回有事件发生的通道的个数.
将socketChannel注册到Selector上, register(Selector sel, int ops), 一个selector上可以注册多个SocketChannel
注册后返回一个 SelectionKey, 会和该Selector 关联(集合)
进一步得到各个 SelectionKey (有事件发生)
在通过 SelectionKey 反向获取 SocketChannel , 方法 channel()
判断该Channel的事件类型，对不同事件进行不同的业务处理

NIOServer

package com.kag.nio.groupchat;

import org.apache.log4j.Logger;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;

/**
 * @author KHighness
 * @date 2020/9/19 22:59
 * @apinote
 */

public class GroupChatServer {

    private Logger log = Logger.getLogger(GroupChatServer.class);

    private Selector selector;

    private ServerSocketChannel listenChannel;

    private final int PORT = 3333;

    public GroupChatServer() {
        try {
            // 1、获取选择器
            selector = Selector.open();
            // 2、获取通道
            listenChannel = ServerSocketChannel.open();
            // 3、绑定端口
            listenChannel.socket().bind(new InetSocketAddress(PORT));
            // 4、设置非阻塞
            listenChannel.configureBlocking(false);
            // 5、将通道注册到选择器，注册操作：“接收”
            listenChannel.register(selector, SelectionKey.OP_ACCEPT);
        } catch (IOException e) {
            log.info(e.getMessage());
        }
    }

    // 监听
    public void listen() {
        try {
            // 6、采用轮询的方式，查询获取“准备就绪”的注册过的操作
            while (true) {
                int count = selector.select();
                if (count > 0) {
                    // 7、获取当前选择器中所有注册的选择键
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext()) {
                        // 8、获取“准备就绪”的时间
                        SelectionKey selectionKey = iterator.next();

                        // 9、判断selectionKey是具体的什么事件
                        // 监听到accept事件
                        if (selectionKey.isAcceptable()) {
                            // 10、若接受的事件是“接收就绪”操作，就获取客户端连接
                            SocketChannel socketChannel = listenChannel.accept();
                            // 11、切换为非阻塞模式
                            socketChannel.configureBlocking(false);
                            // 将该通道注册到选择器上
                            socketChannel.register(selector, SelectionKey.OP_READ);
                            log.info("[" + socketChannel.getRemoteAddress().toString().substring(1) + "]上线");
                        }

                        // 监听到read事件
                        if (selectionKey.isReadable()) {
                            // 处理读 (专门方法)
                            readMessage(selectionKey);
                        }

                        // end：移除选择键，防止重复操作
                        iterator.remove();

                    }
                } else {
                    log.info("等待····");
                }
            }
        } catch (IOException e) {
            log.error(e.getMessage());
        }
    }

    // 读取客户端消息
    public void readMessage(SelectionKey selectionKey) {
        // 定义一个 SocketChannel
        SocketChannel socketChannel = null;
        try {
            // 13、获取该选择器上的“读就绪”状态的通道
            socketChannel = (SocketChannel) selectionKey.channel();
            // 14、读取数据
            ByteBuffer buffer = ByteBuffer.allocate(1024);

            int count = socketChannel.read(buffer);
            // 根据count的h值做处理
            if (count > 0) {
                // 把缓存区的数据转成字符串
                String msg = new String(buffer.array());
                // 输出该消息
                log.info("客户端-" + msg);
                // 向其他客户端转发消息(排除自己)，专门写一个方法来处理
                sendMessageToOtherClients(msg, socketChannel);
            }
        } catch (Exception e) {
            log.error(e.getMessage());
            try {
                log.info("[" + socketChannel.getRemoteAddress().toString().substring(1) + "]离线");
                // 取消注册
                selectionKey.cancel();
                // 关闭通道
                socketChannel.close();
            } catch (IOException ioException) {
                log.error(e.getMessage());
            }
        }
    }

    // 转发消息给其他客户
    private void sendMessageToOtherClients(String msg, SocketChannel self) throws IOException {
        log.info("服务器转发消息");
        // 遍历所有注册到 selector 上的SocketChannel，并排除self
        for (SelectionKey key : selector.keys()) {
            // 通过 key 取出 对应的 SocketChannel
            Channel targetChannel = key.channel();
            // 排除自己
            if (targetChannel instanceof SocketChannel && targetChannel != self) {
                // 转型
                SocketChannel dest = (SocketChannel) targetChannel;
                // 将msg存储到buffer
                ByteBuffer byteBuffer = ByteBuffer.wrap(msg.getBytes());
                // 将buffer的数据写入通道
                dest.write(byteBuffer);
            }
        }
    }

    public static void main(String[] args) {
        // 创建服务器对象
        GroupChatServer groupChatServer = new GroupChatServer();
        groupChatServer.listen();
    }
}

NIOClient

package com.kag.nio.groupchat;

import org.apache.log4j.Logger;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Scanner;

/**
 * @auther KHighness
 * @date 2020/9/20 0:53
 * @apiNote
 */

public class GroupChatClient {

    private Logger log = Logger.getLogger(GroupChatClient.class);

    private final String HOST = "127.0.0.1";

    private final int    PORT = 3333;

    private Selector selector;

    private SocketChannel socketChannel;

    private String username;

    // 构造器
    public GroupChatClient() throws IOException {
        selector = Selector.open();
        // 连接服务器
        socketChannel = socketChannel.open(new InetSocketAddress(HOST, PORT));
        // 设置非阻塞
        socketChannel.configureBlocking(false);
        // 将socketChannel注册到selector
        socketChannel.register(selector, SelectionKey.OP_READ);
        // 得到username
        username = socketChannel.getLocalAddress().toString().substring(1);
        log.info("[" + username + "]已就绪···");
    }

    // 向服务器发送消息
    public void sendMessage(String msg) {
        msg = username + ": " + msg;
        try {
            socketChannel.write(ByteBuffer.wrap(msg.getBytes()));
        } catch (IOException e) {
            log.error(e.getMessage());
        }
    }

    // 从服务器回复的消息
    public void readMessage() {
        try {
            int readChannels = selector.select();
            if (readChannels > 0) {
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    if (key.isReadable()) {
                        // 得到相关的通道
                        SocketChannel socketChannel = (SocketChannel) key.channel();
                        // 得到一个buffer
                        ByteBuffer buffer = ByteBuffer.allocate(1024);
                        // 读取
                        socketChannel.read(buffer);
                        // 把读到的缓冲区的数据转成字符串
                        String msg = new String(buffer.array());
                        log.info(msg);
                    }
                }
                // 删除当前 SelectionKey
                iterator.remove();
            } else {
                //log.error("无可用的通道···");
            }
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

    public static void main(String[] args) throws Exception{
        // 启动客户端
        GroupChatClient chatClient = new GroupChatClient();

        // 启动一个线程，每隔3秒，读取从服务器
        new Thread() {
            public void run() {
                while (true) {
                    chatClient.readMessage();
                    try {
                        Thread.currentThread().sleep(3000);
                    } catch (InterruptedException e) {
                        System.err.println(e.getMessage());
                    }
                }
            }
        }.start();

        // 发送数据给服务器
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNextLine()) {
            String s = scanner.nextLine();
            chatClient.sendMessage(s);
        }
    }
}

1.2.5 零拷贝

说明

零拷贝，是从OS(操作系统)的角度来说的。因为内核缓冲区之间，没有数据是重复的
零拷贝不仅仅带来更少的数据复制，还能带来其他的性能优势，更少的CPU缓存伪共享以及无CPU校验和计算

mmap

mmap通过内存映射，将文件映射到内核缓冲区，同时，用户空间可以共享内核空间的数据。这样，在进行网络传输时，就可以减少内核空间到用户空间的拷贝次数

mmap和sendFile的区别

mmp适合小数据量读写，sendFile适合大文件传输
mmap需要4次上下文切换，3次数据拷贝；sendFile需要3次上下文切换，最少2次数据拷贝
sendFile可以利用DMA(direct memory access: 直接内存拷贝)方式，减少CPU拷贝，mmap则不能(必须从内核拷贝到Socket缓冲区)

2 概述

🌐官网: Netty.io

2.1 介绍

Netty是一个异步事件驱动的网络应用程序框架，用于快速开发可维护的高性能协议服务器和客户端。

2.2 设计

适用于各种传输类型的统一API-阻塞和非阻塞套接字
基于灵活且可扩展的事件模型，可将关注点明确分离
高度可定制的线程模型-单线程，一个或多个线程池，例如SEDA
真正的无连接数据报套接字支持（从3.1开始）

2.3 性能

更高的吞吐量，更低的延迟
减少资源消耗
减少不必要的内存复制

2.4 架构

3 Reactor

3.1 介绍

反应器模式 | 分发者模式 | 通知者模式

基于I/O复用模型: 多个连接共用一个阻塞对象，应用程序只需要再一个阻塞对象等待，无需阻塞等待所有连接。当某个连接有新的数据可以处理时，操作系统通知应用程序，线程从阻塞状态返回，开始进行业务处理
基于线程池复用线程资源: 不必再为每个连接创建线程，键连接完成后的业务处理热舞分配给线程进行处理，一个线程可以处理多个连接的业务

3.2 组成

Reactor: Reactor在一个单独的线程中运行，负责监听和分发事件，分发给适当的处理程序来对IO事件做出反应。
Handlers: 处理程序执行I/O事件要完成的实际事件。Reactor通过调度适当的处理程序来响应I/O事件，处理程序执行非阻塞操作。

3.3 分类

单Reactor单线程 ━━━━━▶ 前台接待员和服务员是同一个人，全程为顾客服务
单Reactor多线程 ━━━━━▶ 1个前台接待员，多个服务员，接待员只负责接待
主从Reactor多线程 ━━━━━▶ 多个前台接待员，多个服务生

3.2.1 单Reactor单线程

图示

分析

优点：模型简单，没有多线程、进程通信、竞争的问题，全部都在一个线程中完成
缺点：性能问题，只有一个线程，无法发挥多核CPU的性能。Handler在处理某个连接上的业务时，整个进程无法处理其他连接事件，很容易导致性能瓶颈
缺点：可靠性问题，线程意外终止，或者进入死循环，会导致整个系统通信模块不可用，不能接收和处理外部消息，造成节点故障
使用场景：客户端的数量有限，业务处理非常快速

3.2.2 单Reactor多线程

图示

方案说明

Reactor对象通过select监控客户端请求事件，收到事件后，通过dispatch进行分发
如果建立连接请求，则由Acceptor通过accept处理连接请求，然后从黄健一个Handler对象处理完成连接后的各种事件
如果不是连接请求，则由Reactor分发调用连接对应的handler读取数据后，会分发给后面的Worker线程池的某个线程处理业务
Handler只负责响应事件，不做具体的业务处理，通过read读取数据后，会分发给后面的worker线程池的某个线程处理业务
Worker线程池会分配独立线程完成真正的业务，并将结果返回给Handler
Handler收到响应后，通过send将结果返回给Client

分析

优点：可以充分的利用多核CPU的处理能力
缺点：多线程数据共享和访问比较复杂，Reactor处理所有的事件的监听和响应，在单线程运行，在高并发场景容易成为性能瓶颈

3.2.3 主从Reactor多线程

图示

方案说明

Reactor主线程MainReactor对象通过select监听连接事件，收到事件后，通过Acceptor处理连接事件
当Acceptor处理连接事件后，MainReactor将连接分配给SubReactor
SubReactor将连接加入到连接队列进行监听，并创建Handler进行各种事件处理
当有新事件发生时，SubReactor就会调用对应的Handler处理
Worker线程池分配独立的Worker线程进行业务处理，并返回结果
Handler收到响应的结果后，再通过send将结果返回给Client
Reactor主线程可以对于多个Reactor子线程，即MainReactor，可以关联多个SubReactor

分析

优点：父线程与子线程的职责明确，父线程只需要接收新连接，子线程完成后续的业务处理
优点：父线程与子线程的数据交互简单，Reactor主线程只需要把新连接传给子线程，子线程无需返回数据
缺点：编程复杂度较高

3.4 优点

响应快，不必为单个同步时间所阻塞，虽然Reactor本身依然是同步的
可以最大程度的比曼复杂的多线程及同步问题，并且避免了多线程/进程的切换开销
扩展性好，可以方便的通过增加Reactor实例个数来充分利用CPU资源
复用性好，Reactor模型本身与具体事件处理逻辑无关，具有很高的复用性

4 架构

4.1 图示

4.2 说明

Netty抽象出两组线程池BossGroup专门负责接收客户端的连接，WorkerGroup专门负责网络的读写
BossGroup和WorkerGroup类型的本质都是NioEventLoopGroup
NioEventLoopGroup相当于一个事件循环组，这个组中含有多个事件循环，每一个事件循环是NioEventLoop
NioEventLoop表示一个不断循环的执行处理任务的线程，每个NioEventLoop都有一个selector，用于监听绑定在其上的socket的网络通讯
NioEventLoopGroup可以有多个线程，即可以含有多个NioEventLoop
每个Boss NioEventLoop循环执行的步骤有三步:
1. 轮询accept事件
2. 处理accept事件，与client建立连接，生成NioSocketChannel，并将其注册到某个Worker NioEventLoop上的selector
3. 处理任务队列的任务，即runAllTasks
每个Worker NIOEventLoop循环执行的步骤
1. 轮询read、write事件
2. 处理i/o事件，即read、write事件，在对应NioSocketChannel处理
3. 处理任务队列的任务，即runAllTasks

4.3 快速开始

引入依赖(pom)

<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    <java.version>1.8</java.version>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <lombok.version>1.18.12</lombok.version>
    <log4j.version>1.2.17</log4j.version>
    <netty.version>4.1.2.Final</netty.version>
    <grpc.version>1.6.1</grpc.version>
    <protobuf.version>3.9.0</protobuf.version>
</properties>

<dependencies>

    <!-- lombok -->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>${lombok.version}</version>
        <scope>provided</scope>
    </dependency>

    <!-- log4j -->
    <dependency>
        <groupId>log4j</groupId>
        <artifactId>log4j</artifactId>
        <version>${log4j.version}</version>
    </dependency>

    <!-- netty-all -->
    <dependency>
        <groupId>io.netty</groupId>
        <artifactId>netty-all</artifactId>
        <version>${netty.version}</version>
    </dependency>

    <!-- protobuf-java -->
    <dependency>
        <groupId>com.google.protobuf</groupId>
        <artifactId>protobuf-java</artifactId>
        <version>${protobuf.version}</version>
    </dependency>

    <!-- grpc -->
    <dependency>
        <groupId>io.grpc</groupId>
        <artifactId>grpc-core</artifactId>
        <version>${grpc.version}</version>
    </dependency>
    <dependency>
        <groupId>io.grpc</groupId>
        <artifactId>grpc-protobuf-lite</artifactId>
        <version>${grpc.version}</version>
    </dependency>
    <dependency>
        <groupId>io.grpc</groupId>
        <artifactId>grpc-protobuf</artifactId>
        <version>${grpc.version}</version>
    </dependency>

</dependencies>

<build>
    <extensions>
        <extension>
            <groupId>kr.motd.maven</groupId>
            <artifactId>os-maven-plugin</artifactId>
            <version>1.4.1.Final</version>
        </extension>
    </extensions>
    <plugins>
        <plugin>
            <groupId>org.xolstice.maven.plugins</groupId>
            <artifactId>protobuf-maven-plugin</artifactId>
            <version>0.5.1</version>
            <configuration>
                <!-- 编译proto文件目录 -->
                <protoSourceRoot>${project.basedir}/src/main/java/com/kag/codec/proto</protoSourceRoot>
                <!-- 输出java文件目录-->
                <outputDirectory>${project.basedir}/src/main/java/com/kag/codec/generated</outputDirectory>
                <protocArtifact>com.google.protobuf:protoc:3.1.0:exe:${os.detected.classifier}</protocArtifact>
                <pluginId>grpc-java</pluginId>
            </configuration>
        </plugin>
    </plugins>
</build>

日志配置(log4j)

log4j.rootLogger = INFO,KAG,CONSOLE
#日志最低的输出级别
log4j.appender.KAG.Threshold=INFO
log4j.appender.KAG.encoding=UTF-8
#每天产生一个文件DailyRollingFileAppender
log4j.appender.KAG = org.apache.log4j.DailyRollingFileAppender
#日志文件的保存位置及文件名
log4j.appender.KAG.File=log/NettyPro.log
#有日志时立即输出，默认是true
log4j.appender.KAG.ImmediateFlush=true
log4j.appender.KAG.DatePattern='_'yyyy-MM-dd
#日志布局方式
log4j.appender.KAG.layout=org.apache.log4j.PatternLayout
#日志文件中日志的格式
log4j.appender.KAG.layout.ConversionPattern=%-d{yyyy-MM-dd HH:mm:ss.SSS} KAG %-5p [%c] - %m%n
#使用org.apache.log4j.ConsoleAppender指定要把日志输出到控制台
log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
log4j.appender.Threshold=INFO
#输出目标：控制台
log4j.appender.CONSOLE.Target=System.out
log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
log4j.appender.CONSOLE.layout.ConversionPattern=%-d{yyyy-MM-dd HH:mm:ss.SSS} KAG %-5p [%c] - %m%n

服务器

package com.kag.netty.simple;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

/**
 * @auther KHighness
 * @date 2020/9/21 21:53
 * @apiNote Netty服务器
 */
public class NettyServer {
    public static void main(String[] args) throws Exception{
        /*
         * 创建BossGroup 和 WorkerGroup
         * 说明：
         * 1、创建两个线程组：bossGroup 和 workerGroup
         * 2、bossGroup处理连接请求
         * 3、workerGroup处理和客户端的业务
         * 4、两个都是无限循环
         */
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            // 创建服务器端的启动对象，配置参数
            ServerBootstrap bootstrap = new ServerBootstrap();
            // 使用链式编程来进行设置
            bootstrap.group(bossGroup, workerGroup)                         // 设置两个线程组
                    .channel(NioServerSocketChannel.class)                  // 使用NioSocketChannel作为服务器的通道实现
                    .option(ChannelOption.SO_BACKLOG, 128)                  // 设置线程队列得到连接个数
                    .childOption(ChannelOption.SO_KEEPALIVE, true)          // 设置保持活动连接状态
                    .childHandler(new ChannelInitializer<SocketChannel>() { // 创建一个通道测试对象
                        // 给Pipeline设置处理器
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new NettyServerHandler());
                        }
                    });  // 给workerGroup的EventLoop对应的管道设置处理器

            System.out.println("服务器已就绪···");
            // 绑定一个端口并且同步
            // 绑定端口并启动服务器
            ChannelFuture channelFuture = bootstrap.bind(3333).sync();
            // 对关闭通道进行监听
            channelFuture.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

服务器业务处理器

package com.kag.netty.simple;

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

/**
 * @auther KHighness
 * @date 2020/9/21 22:20
 * @apiNote 服务器业务处理器
 * 自定义一个Handler需要继承Netty规定好的某个HandlerAdapter
 */
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    /**
     * 读取数据事件
     * @param ctx 上下文对象，含有管道pipline[处理数据]，通道channel[传输数据]，地址
     * @param msg 客户端发送的数据
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("server ctx = " + ctx);
        // 将msg转成ByteBuffer
        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println("客户端发送消息：" + byteBuf.toString(CharsetUtil.UTF_8));
        System.out.println("客户端地址：" + ctx.channel().remoteAddress());
    }

    /**
     * 数据读取完毕
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        // 将数据写入到缓存，并刷新
        ctx.writeAndFlush(Unpooled.copiedBuffer("Hello, 客户端~", CharsetUtil.UTF_8));
    }

    /**
     * 处理异常
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println(cause.getMessage());
        ctx.close();
    }
}

客户端

package com.kag.netty.simple;

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

/**
 * @auther KHighness
 * @date 2020/9/21 22:20
 * @apiNote Netty客户端
 * 自定义一个Handler需要继承Netty规定好的某个HandlerAdapter
 */
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    /**
     * 读取数据事件
     * @param ctx 上下文对象，含有管道pipline[处理数据]，通道channel[传输数据]，地址
     * @param msg 客户端发送的数据
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("server ctx = " + ctx);
        // 将msg转成ByteBuffer
        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println("客户端发送消息：" + byteBuf.toString(CharsetUtil.UTF_8));
        System.out.println("客户端地址：" + ctx.channel().remoteAddress());
    }

    /**
     * 数据读取完毕
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        // 将数据写入到缓存，并刷新
        ctx.writeAndFlush(Unpooled.copiedBuffer("Hello, 客户端~", CharsetUtil.UTF_8));
    }

    /**
     * 处理异常
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println(cause.getMessage());
        ctx.close();
    }
}

客户端业务处理

package com.kag.netty.simple;

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

/**
 * @auther KHighness
 * @date 2020/9/22 19:09
 * @apiNote 客户端业务处理
 */
public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    /**
     * 通道就绪->触发
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("client " +  ctx);
        ctx.writeAndFlush(Unpooled.copiedBuffer("Hello, server~喵", CharsetUtil.UTF_8));
    }

    /**
     * 有读取事件时->触发
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println("服务器回复消息：" + byteBuf.toString(CharsetUtil.UTF_8));
        System.out.println("服务器的地址：" + ctx.channel().remoteAddress());
    }

    /**
     * 处理异常
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println(cause.getMessage());
        ctx.close();
    }
}

4.4 任务队列

任务队列中的Task有三种典型使用场景

用户程序自定义的普通任务
用户自定义定时任务
非当前Reactor线程调用Channel的各种方法

4.5 异步模型

4.5.1 基本介绍

异步的概念和同步相对。当一个异步过程调用发出后，调用者不能立刻得到结果。实际处理这个调用的组件在完成后，通过状态、通知和回调来通知调用者
Netty中的I/O操作是异步的，包括Bind、Write、Connect等操作会简单的返回一个ChannelFuture
调用者并不能理科获得结果，而是通过Future-Listener机制，用户可以方便的主动获取或者通过通知机制获得IO操作结果
Netty的异步模型是建立在future和callback之上的。callback就是回调。重点是future，它的核心思想是：结社一个方法fun，计算过程可能非常耗时，等待fun返回显然不合适。那么可以在调用fun的时候立马返回一个future，后续可以通过future去监控方法fun的处理过程

4.5.2 Future说明

表示异步的执行结果，可以通过它提供的方法来检测执行是否完成
ChannelFuture是一个接口: public interface ChannelFuture extends Future

我们可以添加监听器，当监听的事件发生时，就会通知到监听器

4.5.3 Future-Listener机制

当future对象刚刚创建时，处于非完成状态，调用者可以通过返回的ChannelFuture来获取操作执行的状态，注册监听函数来执行完成后的操作

常见操作：

方法	描述
isDone()	判断当前操作是否完成
isSuccess()	判断已完成的当前操作是否成功
getCause()	获取已完成的当前操作失败的原因
isCancelled()	判断已完成的当前操作是否被取消
addListener()	注册监听器，当操作已完成(isDone返回完成)，将会通知指定的监听器；如果Future对象已完成，则通知指定的监听器

例如：

// 绑定一个端口并且同步
// 绑定端口并启动服务器
ChannelFuture CF = bootstrap.bind(3333).sync();

// 给CF注册监听器
CF.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture channelFuture) throws Exception {
        if (CF.isSuccess()) {
            System.out.println("服务器监听端口[3333]成功");
        } else {
            System.out.println("服务器监听端口[3333]失败");
        }
    }
});

4.6 Http服务

服务器

package com.kag.netty.http;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

/**
 * @auther KHighness
 * @date 2020/9/26 10:34
 * @apiNote Netty-HTTP服务器
 */
public class TestServer {
    public static void main(String[] args) {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new TestServerInitializer());
            ChannelFuture CF = serverBootstrap.bind(3333).sync();
            CF.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            System.out.println(e.getMessage());
        } finally{
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

服务器初始化

package com.kag.netty.http;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpServerCodec;

/**
 * @auther KHighness
 * @date 2020/9/26 10:39
 * @apiNote 服务器初始化
 */
public class TestServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        // 得到管道
        ChannelPipeline pipeline = socketChannel.pipeline();
		// 添加处理器
        // 加入一个Netty提供的HttpServerCodeC codec => [coder - decoder]
        // 1、HttpServerCodeC: 处理http的编&解码器
        pipeline.addLast("MyHttpServerCodec", new HttpServerCodec());
        // 2、增加一个自定义handler
        pipeline.addLast("MyTestServerHandler", new TestHttpServerHandler());
    }
}

服务器业务处理器

package com.kag.netty.http;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpServerCodec;

/**
 * @auther KHighness
 * @date 2020/9/26 10:39
 * @apiNote 服务器业务处理器
 */
public class TestServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        // 向管道加入处理器

        // 得到管道
        ChannelPipeline pipeline = socketChannel.pipeline();

        // 加入一个Netty提供的HttpServerCodeC codec => [coder - decoder]
        // 1、HttpServerCodeC: 处理http的编&解码器
        pipeline.addLast("MyHttpServerCodec", new HttpServerCodec());
        // 2、增加一个自定义handler
        pipeline.addLast("MyTestServerHandler", new TestHttpServerHandler());
    }
}

5 核心

5.1 BootStrap、ServerBootStrap

说明

BootStrap意思是引导程序，一个Netty应用通常由一个BootStrap开始，主要作用是配置整个Netty程序，串联各个组件，Netty中BootStrap类是客户端程序的启动引导类，ServerBootStrap是服务器端启动引导类。

常用方法

方法名称	方法介绍
public ServerBootStrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup)	该方法用于服务器端，用来设置两个EventLoop
public B group(EventLoopGroup group)	该方法用于客户端，用来设置一个EventLoop
public B channel(Class<? extends C> channelClass)	该方法用来设置一个服务器端的通道实现
public B option(ChannelOption option, T value)	用来给ServerChannel添加配置
public ServerBootstrap childOption(ChannelOption childOption, T value)	用来给接收到的通道添加配置
public ServerBootstrap childHandler(ChannelHandler childHandler)	该方法用来设置业务处理类(自定义的handler)
public ChannelFuture bind(int inetPort)	该方法用于服务器端，用来设置占用的端口号
public ChannelFuture connect(String inetHost, int inetPort)	该方法用于客户端，用来连接服务器

5.2 Future、ChannelFuture

说明

Netty中所有的IO操作都是异步的，不能立刻得知消息是否被正确处理。但是可以过一会等它执行完或者直接注册一个监听，具体的实现就是通过Future和ChannelFuture，它们可以注册一个监听，当操作执行成功或失败时监听会自动触发注册的监听事件。

常用方法

方法名称	方法介绍
Channel channel()	返回当前正在进行IO操作的通道
ChannelFuture sync()	等待异步操作执行完毕，相当于将阻塞在当前

5.3 Channel

说明

Netty网络通信的组件，能够用于执行网络I/O操作
通过Channel可获得当前网络连接的通道的状态
通过Channel可获得网络连接的配置参数
Channel通过异步的网络I/O操作(比如：建立连接、读写和绑定端口)，异步调用意味着任何I/O调用都将立即返回，并且不保证在调用结束时所请求的I/O操作已完成
调用立即返回一个ChannelFuture实例，通过注册器到ChannelFuture上，可以在I/O操作成功、失败或取消时回调通知调用方
支持关联I/O操作与对应的处理程序
不同协议、不同的阻塞类型的连接都有不同的Channel类型与之对应

常见Channel类型

名称	介绍
NioSocketChannel	异步的客户端TCP Socket连接
NioServerSocketChannel	异步的服务器端TCP Socket连接
NioDatagramChannel	异步的UDP连接
NioSctpChannel	异步的客户端Sctp连接
NioSctpServerChannel	异步的Sctp服务器端连接，这些通道涵盖了UDP和TCP网络I/O以及文件I/O

5.4 Selector

说明

Netty基于Selector对象实现I/O多路复用，通过Selector一个线程可以监听多个连接的Channel事件
当向一个Selector中注册Channel后，Selector内部的机制就可以自动不断地查询(Select)这些注册的Channel是否有已就绪的I/O事件(比如：可读、可写、网络连接完成等)，这样程序就可以很简单地使用一个线程高效地管理多个Channel
同时，Netty中对Selector中的selectedKey集合进行了替换，它替换成了一个它自己实现的一个set集合，这样效率更高

5.5 ChannelHandler及其实现类

说明

ChannelHandler是一个接口，处理I/O事件或拦截I/O操作，并将其转发到ChannelPipeline(业务处理链)中的下一个处理程序
ChannelHandler本身并没有提供很多方法，因为这个接口有许多的方法需要实现，方便使用期间，可以继承它的子类

我们经常需要自定义一个Handler类取继承ChannelInboundHandlerAdapter，然后通过重写相应方法实现业务逻辑

一般需要重写的方法

public class ChannelInboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelInboundHandler {

    //通道注册事件
    @Skip
    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelRegistered();
    }

    //通道取消注册事件
    @Skip
    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelUnregistered();
    }

    //通道就绪事件
    @Skip
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelActive();
    }

    //通道断联事件
    @Skip
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelInactive();
    }

    //通道读取数据事件
    @Skip
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ctx.fireChannelRead(msg);
    }

    //通道数据读取完毕事件
    @Skip
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelReadComplete();
    }

    //用法事件触发
    @Skip
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        ctx.fireUserEventTriggered(evt);
    }

    //通道可写性更改事件
    @Skip
    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelWritabilityChanged();
    }

    //通道发生异常事件
    @Skip
    @Override
    @SuppressWarnings("deprecation")
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
            throws Exception {
        ctx.fireExceptionCaught(cause);
    }
}

ChannelInboundHandler 用于处理入站I/O事件
ChannelOutboundHandler 用于处理出战I/O事件

适配器

ChannelInboundHandlerAdapter 用于处理入站 I/O 事件
ChannelOutboundHandlerAdapter 用于处理出站 I/O 操作
ChannelDuplexHandler 用于处理入站和出站事件

5.6 Pipeline、Channel

说明

ChannelPipeline是一个Handler的集合，它负责处理和拦截inbound和outbound的事件和操作，相当于一个贯穿Netty的链（通俗的讲：ChannelPipeline是保存ChannelHandler的list，用于处理或拦截Channel的入站事件和出战操作）
ChanenelPipline实现了一种高级形式的拦截过滤器模式，使用户可以完全控制事件的处理方式，以及Channel中各个的ChannelHandler如何相互交互
在Netty中每个Channel都有且仅有一个ChannelPipeline与之对应，他们的组成关系如下
一个Channel包含一个ChannelPipeline，而ChannelPipeline中又维护了一个由ChannelHandlerContext组成的双向链表，并且每个ChannelHandlerContext中又关联着一个ChannelHandler
入站事件和出战事件在一个双向链表中，入站事件会从链表head往后传递到最后一个入站的handler，出站事件会从链表tail往前传递到最前一个出站的handler，两种类型的handler互不干扰

常用方法

方法名称	方法介绍
ChannelPipeline addFirst(ChannelHandler… handlers)	把一个业务处理类（handler）添加到链中的第一个位置
ChannelPipeline addLast(ChannelHandler… handlers)	把一个业务处理类（handler）添加到链中的最后一个位置

5.7 ChannelHandlerContext

说明

保存Channel相关的所有上下文信息，同时关联一个ChannelHandler对象
即ChannelHandlerContext中包含一个具体的事件处理器ChannelHandler，同时ChannelHandlerContext中也绑定了对应的Pipeline和Channel的信息，方便对ChannelHandler进行调用

常见方法

方法名称	方法介绍
ChannelFuture close()	关闭通道
ChannelOutboundInvoker flush()	刷新
ChannelFuture writeAndFlush(Object msg)	将数据写到ChannelPipeline中当前ChannelHandler的下一个ChannelHandler开始处理

5.8 ChannelOption

说明

Netty在创建Channel实例后，一般都需要设置ChannelOption参数

参数如下：

ChannelOption.SO_BACKLOG:
- 对应TCP/IP协议listen函数中的backlog参数，用来初始化服务器可连接队列大小
- 服务端处理客户端连接请求是顺序处理的，所以同一时间只能处理一个客户端连接。多个客户端来的时候，服务端将不能处理的客户端连接请求放在队列中等待处理，backlog参数指定了队列的大小
ChannelOption.SO_KEEPALIVE:
- 一直保持连接活动状态

5.9 EventLoopGroup和其实现类NioEventLoopGroup

说明

EventLoopGroup是一组EventLoop的抽象，Netty为了更好的利用多核CPU资源，一般会有多个EventLoop同时工作，每个EventLoop维护者一个Selector实例。
EventLoopGroup提供next接口，可以从组里面按照一定规则获取其中一个EventLoop来处理任务。在Netty服务器端编程中，我们一般都需要提供两个EventLoopGroup，BossGroup和WorkerGroup
通常一个服务端口即一个ServerSocketChannel对应一个Selector和一个EventLoop线程。BossEventLoop负责接收客户端的连接并将SocketChannel交给WorkerEventLoopGroup来进行IO处理
BossEventLoopGroup通常是一个单线程的EventLoop，EventLoop维护着一个注册了ServerSocketChannel的Selector实例BossEventLoop不断轮询Selector将连接事件分离出来
通常是OP_ACCEPT事件，然后将接收到的SocketChannel交给WorkerEventLoopGroup
WorkerEventLoopGroup会由next选择其中一个EventLoop来将这个SocketChannel注册到其维护的Selector并对其后续的IO事件进行处理

常用方法

方法名称	方法介绍
public NioEventLoopGroup()	构造方法
public Future<?> shutdownGracefully()	断开连接，关闭线程

5.10 Unpooled

说明

Netty提供一个专门用来操作缓冲区（即Netty的数据容器）的工具类

常用方法

方法名称	方法介绍
public static ByteBuf copiedBuffer(CharSequence string, Charset charset)	通过给定的数据和字符编码返回一个 ByteBuf 对象（类似于 NIO 中的 ByteBuffer 但有区别）

5.11 群聊系统

要求

编写一个 Netty 群聊系统，实现服务器端和客户端之间的数据简单通讯（非阻塞）
实现多人群聊
服务器：可以监测用户上线，离线，并实现消息转发功能
客户端：通过channel 可以无阻塞发送消息给其它所有用户，同时可以接受其它用户发送的消息(由服务器转发得到)

群聊服务器

package com.kag.netty.groupChat;

import com.kag.nio.groupchat.GroupChatClient;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

/**
 * @auther KHighness
 * @date 2020/10/2 8:19
 * @apiNote
 */

public class GroupChatServer {

    private int port;

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

    public void run() {
        // 创建两个线程组
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        ServerBootstrap serverBootstrap = new ServerBootstrap();

        serverBootstrap.group(workerGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 128)
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel channel) throws Exception {
                        // 获取pipeline
                        ChannelPipeline pipeline = channel.pipeline();;
                        // 向pipeline加入解码器
                        pipeline.addLast("decoder", new StringDecoder());
                        // 向pipeline加入编码器
                        pipeline.addLast("encoder", new StringEncoder());
                        // 加入自己的业务处理handler
                        pipeline.addLast(new GroupChatServerHandler());
                    }
                });

        try {
            System.out.println("▶-----Netty服务器启动-----◀");
            ChannelFuture channelFuture = serverBootstrap.bind(port).sync();

            // 监听关闭
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            System.out.println(e.getMessage());
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        new GroupChatServer(3333).run();
    }
}

服务器业务处理器

package com.kag.netty.groupChat;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.GlobalEventExecutor;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * @auther KHighness
 * @date 2020/10/2 8:33
 * @apiNote
 */
public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> {

    // 定义一个channel组，管理所有的channel
    // GlobalEventExecutor.INSTANCE 是全局的时间执行器，是一个单例
    EventExecutor executor;
    private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    private String now() {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        return simpleDateFormat.format(new Date());
    }

    /**
     * HandlerAdded
     * 表示连接建立，一旦连接，第一个被执行
     * @param ctx
     * @throws Exception
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        // 将该客户加入聊天的信息推送给其他在线的客户
        // 该方法会将channelGroup中的所有channel遍历，并发送消息
        channelGroup.writeAndFlush(now() + " [客户端]" + channel.remoteAddress() + "加入聊天\n");
        channelGroup.add(channel);
    }

    /**
     * HandlerRemoved
     * 表示断开连接，将XX客户离线信息推送给当前在线客户
     * @param ctx
     * @throws Exception
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        channelGroup.writeAndFlush(now() + " [客户端]" + channel.remoteAddress() + "离开聊天\n");
        System.out.println("Channel Group Size = " + channelGroup.size());
    }

    /**
     * channelActive
     * 表示channel处于活动状态，提示XX上线
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println(now() + " " + ctx.channel().remoteAddress() + "上线");
    }

    /**
     * channelInactive
     * 表示channel处于非活动状态
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println(now() + " " + ctx.channel().remoteAddress() + "离线");
    }

    /**
     * ChannelRead0
     * 转发消息
     * @param channelHandlerContext
     * @param s
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
        Channel channel = channelHandlerContext.channel();
        // 遍历ChannelGroup
        channelGroup.forEach(ch -> {
            if (channel != ch) // 不是当前channel，直接转发
                ch.writeAndFlush(now() + " [客户]" + channel.remoteAddress() + "：" + s + "\n");
            else // 当前channel是自己
                ch.writeAndFlush(now() + " [自己]: " + s + "\n");
        });
    }

    /**
     * 异常处理
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
    }

}

群聊客户端

package com.kag.netty.groupChat;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.net.SocketAddress;
import java.util.Scanner;

/**
 * @auther KHighness
 * @date 2020/10/2 9:22
 * @apiNote
 */
public class GroupChatClient {

    private final String host;
    private final int port;

    public GroupChatClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

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

        Bootstrap bootstrap = new Bootstrap()
                .group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel channel) throws Exception {
                        // 得到pipeline
                        ChannelPipeline pipeline = channel.pipeline();
                        // 加入相关handler
                        pipeline.addLast("decoder", new StringDecoder());
                        pipeline.addLast("encoder", new StringEncoder());
                        // 加入自定义的handler
                        pipeline.addLast(new GroupChatClientHandler());
                    }
                });

        try {
            ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
            Channel channel = channelFuture.channel();
            System.out.println("▶-----" + channel.localAddress().toString().substring(1) + "-----◀");
            // 输入信息
            Scanner scanner = new Scanner(System.in);
            while (scanner.hasNextLine()) {
                String msg = scanner.nextLine();
                // 通过channel发送到服务器端
                channel.writeAndFlush(msg + "\r\n");
            }
        } catch (InterruptedException e) {
            System.out.println(e.getMessage());
        } finally {
            group.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception{
        new GroupChatClient("127.0.0.1", 3333).run();
    }
}

客户端业务处理器

package com.kag.netty.groupChat;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

/**
 * @auther KHighness
 * @date 2020/10/2 10:06
 * @apiNote
 */

public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> {
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
        System.out.println(s.trim());
    }
}

5.12 心跳检测

要求

当服务器超过3秒没有读操作时，就提示读空闲
当服务器超过5秒没有写操作时，就提示写空闲
当服务器超过7秒没有读或写时，就提示读写空闲

服务器

package com.kag.netty.heartbeat;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.TimeUnit;

/**
 * @auther KHighness
 * @date 2020/10/2 10:25
 * @apiNote 服务器
 */
public class MyServer {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))  //为BossGroup中的请求添加日志处理Handler
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            /**
                             * 加入一个 netty 提供的 IdleStateHandler
                             * 说明
                             * 1、IdleStateHandler 是 netty 提供的检测空闲状态的处理器
                             * 2、long readerIdleTime：表示多长时间没有读，就会发送一个心跳检测包检测是否还是连接的状态
                             * 3、long writerIdleTime：表示多长时间没有写，就会发送一个心跳检测包检测是否还是连接的状态
                             * 4、long allIdleTime：表示多长时间没有读写，就会发送一个心跳检测包检测是否还是连接的状态
                             * 5、当 IdleStateEvent 触发后，就会传递给管道的下一个 Handler，通过调用（触发）下一个Handler的 userEventTriggered，在该方法区处理这个事件。
                             */
                            pipeline.addLast(new IdleStateHandler(3, 5, 7, TimeUnit.SECONDS));
                            //加入一个对空闲检测进一步处理的Handler（自定义）
                            pipeline.addLast(new MyServerHandler ());
                        }
                    });
            //启动服务器，设置为同步模式
            ChannelFuture channelFuture = serverBootstrap.bind(3333).sync();
            channelFuture.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

服务器业务处理器

package com.kag.netty.heartbeat;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleStateEvent;

/**
 * @auther KHighness
 * @date 2020/10/2 10:48
 * @apiNote 服务器业务处理器
 */
public class MyServerHandler extends ChannelInboundHandlerAdapter {
    /**
     * @param ctx 上下文
     * @param evt 事件
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            // 将evt向下转型
            IdleStateEvent event = (IdleStateEvent) evt;
            String eventType = null;
            switch (event.state()) {
                case READER_IDLE:
                    eventType = "读空闲";
                    break;
                case WRITER_IDLE:
                    eventType = "写空闲";
                    break;
                case ALL_IDLE:
                    eventType = "读写空闲";
                    break;
            }
            System.out.println(ctx.channel().remoteAddress() + "——>超时：" + eventType);

            // 如果发生空闲，直接关闭通道
//            ctx.channel().close();
//            System.out.println("服务器关闭通道···");
        }
    }
}

5.13 简单总结

服务器

创建两个线程组，bossGroup和workerGroup
创建服务器启动对象ServerBootStrap
链式编程配置ServerBootStrap的参数
1. group：设置线程组bossGroup和workerGroup
2. channel：设置通道实现，一般选择NioServerSocketChannel
3. option：设置可连接线程队列以及大小，一般选择SO_BACKLOG
4. childOption：设置保持活动连接状态，选择SO_KEEPALIVE
5. handler：给bossGroup设置Handler
6. childHandler：给workerGroup设置Hadnler
ServerBootStrap绑定端口，设置同步，并且监听通道关闭事件

客户端

创建一个线程组group
创建客户端启动对象BootStrap
链式编程配置BootStrap的参数
1. group：设置线程组group
2. channel：设置通道实现，一般选择NioSocketChannel
3. handler：设置Handler
BootStrap绑定端口，设置同步，并且监听通道关闭事件

常见Handler

Handler	描述
SimpleChannelInboundHandler	处理通信(服务器最常用)
IdleStateHandler	检测空闲状态(心跳检测)
WebSocketServerProtocolHandler	将http协议升级ws协议，保持长连接

Handler常用方法

Method	描述
handlerAdded(ChannelHandlerContext ctx)	连接建立，一旦建立连接，第一个被执行的方法
handlerRemoved(ChannelHandlerContext ctx)	连接断开，将XX客户离线信息推送给当前在线客户
channelActive(ChannelHandlerContext ctx)	表示channel处于活动状态，提示XX上线
channelInactive(ChannelHandlerContext ctx)	表示channel处于非活动状态，提示XX离线
channelRead0(ChannelHandlerContext channelHandlerContext, String s)	读取数据，并进行消息转发
exceptionCaught(ChannelHandlerContext ctx, Throwable cause)	异常处理
userEventTriggered(ChannelHandlerContext ctx, Object evt)	事件触发器。在`IdleStateHandler`后面加上一个触发器，可以检测心跳。

时隔7个月，咕嘎的protobuf召唤我续更了。

6 ProtoBuf

6.1 编码解码

基本介绍

编写网络应用程序时，因为数据在网络中传输的都是二进制字节码数据，在发送数据时就需要编码，接收数据时就需要解码。
codec(编解码器)的组成部分有两个：decoder(解码器)和encoder(编码器)。encoder负责把业务数据转换成字节码数据，decoder负责把字节码数据转换成业务数据。

Netty本身的编码解码的机制和问题分析

Netty提供了一些编解码器，StringEncoder(字符串数据编码器)、StringDecoder(字符串数据解码器)、ObjectEncoder(Java对象编码器)、ObjectDecoder(Java对象解码器)。但是底层使用的仍然是Java序列化技术，存在如下问题：

无法跨语言(比如服务器是Java写的，就要求客户端也必须是Java写的)
序列化后的体积太大，是二进制编码的5倍多
序列化性能太低

6.2 Protobuf

📓官方文档(VPN)：protocol-buffers

基本介绍

ProtoBuf是Google发布的开源项目，全称Google Protocol Buffers，是一种轻便搞笑的结构化数据存储格式，可以用于结构化数据串行化，或者说序列化。它很适合做数据存储或者RPC(远程过程调用 remote procedure call)数据交换格式。
ProtoBuf是以message的方式来管理数据的，支持跨平台、跨语言。
目前很多公司：http + json => tcp + protobuf

简单demo

syntax = "proto3"; // 版本

option java_outer_classname = "StudentPOJO"; // 生成外部类名同时也是文件名

// protobuf使用message管理数据
message Student {
  // 会在StudentPOJO外部类生成一个内部类Student，他是真正发送的POJO对象
  int32 id = 1; // Student类中有一个属性名字为id，类型为int32(protobuf整数)，1表示属性序号
  string name = 2;
}

服务器

package com.kag.codec;

import com.kag.codec.generated.StudentPOJO;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;

/**
 * @author KHighness
 * @since 2021-05-31
 */
public class NettyServer {
    static class NettyServerHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            StudentPOJO.Student student = (StudentPOJO.Student) msg;
            System.out.printf("student[id = %d, name = %s]\n", student.getId(), student.getName());
        }
    }

    public static void main(String[] args) {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast("decoder", new ProtobufDecoder(StudentPOJO.Student.getDefaultInstance()));
                            pipeline.addLast(new NettyServerHandler());
                        }
                    });
            ChannelFuture channelFuture = serverBootstrap.bind(3333).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

客户端

package com.kag.codec;

import com.kag.codec.generated.StudentPOJO;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufEncoder;

/**
 * @author KHighness
 * @since 2021-05-31
 */
public class NettyClient {
    static class NettyClientHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ctx.writeAndFlush(StudentPOJO.Student.newBuilder().setId(1).setName("KHighness").build());
        }
    }

    public static void main(String[] args) {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap clietBootstrap = new Bootstrap();
            clietBootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast("encoder", new ProtobufEncoder());
                            pipeline.addLast(new NettyClientHandler());
                        }
                    });
            ChannelFuture channelFuture = clietBootstrap.connect("127.0.0.1", 3333).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            group.shutdownGracefully();
        }
    }
}

6.3 多种类型

在proto文件中使用dataType标识类型

MyDataInfo

syntax = "proto3";
option optimize_for = SPEED; // 加速解析
option java_outer_classname = "MyDataInfo"; // 指定外部类名

message MyMessage {
  enum DataType {
    StudentType = 0;
    WorkerType = 1;
  }

  // 标识传递的是哪个枚举类型
  DataType data_type = 1;

  // 标识每次枚举类型最多只能出现其中一个，节省空间
  oneof dataBody {
    Student student = 2;
    Worker worker = 3;
  }
}

message Student {
  int32 id = 1;
  string name = 2;
}
message Worker {
  string name = 1;
  int32 age = 2;
}

服务器

package com.kag.codec;

import com.kag.codec.generated.MyDataInfo;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;

/**
 * @author KHighness
 * @since 2021-05-31
 */
public class NettyServer {
    static class NettyServerHandler extends SimpleChannelInboundHandler<MyDataInfo.MyMessage> {
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, MyDataInfo.MyMessage myMessage) throws Exception {
            // 根据DataType显示
            MyDataInfo.MyMessage.DataType dataType = myMessage.getDataType();
            if (dataType == MyDataInfo.MyMessage.DataType.StudentType) {
                MyDataInfo.Student student = myMessage.getStudent();
                System.out.printf("Student[id = %d, name = %s]\n", student.getId(), student.getName());
            } else if (dataType == MyDataInfo.MyMessage.DataType.WorkerType) {
                MyDataInfo.Worker worker = myMessage.getWorker();
                System.out.printf("Worker[name = %s, age = %d]\n", worker.getName(), worker.getAge());
            } else {
                System.out.println("传输类型不对");
            }
        }
    }

    public static void main(String[] args) {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast("decoder", new ProtobufDecoder(MyDataInfo.MyMessage.getDefaultInstance()));
                            pipeline.addLast(new NettyServerHandler());
                        }
                    });
            ChannelFuture channelFuture = serverBootstrap.bind(3333).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

客户端

package com.kag.codec;

import com.kag.codec.generated.MyDataInfo;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufEncoder;

import java.util.Random;

/**
 * @author KHighness
 * @since 2021-05-31
 * @apiNote
 */
public class NettyClient {
    static class NettyClientHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            Random random = new Random(System.currentTimeMillis());
            int randomValue = random.nextInt(2);
            MyDataInfo.MyMessage myMessage = null;
            if (randomValue == 0) { // 发送Student对象
                myMessage = MyDataInfo.MyMessage.newBuilder().setDataType(MyDataInfo.MyMessage.DataType.StudentType)
                        .setStudent(MyDataInfo.Student.newBuilder().setId(2).setName("FlowerK").build()).build();
            } else { // 发送Worker对象
                myMessage = MyDataInfo.MyMessage.newBuilder().setDataType(MyDataInfo.MyMessage.DataType.WorkerType)
                        .setWorker(MyDataInfo.Worker.newBuilder().setName("RubbishK").setAge(19).build()).build();
            }
            ctx.writeAndFlush(myMessage);
        }
    }

    public static void main(String[] args) {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap clietBootstrap = new Bootstrap();
            clietBootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast("encoder", new ProtobufEncoder());
                            pipeline.addLast(new NettyClientHandler());
                        }
                    });
            ChannelFuture channelFuture = clietBootstrap.connect("127.0.0.1", 3333).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            group.shutdownGracefully();
        }
    }
}

7. Handler

7.1 介绍

概述

再回顾ChannelHandler和ChannelPipeline的关系，可以将ChannelPipeline理解为一个双向链表，ChannelHandlerContext看做链表中的一个节点，ChannelHandler则为每个节点中保存的一个属性对象。

ChannelHandler充当了处理入站和出站数据的应用程序逻辑的容器。实现ChannelInboundHandler接口处理入站业务，实现ChannelOutboundHandler接口处理出站业务。
ChannelPipeline提供了ChannelHandler链的容器。以客户端应用程序为例，如果事件的运动方向是从客户端到服务端的，那我们称这些事件为出站的，即客户端发送给服务端的数据会通过Pipeline中一系列ChannelOutboundHandler，并被这些Handler处理，反之则称为入站。以服务端为例，接收数据的过程就是入站，发送数据的过程就是出站。

类图

InboundHandler处理入站，OutboundHandler处理出站。

一般来说，在我们接收数据时将数据解码后，就进行业务的相关处理，所以上图的入站的常用类更多。在数据出站时，一般我们只需要将数据编码后直接发出。

由于不可能知道远程节点是否会一次性发送一个完整的信息，tcp有可能出现粘包拆包的问题，ByteToMessageDecoder会对入站数据进行缓冲，直到它准备好被处理。

7.2 理解

如下图所示，橙色表示出站，绿色表示入站。

当我们的Netty服务器初始化时，在ChannelInitializer类中这样实现initChannel，就会获得上图的Handler链表：

@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
    ChannelPipeline pipeline = socketChannel.pipeline();
    pipeline.addLast(new LongToByteEncoder());  // out
    pipeline.addLast(new ByteToLongDecoder());  // in
    pipeline.addLast(new OutBoundHandler());    // out
    pipeline.addLast(new InBoundHandler());     // in
}

当一个请求来了的时候，首先会将请求发给pipeline中位于链表头部的Handler。

首先LongToByteEncoder接收到入站请求，但是它属于OutBound，所以收到入站请求时不作处理，直接转发给它的下一个ByteToLongDecoder，因为它属于InBound，所以需要将请求的数据进行处理，然后转发给下一个。之后又是OutBoundHandler，不作处理，最后到了InBoundHandler，进行业务处理。

如果需要返回数据，我们就调用writeAndFlush()方法。当他一调用，就会触发出站的请求，然后就由当前所在的Handler往回调用。往回调用的图中，如果遇到InBound就直接跳过，转发给下一个Handler，直到最后消息返回。

总结

InBoundHandler的调用顺序和pipeline的添加顺序相同
OutBoundHandler的调用顺序和pipeline的添加顺序相反
InBoundHandler一旦进行writeAndFlush，只有这个InBoundHandler之前添加的OutBoundHandler能够处理他
pipeline的末尾不能有OutBoundHandler，因为前后的InBoundHandler处理完数据返回消息调用writeAndFlush时，最后一个OutBoundHandler无法被调用。所以平常我们将OutBoundHandler写在前面，InBoundHandler写在后面。

7.3 常用

概述

入站解码，出栈编码，Netty提供了很多编解码器。

（1）解码器ByteToMessageDecoder

由于不可能知道远程节点是否会一次性发送一个完整的信息，TCP有可能出现粘包拆包的问题，这个类会对入站数据进行缓冲，直到它准备好被处理。

（2）解码器ReplayingDecoder

ReplayingDecoder扩展了ByteToMessageDecoder，使用这个leukemia，我们不必调用readableBytes()方法。泛型参数指定了用户状态管理的类型，其中Void代表不需要状态管理。

使用方便，但是也有局限性：

并不是所有的ByteBuf操作都被支持，如果调用了一个不被支持的方法，将会抛出一个UnsupportedOperationException。
ReplayingDecoder在某些情况下可能稍慢于ByteToMessageDecoder，例如网络缓慢并消息格式复杂时，消息会被拆成多个碎片。

（3）其他解码器

LineBasedFrameDecoder：使用行尾控制符（\n）作为分隔符类解析数据。
DelimiterBasedFrameDecoder：使用自定义的特殊字符作为消息的分隔符。
HttpObjectDecoder：HTTP数据解码器。
LengthFieldBasedFrameDecoder：通过制定长度来表示整包信息，这样就可以自动的处理黏包的半包消息。

注意

不论是解码器还是编码器，接收的消息类型必须与待处理的消息类型一致，否则该Handler不会被执行。
在解码器进行数据解码时，需要判断缓存区ByteBuf的数据是否足够，否则接收到的结果可能会与期望结果不一致。

7.3 实例

要求

使用自定义的编码器和解码器来说明Netty Handler调用机制
客户端发送Long类型数据到服务器，服务器发送Long类型数据到客户端

编解码器

/**
 * @apiNote Long=>Byte编码器
 */
public class LongToByteEncoder extends MessageToByteEncoder<Long> {
    private final Logger log = Logger.getLogger(LongToByteEncoder.class);
    /**
     * @param ctx  上下文
     * @param msg  出站数据
     * @param out  编码后的数据，传给下一个handler
     */
    @Override
    protected void encode(ChannelHandlerContext ctx, Long msg, ByteBuf out) throws Exception {
        log.info("LongToByteEncoder => 编码");
        log.info("msg: " + msg);
        out.writeLong(msg);
    }
}

/**
 * @apiNote Byte=>Long解码器
 */
public class ByteToLongDecoder extends ByteToMessageDecoder {
    private final Logger log = Logger.getLogger(ByteToLongDecoder.class);
    /**
     * @param ctx  上下文
     * @param in   入站数据
     * @param out  解码后的数据，传给下一个handler
     */
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        log.info("ByteToLongDecoder => 解码");
        // Long八个字节
        if (in.readableBytes() >= 8) {
            out.add(in.readLong());
        }
    }
}

服务器

/**
 * @apiNote Server
 */
public class Server {
    public static void main(String[] args) {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ServerInitializer());
            ChannelFuture future = serverBootstrap.bind(3333).sync();
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

/**
 * @apiNote Server初始化
 */
public class ServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        // 加入出站Handler，数据编码
        pipeline.addLast(new LongToByteEncoder());
        // 加入入站Handler，数据解码
        pipeline.addLast(new ByteToLongDecoder());
        // 加入自定义Handler，处理业务
        pipeline.addLast(new ServerHandler());
    }
}

/**
 * @apiNote Server业务处理Handler
 */
public class ServerHandler extends SimpleChannelInboundHandler<Long> {
    private final Logger log = Logger.getLogger(ServerHandler.class);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {
        log.info("收到客户端数据：" + msg);
        log.info("发送初始化数据：" + 98765L);
        ctx.writeAndFlush(98765L);
    }

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

客户端

/**
 * @apiNote Client
 */
public class Client {
    public static void main(String[] args) {
        NioEventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ClientInitializer());
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 3333).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            group.shutdownGracefully();
        }
    }
}

/**
 * @apiNote Client初始化
 */
public class ClientInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        // 加入出站Handler，数据编码
        pipeline.addLast(new LongToByteEncoder());
        // 加入入站Handler，数据编码
        pipeline.addLast(new ByteToLongDecoder());
        // 加入自定义Handler，处理业务
        pipeline.addLast(new ClientHandler());
    }
}

/**
 * @apiNote Client业务处理Handler
 */
public class ClientHandler extends SimpleChannelInboundHandler<Long> {
    private final Logger log = Logger.getLogger(ClientHandler.class);

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Long msg) throws Exception {

    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("成功连接到服务器");
        log.info("发送初始化数据：" + 12345L);
        ctx.writeAndFlush(12345L);
    }
}

8. TCP

8.1 问题介绍

TCP拆包和粘包

TCP是面向连接、面向字节流的，提供高可靠性服务。服务器和客户端都要有成对的socket，因此，发送端为了将多个发给接收端的包，更有效的发给对方，使用了优化方法（Magle算法），将多次间隔较小且数据量小的数据，合并成一个大的数据块，然后进行封包。这样做虽然提高了效率，但是接收端就难于分辨出完整的数据包，因为面向流的通信是无消息保证边界的。
由于TCP无消息保护边界，需要在接收端处理消息边界问题，也就是我们所说的拆包和粘包问题。

图示

8.2 出现场景

使用代码模拟：

// 服务器Handler
class ServerHandler extends SimpleChannelInboundHandler<ByteBuf> {
    private int count;
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        byte[] buffer = new byte[msg.readableBytes()];
        msg.readBytes(buffer);
        String message = new String(buffer);
        System.out.println("服务器接收到消息：" + message);
        System.out.println("服务器接收到消息数：" + ++count);
        // 返回UUID
        ctx.writeAndFlush(Unpooled.copiedBuffer(UUID.randomUUID().toString(), StandardCharsets.UTF_8));
    }
}

// 客户端Handler
class ClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    private int count;
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        byte[] buffer = new byte[msg.readableBytes()];
        msg.readBytes(buffer);
        String message = new String(buffer);
        System.out.println("客户端接收到消息：" + message);
        System.out.println("客户端接收到消息数：" + ++count);
    }
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        for (int i = 0; i < 5; i++) {
            ctx.writeAndFlush(Unpooled.copiedBuffer("Hello, Khighness-" + i, StandardCharsets.UTF_8));
        }
    }
}

在上述代码中，客户端连接服务器的时候，会向服务器发送5次Hello, Khighness-i，服务器每收到一个消息会返回一个UUID。

运行服务器，开启三个客户端，结果如下：

# Server
服务器接收到消息：Hello, Khighness-0
服务器接收到消息数：1
服务器接收到消息：Hello, Khighness-0
服务器接收到消息数：1
服务器接收到消息：Hello, Khighness-1Hello, Khighness-2
服务器接收到消息数：2
服务器接收到消息：Hello, Khighness-3Hello, Khighness-4
服务器接收到消息数：3
服务器接收到消息：Hello, Khighness-0
服务器接收到消息数：1
服务器接收到消息：Hello, Khighness-1
服务器接收到消息数：2
服务器接收到消息：Hello, Khighness-2Hello, Khighness-3Hello, Khighness-4
服务器接收到消息数：3
# Client-1
客户端接收到消息：f507ecd7-fa87-49d6-a702-7c25a999c729
客户端接收到消息数：1
# Client-2
客户端接收到消息：6cc98a92-8f7e-426e-8d97-59e46d06489b93775b56-f04b-4cde-99b2-2c46f1db70afb3d2d5a0-e977-4b43-9796-433d8a230ec2
客户端接收到消息数：1
# Client-3
客户端接收到消息：18da0b29-8b10-4bc1-a390-e72b3ecaa597443044ae-8634-4acb-b8fd-0c7409ce8baab50889ed-2988-484d-a55f-e91e15752e80
客户端接收到消息数：1

8.3 解决方案

发送端每发送一次消息，就需要在消息的内容之前携带消息的长度，这样，接收端每次先接收消息的长度，再根据长度去读取该消息剩余的内容。如果socket中还有没有读取的内容，也只能放在下一次读取事件中进行。

1）使用自定义协议+编解码器来解决

2）关键就是要解决服务器端每次读取数据长度的问题

自定义协议包

/**
 * @author KHighness
 * @since 2021-06-06
 * @apiNote 协议包
 */
public class MessageProtocol {
    private int len;
    private byte[] content;

    public int getLen() {
        return len;
    }

    public void setLen(int len) {
        this.len = len;
    }

    public byte[] getContent() {
        return content;
    }

    public void setContent(byte[] content) {
        this.content = content;
    }

    public MessageProtocol(int len, byte[] content) {
        this.len = len;
        this.content = content;
    }
}

自定义编码器

/**
 * @author KHighness
 * @since 2021-06-06
 * @apiNote 自定义编码器
 */
public class MessageEncoder extends MessageToByteEncoder<MessageProtocol> {
    @Override
    protected void encode(ChannelHandlerContext ctx, MessageProtocol msg, ByteBuf out) throws Exception {
        out.writeInt(msg.getLen());
        out.writeBytes(msg.getContent());
    }
}

自定义解码器

/**
 * @author KHighness
 * @since 2021-06-06
 * @apiNote 自定义解码器
 */
public class MessageDecoder extends ReplayingDecoder<Void> {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        // 根据长度读取
        int len = in.readInt();
        byte[] content = new byte[len];
        in.readBytes(content);
        // 封装MessageProtocol
        MessageProtocol messageProtocol = new MessageProtocol(len, content);
        out.add(messageProtocol);
    }
}

服务器

/**
 * @author KHighness
 * @since 2021-06-06
 * @apiNote Server
 */
public class Server {
    static class ServerHandler extends SimpleChannelInboundHandler<MessageProtocol> {
        private int count;
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {
            String message = new String(msg.getContent(), StandardCharsets.UTF_8);
            System.out.println("服务器接收到消息：" + message);
            System.out.println("服务器接收到消息数：" + ++count);
            // 返回UUID
            byte[] content = UUID.randomUUID().toString().getBytes(StandardCharsets.UTF_8);
            MessageProtocol messageProtocol = new MessageProtocol(content.length, content);
            ctx.writeAndFlush(messageProtocol);
        }
    }

    public static void main(String[] args) {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap()
                    .channel(NioServerSocketChannel.class).group(bossGroup, workerGroup)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline()
                                    .addLast(new MessageEncoder()).addLast(new MessageDecoder()).addLast(new ServerHandler());
                        }
                    });
            serverBootstrap.bind("127.0.0.1",3333).sync().channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

客户端

/**
 * @author KHighness
 * @since 2021-06-06
 * @apiNote Client
 */
public class Client {
    static class ClientHandler extends SimpleChannelInboundHandler<MessageProtocol> {
        private int count;
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {
            String message = new String(msg.getContent(), StandardCharsets.UTF_8);
            System.out.println("客户端接收到消息：" + message);
            System.out.println("客户端接收到消息数：" + ++count);
        }
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            for (int i = 0; i < 5; i++) {
                String msg = "你好，炒菜K殿下";
                byte[] buffer = msg.getBytes(StandardCharsets.UTF_8);
                MessageProtocol messageProtocol = new MessageProtocol(buffer.length, buffer);
                ctx.writeAndFlush(messageProtocol);
            }
        }
    }

    public static void main(String[] args) {
        NioEventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap()
                    .group(group).channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline()
                                    .addLast(new MessageEncoder()).addLast(new MessageDecoder()).addLast(new ClientHandler());
                        }
                    });
            bootstrap.connect("127.0.0.1", 3333).sync().channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            group.shutdownGracefully();
        }
    }
}

测试结果

# Server
服务器接收到消息：你好，炒菜K殿下
服务器接收到消息数：1
服务器接收到消息：你好，炒菜K殿下
服务器接收到消息数：2
服务器接收到消息：你好，炒菜K殿下
服务器接收到消息数：3
服务器接收到消息：你好，炒菜K殿下
服务器接收到消息数：4
服务器接收到消息：你好，炒菜K殿下
服务器接收到消息数：5
# Client
客户端接收到消息：3e4a9945-4a6b-4867-b144-80d0d8b8e73a
客户端接收到消息数：1
客户端接收到消息：39d2cb2d-99df-417f-9d0e-9d52ced8fcf9
客户端接收到消息数：2
客户端接收到消息：2aa72ca6-e47c-4afa-80b5-5534861f4a5f
客户端接收到消息数：3
客户端接收到消息：3eecf174-a724-468b-814d-b252f2ada736
客户端接收到消息数：4
客户端接收到消息：578d7a67-b46a-48ab-8793-46e7064169dd
客户端接收到消息数：5