netty学习笔记

超、凢脫俗 2022-12-16 06:52 243阅读 0赞

文章目录

  • 一、I/O模型
    • 1、Java BIO编程
      • 1.1 java BIO工作机制
      • 1.2 java BIO应用实例
      • 1.3 java 问题分析
    • 2、java NIO编程
      • 2.1 NIO和BIO的比较
      • 2.2 NIO三大核心原理示意图
      • 2.3 应用实例-本地文件写数据
      • 2.4 应用实例-本地文件读数据
      • 2.5 应用实例-使用一个Buffer完成文件读取、写入
      • 2.6 Selector(选择器)
      • 2.7 SelectionKey
      • 2.8 ServerSocketChannel
      • 2.9 SocketChannel
      • 2.10 NIO 网络编程应用实例-群聊系统
      • 2.11 NIO与零拷贝
    • 3、JavaAIO 基本介绍
    • 4、BIO、NIO与AIO对比表
  • 二、Netty
    • 1、Netty概述
      • 1.1 原生NIO存在的问题
      • 1.2 Netty的优点
      • 1.3 Netty版本说明
    • 2、Netty高性能架构设计
      • 2.1 线程模型基本介绍
      • 2.2 传统阻塞I/O服务模型
      • 2.3 Reactor模式
      • 2.4 Reactor模式中核心组成：
      • 2.5 Reactor模式分类
      • 2.6 单Reactor 单线程
      • 2.7 单Reactor多线程
      • 2.8 主从Reactor多线程
      • 2.9 Reactor模式小结
      • 2.10 Netty模型
      • 2.11 Netty快速入门实例-TCP服务
      • 2.12 异步模型
  • 三、Netty核心模块组件
    • 1、Bootstrap、ServerBootstrap
    • 2、Future、ChannelFuture
    • 3、Channel
    • 4、Selector
    • 5、ChannelHandler及其实现类
    • 6、Pipeline和ChannelPipeline
    • 7、ChannelHandlerContext
    • 8、ChannelOption
    • 9、EventLoopGroup和其实现类NioEventLoopGroup
    • 10、Unpooled类
    • 11、Netty应用实例-群聊系统
  • 四、TCP粘包，拆包及解决方法
    • 1、粘包、拆包表现形式
    • 2、粘包、拆包发生原因
    • 3、粘包、拆包解决办法
  • 五、用Netty自己实现dubboRPC
    • 1、RPC基本介绍
    • 2、RPC调用流程图
    • 3、RPC 调用流程说明
    • 4、自己实现dubboRPC(基于Netty)

一、I/O模型

Java共支持3种网络编程模型/IO模式：BIO、NIO、AIO
Java BIO：同步并阻塞(传统阻塞型)，服务器实现模式为一个连接一个线程，即客户端有连接请求时服务器端就需要启动一个线程进行处理，如果这个连接不做任何事情会造成不必要的线程开销，可以通过线程池机制改善。
Java NIO：同步非阻塞，服务器实现模式为一个线程处理多个请求(连接)，即客户端发送的连接请求都会注册到多路复用器上，多路复用器轮询到连接有I/O请求就进行处理。
Java AIO(NIO.2)：异步非阻塞，AIO引入异步通道的概念，采用了Proactor模式，简化了程序编写，有效的请求才启动线程，它的特点是先由操作系统完成后才通知服务端程序启动线程去处理，一般适用于连接数较多且连接时间较长的应用。

1、Java BIO编程

1.1 java BIO工作机制

在这里插入图片描述
对BIO编程流程的梳理
1）服务器端启动一个ServerSocket
2）客户端启动Socket对服务器进行通信，默认情况下服务器端需要对每个客户建立一个线程与之通讯
3）客户端发出请求后,先咨询服务器是否有线程响应，如果没有则会等待，或者被拒绝
4）如果有响应，客户端线程会等待请求结束后，在继续执行

1.2 java BIO应用实例

实例说明：
1）使用BIO模型编写一个服务器端，监听6666端口，当有客户端连接时，就启动一个线程与之通讯。
2）要求使用线程池机制改善，可以连接多个客户端。
3）服务器端可以接收客户端发送的数据(telnet方式即可)。
4）代码演示

  1. package com.wolfx.bio;
  3. import java.io.IOException;
  4. import java.io.InputStream;
  5. import java.net.ServerSocket;
  6. import java.net.Socket;
  7. import java.util.concurrent.ExecutorService;
  8. import java.util.concurrent.Executors;
  10. /**
  11.  * @description:
  12.  * @author: sukang
  13.  * @date: 2020-10-19 15:19
  14.  */
  15. public class BIOServer {
  16.     public static void main(String[] args) throws IOException {
  17.         //线程池机制
  18.         //思路
  19.         //1、创建一个线程池
  20.         //2、如果有客户端连接，就创建一个线程，与之通讯(单独写一个方法)
  21.         ExecutorService executorService = Executors.newCachedThreadPool();
  23.         //创建ServerSocket
  24.         ServerSocket serverSocket = new ServerSocket(6666);
  25.         System.out.println("服务器启动了");
  27.         while(true){
  28.             System.out.println("线程信息 id =" + Thread.currentThread().getId() + "  名字 =" + Thread.currentThread().getName());
  29.             //监听，等待客户端连接
  30.             System.out.println("等待连接...");
  31.             final Socket accept = serverSocket.accept();
  32.             System.out.println("连接到一个客户端");
  34.             //就创建一个线程，与之通讯(单独写一个方法)
  35.             executorService.execute(new Runnable() {
  36.                 public void run() {
  37.                     //可以和客户端通讯
  38.                     handler(accept);
  39.                 }
  40.             });
  41.         }
  42.     }
  44.     //编写一个handler方法，和客户端通讯
  45.     public static void handler(Socket socket){
  46.         try {
  47.             System.out.println("线程信息 id =" + Thread.currentThread().getId() + " 名字 =" + Thread.currentThread().getName());
  48.             byte[] bytes = new byte[1024];
  49.             //通过socket获取输入流
  50.             InputStream inputStream = socket.getInputStream();
  52.             //循环的读取客户端发送的数据
  53.             while(true){
  54.                 System.out.println("线程信息id="+Thread.currentThread().getId()+"名字="+Thread.currentThread().getName());
  55.                 System.out.println("read....");
  56.                 int read = inputStream.read(bytes);
  57.                 if(read != -1){
  58.                     System.out.println(new String(bytes,0,read));//输出客户端发送的数据
  59.                 }else {
  60.                     break;
  61.                 }
  62.             }
  63.         } catch ( IOException e ) {
  64.             e.printStackTrace();
  65.         } finally {
  66.             System.out.println("关闭和client的连接");
  67.             try {
  68.                 socket.close();
  69.             } catch ( IOException e ) {
  70.                 e.printStackTrace();
  71.             }
  72.         }
  73.     }
  74. }

启动BIOServer
在这里插入图片描述
telnet方式测试
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

1.3 java 问题分析

1）每个请求都需要创建独立的线程，与对应的客户端进行数据Read，业务处理，数据Write。
2）当并发数较大时，需要创建大量线程来处理连接，系统资源占用较大。
3）连接建立后，如果当前线程暂时没有数据可读，则线程就阻塞在Read操作上，造成线程资源浪费

2、java NIO编程

2.1 NIO和BIO的比较

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

2.2 NIO三大核心原理示意图

Selector、Channel和Buffer的关系图(简单版)
在这里插入图片描述
关系图的说明：
1）每个channel都会对应一个Buffer
2）Selector对应一个线程，Selector对应多个channel(连接)
3）该图反应了有三个channel注册到该selector
4）程序切换到哪个channel是有事件决定的,Event就是一个重要的概念
5）Selector会根据不同的事件，在各个通道上切换
6）Buffer就是一个内存块，底层是有一个数组
7）数据的读取写入是通过Buffer,这个和BIO,BIO中要么是输入流，或者是输出流,不能双向，但是NIO的Buffer是可以读也可以写,需要flip方法切换channel是双向的,可以返回底层操作系统的情况,比如Linux，底层的操作系统通道就是双向的

2.3 应用实例-本地文件写数据

1）使用前面学习后的ByteBuffer(缓冲)和FileChannel(通道)，将”hello,尚硅谷”写入到file01.txt中
2）文件不存在就创建
3）代码演示

  1. package com.wolfx.nio;
  3. import java.io.FileOutputStream;
  4. import java.io.IOException;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.FileChannel;
  8. /**
  9.  * @description:
  10.  *          本地文件写数据
  11.  * @author: sukang
  12.  * @date: 2020-10-19 16:55
  13.  */
  14. public class NIOFileChannel01 {
  16.     public static void main(String[] args) throws IOException {
  17.         String str = "你好！世界";
  18.         //创建一个文件输入流
  19.         FileOutputStream fileOutputStream = new FileOutputStream("d:\\file01.txt");
  20.         //通过文件输出流获取文件FileChannel
  21.         FileChannel channel = fileOutputStream.getChannel();
  23.         //创建一个字节缓冲区
  24.         ByteBuffer allocate = ByteBuffer.allocate(1024);
  25.         //将字符放在缓冲区内
  26.         allocate.put(str.getBytes());
  27.         //执行缓冲区的flip()方法
  28.         allocate.flip();
  30.         //将数据从缓冲区写到通道中
  31.         channel.write(allocate);
  32.         //关闭文件流
  33.         fileOutputStream.close();
  34.     }
  35. }

测试
在这里插入图片描述

2.4 应用实例-本地文件读数据

1）使用前面学习后的ByteBuffer(缓冲)和FileChannel(通道)，将file01.txt中的数据读入到程序，并显示在控制台屏幕
2）假定文件已经存在
3）代码演示

  1. package com.wolfx.nio;
  3. import java.io.File;
  4. import java.io.FileInputStream;
  5. import java.io.IOException;
  6. import java.nio.ByteBuffer;
  7. import java.nio.channels.FileChannel;
  9. /**
  10.  * @description:
  11.  *          本地文件读数据
  12.  * @author: sukang
  13.  * @date: 2020-10-19 16:55
  14.  */
  15. public class NIOFileChannel02 {
  17.     public static void main(String[] args) throws IOException {
  18.         //创建一个文件
  19.         File file = new File("d:\\file01.txt");
  20.         //创建一个文件输入流
  21.         FileInputStream fileInputStream = new FileInputStream(file);
  23.         //通过文件输入流获取文件FileChannel
  24.         FileChannel channel = fileInputStream.getChannel();
  26.         //创建一个字节缓冲区
  27.         ByteBuffer allocate = ByteBuffer.allocate((int) file.length());
  29.         //从通道中读数据到缓冲区
  30.         channel.read(allocate);
  32.         System.out.println(new String(allocate.array()));
  34.         //关闭文件流
  35.         fileInputStream.close();
  36.     }
  37. }

测试
在这里插入图片描述

2.5 应用实例-使用一个Buffer完成文件读取、写入

1）使用FileChannel(通道)和方法read,write，完成文件的拷贝
2）拷贝一个文本文件1.txt，放在项目下即可
3）代码演示
在这里插入图片描述

  1. package com.wolfx.nio;
  3. import java.io.File;
  4. import java.io.FileInputStream;
  5. import java.io.FileOutputStream;
  6. import java.io.IOException;
  7. import java.nio.ByteBuffer;
  8. import java.nio.channels.FileChannel;
  10. /**
  11.  * @description:
  12.  *          使用一个Buffer完成文件读取、写入
  13.  * @author: sukang
  14.  * @date: 2020-10-19 16:55
  15.  */
  16. public class NIOFileChannel03 {
  18.     public static void main(String[] args) throws IOException {
  19.         //创建一个文件
  20.         File file = new File("1.txt");
  21.         //创建一个文件输入流
  22.         FileInputStream fileInputStream = new FileInputStream(file);
  23.         //创建一个文件输出流
  24.         FileOutputStream fileOutputStream = new FileOutputStream("2.txt");
  26.         //通过文件输入流获取文件fileInputChannel
  27.         FileChannel fileInputChannel = fileInputStream.getChannel();
  28.         //通过文件输出流获取文件fileOutputChannel
  29.         FileChannel fileOutputChannel = fileOutputStream.getChannel();
  31.         //创建一个字节缓冲区
  32.         ByteBuffer allocate = ByteBuffer.allocate((int) file.length());
  34.         //从通道中读数据到缓冲区
  35.         fileInputChannel.read(allocate);
  36.         allocate.flip();
  38.         //从缓冲区把数据写到文件通道中去
  39.         fileOutputChannel.write(allocate);
  41.         //关闭输入输出文件流
  42.         fileInputStream.close();
  43.         fileOutputStream.close();
  44.     }
  45. }

2.6 Selector(选择器)

在这里插入图片描述
说明如下：
1）Java的NIO，用非阻塞的IO方式。可以用一个线程，处理多个的客户端连接，就会使用到Selector(选择器)
2）Selector能够检测多个注册的通道上是否有事件发生(注意:多个Channel以事件的方式可以注册到同一个Selector)，如果有事件发生，便获取事件然后针对每个事件进行相应的处理。这样就可以只用一个单线程去管理多个通道，也就是管理多个连接和请求。
3）当线程从某客户端Socket通道进行读写数据时，若没有数据可用时，该线程可以进行其他任务。
4）线程通常将非阻塞IO的空闲时间用于在其他通道上执行IO操作，所以单独的线程可以管理多个输入和输出通道。
5）由于读写操作都是非阻塞的，这就可以充分提升IO线程的运行效率，避免由于频繁I/O阻塞导致的线程挂起。
6）一个I/O线程可以并发处理N个客户端连接和读写操作，这从根本上解决了传统同步阻塞I/O一连接一线程模型，架构的性能、弹性伸缩能力和可靠性都得到了极大的提升。

NIO非阻塞网络编程原理分析图
NIO非阻塞网络编程相关的(Selector、SelectionKey、ServerScoketChannel和SocketChannel)关系梳理图
在这里插入图片描述
对上图的说明:
1）当客户端连接时，会通过ServerSocketChannel得到SocketChannel
2）Selector进行监听select方法,返回有事件发生的通道的个数
3）将socketChannel注册到Selector上,register(Selectorsel,intops),一个selector上可以注册多个SocketChannel
4）注册后返回一个SelectionKey,会和该Selector关联(集合)
5）进一步得到各个SelectionKey(有事件发生)
6）在通过SelectionKey反向获取SocketChannel,方法channel(
7）可以通过得到的channel,完成业务处理

2.7 SelectionKey

1）SelectionKey，表示Selector和网络通道的注册关系,共四种:
int OP_ACCEPT：有新的网络连接可以accept，值为16
int OP_CONNECT：代表连接已经建立，值为8
int OP_READ：代表读操作，值为1
int OP_WRITE：代表写操作，值为4

2）SelectionKey相关方法
在这里插入图片描述

2.8 ServerSocketChannel

1）ServerSocketChannel在服务器端监听新的客户端Socket连接
2）相关方法如下
在这里插入图片描述

2.9 SocketChannel

1）SocketChannel，网络IO通道，具体负责进行读写操作。NIO把缓冲区的数据写入通道，或者把通道里的数据读到缓冲区。
2）相关方法如下
在这里插入图片描述

2.10 NIO 网络编程应用实例-群聊系统

实例要求:
1）编写一个NIO群聊系统，实现服务器端和客户端之间的数据简单通讯（非阻塞）
2）实现多人群聊
3）服务器端：可以监测用户上线，离线，并实现消息转发功能
4）客户端：通过channel可以无阻塞发送消息给其它所有用户，同时可以接受其它用户发送的消息(有服务器转发得到)
5）目的：进一步理解NIO非阻塞网络编程机制
6）示意图分析和代码
在这里插入图片描述
服务端

  1. package com.wolfx.nio;
  3. import java.io.IOException;
  4. import java.net.InetSocketAddress;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.*;
  7. import java.util.Iterator;
  9. /**
  10.  * @description:
  11.  *          服务端
  12.  * @author: sukang
  13.  * @date: 2020-10-20 10:09
  14.  */
  15. public class GroupChatServer {
  16.     //定义属性
  17.     private Selector selector;
  18.     private ServerSocketChannel listenChannel;
  19.     private static final int PORT = 6667;
  21.     //构造器
  22.     //初始化工作
  23.     public GroupChatServer(){
  24.         try {
  25.             //得到选择器
  26.             selector = Selector.open();
  27.             //ServerSocketChannel
  28.             listenChannel = ServerSocketChannel.open();
  29.             //绑定端口
  30.             listenChannel.socket().bind(new InetSocketAddress(PORT));
  31.             //设置非阻塞模式
  32.             listenChannel.configureBlocking(false);
  33.             //将该listenChannel注册到selector
  34.             listenChannel.register(selector, SelectionKey.OP_ACCEPT);
  36.             System.out.println("基于NIO的聊天室在["+PORT+"]端口启动成功");
  38.         } catch ( IOException e ) {
  39.             e.printStackTrace();
  40.         }
  41.     }
  43.     //监听
  44.     public void listen(){
  45.         try {
  46.             //循环处理
  47.             while (true){
  48.                 int count = selector.select();
  49.                 if(count > 0){//有事件处理
  50.                     //遍历得到selectionKey集合
  51.                     Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
  52.                     while (iterator.hasNext()){
  53.                        //取出selectionkey
  54.                         SelectionKey key = iterator.next();
  56.                         //监听到accept
  57.                         if(key.isAcceptable()){
  58.                             SocketChannel sc = listenChannel.accept();
  59.                             sc.configureBlocking(false);
  61.                             //将该sc注册到selector
  62.                             sc.register(selector, SelectionKey.OP_READ);
  64.                             //提示
  65.                             System.out.println(sc.getRemoteAddress() + "上线");
  66.                         }
  68.                         if(key.isReadable()){//通道发送read事件，即通道是可读的状态
  69.                             //处理读
  70.                             readData(key);
  71.                         }
  73.                         //当前的key删除，防止重复处理
  74.                         iterator.remove();
  75.                     }
  77.                 }else{
  78.                     System.out.println("等待...");
  79.                 }
  80.             }
  81.         } catch ( Exception e ) {
  82.             e.printStackTrace();
  83.         } finally {
  84.         }
  85.     }
  87.     //读取客户端消息
  88.     private void readData(SelectionKey key){
  89.         //取到关联的channel
  90.         SocketChannel channel = null;
  92.         try {
  93.             //得到channel
  94.             channel = (SocketChannel) key.channel();
  95.             //创建buffer
  96.             ByteBuffer buffer = ByteBuffer.allocate(1024);
  97.             int count = channel.read(buffer);
  99.             //根据count的值做处理
  100.             if(count > 0){
  101.                 //把缓存区数据转成字符串
  102.                 String msg = new String(buffer.array());
  103.                 //输出消息
  104.                 System.out.println("from 客户端:" + msg);
  106.                 //向其他的客户端转发消息(去掉自己)，专门写一个方法来处理
  107.                 sendInfoToOtherClients(msg, channel);
  108.             }
  110.         } catch ( IOException e ) {
  111.             try {
  112.                 System.out.println(channel.getRemoteAddress() + " 离线了...");
  113.                 //取消注册
  114.                 key.channel();
  115.                 //关闭通道
  116.                 channel.close();
  117.             } catch ( IOException ex ) {
  118.                 ex.printStackTrace();
  119.             }
  120.         }
  122.     }
  124.     //转发消息给其他客户(通道)
  125.     private void sendInfoToOtherClients(String msg, SocketChannel self) throws IOException {
  126.         System.out.println("服务器转发消息中...");
  127.         //遍历所有注册到selector上的SocketChannel,并排除self
  129.         for (SelectionKey key:selector.keys()) {
  130.             //通过key取出对应的SocketChannel
  131.             Channel targetChannel = key.channel();
  133.             //排除自己
  134.             if(targetChannel instanceof SocketChannel && targetChannel != self){
  135.                 //转型
  136.                 SocketChannel dest = (SocketChannel) targetChannel;
  137.                 //将msg存储到buffer
  138.                 ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes());
  139.                 //将buffer的数据写入通道
  140.                 dest.write(buffer);
  141.             }
  142.         }
  143.     }
  145.     public static void main(String[] args) {
  146.         //创建服务器对象
  147.         GroupChatServer groupChatServer = new GroupChatServer();
  148.         groupChatServer.listen();
  149.     }
  150. }

客户端

  1. package com.wolfx.nio;
  3. import java.io.IOException;
  4. import java.net.InetSocketAddress;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.SelectionKey;
  7. import java.nio.channels.Selector;
  8. import java.nio.channels.SocketChannel;
  9. import java.util.Iterator;
  10. import java.util.Scanner;
  12. /**
  13.  * @description:
  14.  *          客户端
  15.  * @author: sukang
  16.  * @date: 2020-10-20 11:10
  17.  */
  18. public class GroupChatClient {
  20.     //定义相关的属性
  21.     private final String HOST = "127.0.0.1";//服务器的ip
  22.     private final int PORT = 6667;//服务器端口
  23.     private Selector selector;
  24.     private SocketChannel socketChannel;
  25.     private String username;
  27.     //构造器,完成初始化工作
  28.     public GroupChatClient() throws IOException {
  29.         selector = Selector.open();
  30.         //连接服务器
  31.         socketChannel = SocketChannel.open(new InetSocketAddress(HOST,PORT));
  32.         //设置非阻塞
  33.         socketChannel.configureBlocking(false);
  34.         //将channel注册到selector
  35.         socketChannel.register(selector, SelectionKey.OP_READ);
  36.         //得到username
  37.         username = socketChannel.getRemoteAddress().toString();
  38.         System.out.println(username + " is ok...");
  39.     }
  41.     //向服务器发送消息
  42.     public void sendInfo(String info){
  43.         info = username + " 说: " + info;
  45.         try {
  46.             socketChannel.write(ByteBuffer.wrap(info.getBytes()));
  47.         } catch ( IOException e ) {
  48.             e.printStackTrace();
  49.         }
  50.     }
  52.     //读取从服务端回复的消息
  53.     public void readInfo(){
  54.         try {
  55.             int readChannels = selector.select();
  56.             if(readChannels > 0){//有可以用的通道
  57.                 Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
  58.                 while (iterator.hasNext()){
  59.                     SelectionKey key = iterator.next();
  60.                     if(key.isReadable()){
  61.                         //得到相关的通道
  62.                         SocketChannel sc = (SocketChannel) key.channel();
  63.                         //得到一个Buffer
  64.                         ByteBuffer buffer = ByteBuffer.allocate(1024);
  65.                         //读取
  66.                         sc.read(buffer);
  67.                         //把读到的缓冲区的数据转成字符串
  68.                         String msg = new String(buffer.array());
  69.                         System.out.println(msg.trim());
  70.                     }
  71.                 }
  72.                 iterator.remove();
  73.             }else{
  75.             }
  76.         } catch ( IOException e ) {
  77.             e.printStackTrace();
  78.         }
  79.     }
  81.     public static void main(String[] args) throws IOException {
  82.         //启动我们客户端
  83.         final GroupChatClient chatClient = new GroupChatClient();
  85.         //启动一个线程，每隔3秒，从服务器读取数据
  86.         new Thread(){
  87.             @Override
  88.             public void run(){
  89.                 while (true){
  90.                     chatClient.readInfo();
  91.                     try {
  92.                         sleep(3000);
  93.                     } catch ( InterruptedException e ) {
  94.                         e.printStackTrace();
  95.                     }
  96.                 }
  97.             }
  98.         }.start();
  100.         //发送数据给服务器端
  101.         Scanner scanner = new Scanner(System.in);
  103.         while (scanner.hasNextLine()){
  104.             String s = scanner.nextLine();
  105.             chatClient.sendInfo(s);
  106.         }
  107.     }
  108. }

测试结果
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

2.11 NIO与零拷贝

传统IO
传统IO的数据拷贝流程如下图：
在这里插入图片描述
1）数据需要从磁盘拷贝到内核空间，再从内核空间拷到用户空间（JVM）。
2）程序可能进行数据修改等操作。
3）再将数据拷贝到内核空间，内核空间再拷贝到网卡内存，通过网络发送出去（或拷贝到磁盘）。
即数据的读写（这里用户空间发到网络也算作写），都至少需要两次拷贝。
当然磁盘到内核空间属于DMA拷贝（DMA即直接内存存取，原理是外部设备不通过CPU而直接与系统内存交换数据）。而内核空间到用户空间则需要CPU的参与进行拷贝，既然需要CPU参与，也就涉及到了内核态和用户态的相互切换，如下图：
在这里插入图片描述
NIO的零拷贝
零拷贝的数据拷贝如下图：
在这里插入图片描述
内核态与用户态切换如下图：
在这里插入图片描述
改进的地方：

  • 我们已经将上下文切换次数从4次减少到了2次；
  • 将数据拷贝次数从4次减少到了3次（其中只有1次涉及了CPU，另外2次是DMA直接存取）。

但这还没有达到我们零拷贝的目标。如果底层NIC（网络接口卡）支持gather操作，我们能进一步减少内核中的数据拷贝。在Linux 2.4以及更高版本的内核中，socket缓冲区描述符已被修改用来适应这个需求。这种方式不但减少多次的上下文切换，同时消除了需要CPU参与的重复的数据拷贝。用户这边的使用方式不变，而内部已经有了质的改变：
在这里插入图片描述
NIO的零拷贝由transferTo()方法实现。transferTo()方法将数据从FileChannel对象传送到可写的字节通道（如Socket Channel等）。在内部实现中，由native方法transferTo0()来实现，它依赖底层操作系统的支持。在UNIX和Linux系统中，调用这个方法将会引起sendfile()系统调用。
使用场景一般是：

  1. 文件较大，读写较慢，追求速度<script src="https://localhost01.cn/js/jquery-2.0.0.min.js"></script>
  2. JVM内存不足，不能加载太大数据
  3. 内存带宽不够，即存在其他程序或线程存在大量的IO操作，导致带宽本来就小

以上都建立在不需要进行数据文件操作的情况下，如果既需要这样的速度，也需要进行数据操作怎么办？
那么使用NIO的直接内存！

NIO的直接内存
首先，它的作用位置处于传统IO（BIO）与零拷贝之间，为何这么说？

  • 传统IO，可以把磁盘的文件经过内核空间，读到JVM空间，然后进行各种操作，最后再写到磁盘或是发送到网络，效率较慢但支持数据文件操作。
  • 零拷贝则是直接在内核空间完成文件读取并转到磁盘（或发送到网络）。由于它没有读取文件数据到JVM这一环，因此程序无法操作该文件数据，尽管效率很高！

而直接内存则介于两者之间，效率一般且可操作文件数据。直接内存（mmap技术）将文件直接映射到内核空间的内存，返回一个操作地址（address），它解决了文件数据需要拷贝到JVM才能进行操作的窘境。而是直接在内核空间直接进行操作，省去了内核空间拷贝到用户空间这一步操作。
NIO的直接内存是由MappedByteBuffer实现的。核心即是map()方法，该方法把文件映射到内存中，获得内存地址addr，然后通过这个addr构造MappedByteBuffer类，以暴露各种文件操作API。
由于MappedByteBuffer申请的是堆外内存，因此不受Minor GC控制，只能在发生Full GC时才能被回收。而DirectByteBuffer改善了这一情况，它是MappedByteBuffer类的子类，同时它实现了DirectBuffer接口，维护一个Cleaner对象来完成内存回收。因此它既可以通过Full GC来回收内存，也可以调用clean()方法来进行回收。
另外，直接内存的大小可通过jvm参数来设置：-XX:MaxDirectMemorySize。
NIO的MappedByteBuffer还有一个兄弟叫做HeapByteBuffer。顾名思义，它用来在堆中申请内存，本质是一个数组。由于它位于堆中，因此可受GC管控，易于回收。

mmap和sendFile的区别

mmap：通过内存映射，将文件映射到内核缓冲区，同时，用户空间可以共享内核空间的数据。这样，在进行网络传输时，就可以减少内核空间到用户空间的拷贝次数。
sendFile：Linux2.1版本提供了sendFile函数，其基本原理如下：数据根本不经过用户态，直接从内核缓冲区进入到SocketBuffer，同时，由于和用户态完全无关，就减少了一次上下文切换

3、JavaAIO 基本介绍

1）JDK7引入了AsynchronousI/O，即AIO。在进行I/O编程中，常用到两种模式：Reactor和Proactor。Java的NIO就是Reactor，当有事件触发时，服务器端得到通知，进行相应的处理。
2）AIO即NIO2.0，叫做异步不阻塞的IO。AIO引入异步通道的概念，采用了Proactor模式，简化了程序编写，有效的请求才启动线程，它的特点是先由操作系统完成后才通知服务端程序启动线程去处理，一般适用于连接数较多且连接时间较长的应用。
3）目前AIO还没有广泛应用，Netty也是基于NIO，而不是AIO。

4、BIO、NIO与AIO对比表

在这里插入图片描述

二、Netty

1、Netty概述

1.1 原生NIO存在的问题

1）NIO的类库和API繁杂，使用麻烦：需要熟练掌握Selector、ServerSocketChannel、SocketChannel、ByteBuffer等。
2）需要具备其他的额外技能：要熟悉Java多线程编程，因为NIO编程涉及到Reactor模式，你必须对多线程和网络编程非常熟悉，才能编写出高质量的NIO程序。
3）开发工作量和难度都非常大：例如客户端面临断连重连、网络闪断、半包读写、失败缓存、网络拥塞和异常流的处理等等。
4）JDKNIO的Bug：例如臭名昭著的EpollBug，它会导致Selector空轮询，最终导致CPU100%。直到JDK1.7版本该问题仍旧存在，没有被根本解决。

1.2 Netty的优点

Netty对JDK自带的NIO的API进行了封装，解决了上述问题。
1）设计优雅：适用于各种传输类型的统一API阻塞和非阻塞Socket；基于灵活且可扩展的事件模型，可以清晰地分离关注点；高度可定制的线程模型-单线程，一个或多个线程池
2）使用方便：详细记录的Javadoc，用户指南和示例；没有其他依赖项，JDK5（Netty3.x）或6（Netty4.x）就足够了。
3）高性能、吞吐量更高：延迟更低；减少资源消耗；最小化不必要的内存复制。
4）安全：完整的SSL/TLS和StartTLS支持。
5）社区活跃、不断更新：社区活跃，版本迭代周期短，发现的Bug可以被及时修复，同时，更多的新功能会被加入

1.3 Netty版本说明

1）netty版本分为netty3.x和netty4.x、netty5.x
2）因为Netty5出现重大bug，已经被官网废弃了，目前推荐使用的是Netty4.x的稳定版本

2、Netty高性能架构设计

2.1 线程模型基本介绍

1）目前存在的线程模型有：
传统阻塞I/O服务模型
Reactor模式

2）Netty线程模式(Netty主要基于主从Reactor多线程模型做了一定的改进，其中主从Reactor多线程模型有多个Reactor)

2.2 传统阻塞I/O服务模型

在这里插入图片描述模型特点
1）采用阻塞IO模式获取输入的数据
2）每个连接都需要独立的线程完成数据的输入，业务处理，数据返回

问题分析
1）当并发数很大，就会创建大量的线程，占用很大系统资源
2）连接创建后，如果当前线程暂时没有数据可读，该线程会阻塞在read操作，造成线程资源浪费

2.3 Reactor模式

针对传统阻塞I/O服务模型的2个缺点，解决方案：
1）基于I/O复用模型：多个连接共用一个阻塞对象，应用程序只需要在一个阻塞对象等待，无需阻塞等待所有连接。当某个连接有新的数据可以处理时，操作系统通知应用程序，线程从阻塞状态返回，开始进行业务处理Reactor对应的叫法:1.反应器模式2.分发者模式(Dispatcher)3.通知者模式(notifier)
2）基于线程池复用线程资源：不必再为每个连接创建线程，将连接完成后的业务处理任务分配给线程进行处理，一个线程可以处理多个连接的业务。

I/O复用结合线程池，就是Reactor模式基本设计思想，如图
在这里插入图片描述
1）Reactor模式，通过一个或多个输入同时传递给服务处理器的模式(基于事件驱动)
2）服务器端程序处理传入的多个请求,并将它们同步分派到相应的处理线程，因此Reactor模式也叫Dispatcher模式
3）Reactor模式使用IO复用监听事件,收到事件后，分发给某个线程(进程),这点就是网络服务器高并发处理关键

2.4 Reactor模式中核心组成：

1）Reactor：Reactor在一个单独的线程中运行，负责监听和分发事件，分发给适当的处理程序来对IO事件做出反应。它就像公司的电话接线员，它接听来自客户的电话并将线路转移到适当的联系人；
2）Handlers：处理程序执行I/O事件要完成的实际事件，类似于客户想要与之交谈的公司中的实际官员。Reactor通过调度适当的处理程序来响应I/O事件，处理程序执行非阻塞操作。

2.5 Reactor模式分类

根据Reactor的数量和处理资源池线程的数量不同，有3种典型的实现
1）单Reactor单线程
2）单Reactor多线程
3）主从Reactor多线程

2.6 单Reactor 单线程

原理图
在这里插入图片描述
方案说明
1）Select是前面I/O复用模型介绍的标准网络编程API，可以实现应用程序通过一个阻塞对象监听多路连接请求
2）Reactor对象通过Select监控客户端请求事件，收到事件后通过Dispatch进行分发
3）如果是建立连接请求事件，则由Acceptor通过Accept处理连接请求，然后创建一个Handler对象处理连接完成后的后续业务处理
4）如果不是建立连接事件，则Reactor会分发调用连接对应的Handler来响应
5）Handler会完成Read→业务处理→Send的完整业务流程

结合实例：服务器端用一个线程通过多路复用搞定所有的IO操作（包括连接，读、写等），编码简单，清晰明了，但是如果客户端连接数量较多，将无法支撑，前面的NIO案例就属于这种模型。

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

2.7 单Reactor多线程

原理图
在这里插入图片描述
对上图的小结
1）Reactor对象通过select监控客户端请求事件,收到事件后，通过dispatch进行分发
2）如果建立连接请求,则右Acceptor通过accept处理连接请求,然后创建一个Handler对象处理完成连接后的各种事件
3）如果不是连接请求，则由reactor分发调用连接对应的handler来处理
4）handler只负责响应事件，不做具体的业务处理,通过read读取数据后，会分发给后面的worker线程池的某个线程处理业务
5）worker线程池会分配独立线程完成真正的业务，并将结果返回给handler
6）handler收到响应后，通过send将结果返回给client

方案优缺点分析
1）优点：可以充分的利用多核cpu的处理能力
2）缺点：多线程数据共享和访问比较复杂，reactor处理所有的事件的监听和响应，在单线程运行，在高并发场景容易出现性能瓶颈

2.8 主从Reactor多线程

原理图
在这里插入图片描述
上图的方案说明
1）Reactor主线程MainReactor对象通过select监听连接事件,收到事件后，通过Acceptor处理连接事件
2）当Acceptor处理连接事件后，MainReactor将连接分配给SubReactor
3）subreactor将连接加入到连接队列进行监听,并创建handler进行各种事件处理
4）当有新事件发生时，subreactor就会调用对应的handler处理
5）handler通过read读取数据，分发给后面的worker线程处理
6）worker线程池分配独立的worker线程进行业务处理，并返回结果
7）handler收到响应的结果后，再通过send将结果返回给client
8）Reactor主线程可以对应多个Reactor子线程,即MainRecator可以关联多个SubReactor

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

2.9 Reactor模式小结

3种模式用生活案例来理解
1）单Reactor单线程，前台接待员和服务员是同一个人，全程为顾客服
2）单Reactor多线程，1个前台接待员，多个服务员，接待员只负责接待
3）主从Reactor多线程，多个前台接待员，多个服务生

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

2.10 Netty模型

工作原理示意图1-简单版
在这里插入图片描述
对上图说明
1）BossGroup线程维护Selector,只关注Accecpt
2）当接收到Accept事件，获取到对应的SocketChannel，封装成NIOScoketChannel并注册到Worker线程(事件循环)，并进行维护
3）当Worker线程监听到selector中通道发生自己感兴趣的事件后，就进行处理(就由handler)，注意handler已经加入到通道

工作原理示意图2-进阶版
在这里插入图片描述
工作原理示意图-详细版
在这里插入图片描述
对上图的说明小结
1）Netty抽象出两组线程池BossGroup专门负责接收客户端的连接,WorkerGroup专门负责网络的读写
2）BossGroup和WorkerGroup类型都是NioEventLoopGroup
3）NioEventLoopGroup相当于一个事件循环组,这个组中含有多个事件循环，每一个事件循环是NioEventLoop
4）NioEventLoop表示一个不断循环的执行处理任务的线程，每个NioEventLoop都有一个selector,用于监听绑定在其上的socket的网络通讯
5）NioEventLoopGroup可以有多个线程,即可以含有多个NioEventLoop
6）每个BossNioEventLoop循环执行的步骤有3步

  • 轮询accept事件
  • 处理accept事件,与client建立连接,生成NioScocketChannel,并将其注册到某个workerNIOEventLoop上的selector
  • 处理任务队列的任务，即runAllTasks

7）每个WorkerNIOEventLoop循环执行的步骤

  • 轮询read,write事件
  • 处理i/o事件，即read,write事件，在对应NioScocketChannel处理
  • 处理任务队列的任务，即runAllTasks

8）每个WorkerNIOEventLoop处理业务时，会使用pipeline(管道),pipeline中包含了channel,即通过pipeline可以获取到对应通道,管道中维护了很多的处理器

2.11 Netty快速入门实例-TCP服务

实例要求：
1）Netty服务器在6668端口监听，客户端能发送消息给服务器”hello,服务器~”
2）服务器可以回复消息给客户端”hello,客户端~”
3）目的：对Netty线程模型有一个初步认识,便于理解Netty模型理论

代码如下
NettyServer

  1. package com.wolfx.netty;
  3. import io.netty.bootstrap.ServerBootstrap;
  4. import io.netty.channel.ChannelFuture;
  5. import io.netty.channel.ChannelInitializer;
  6. import io.netty.channel.ChannelOption;
  7. import io.netty.channel.nio.NioEventLoopGroup;
  8. import io.netty.channel.socket.SocketChannel;
  9. import io.netty.channel.socket.nio.NioServerSocketChannel;
  11. /**
  12.  * @description:
  13.  *          服务端
  14.  * @author: sukang
  15.  * @date: 2020-10-21 14:54
  16.  */
  17. public class NettyServer {
  19.     public static void main(String[] args) {
  20.         //创建BossGroup和WorkerGroup
  21.         //说明
  22.         //1、创建两个线程组bossGroup和workerGroup
  23.         //2、bossGroup只是处理连接请求， 真正的和客户端业务处理，会交给workerGroup完成
  24.         //3、两个都是无限循环
  25.         //4、bossGroup和workerGroup含有的子线程(NioEventLoop)的个数
  26.         //默认实际cpu核数*2
  27.         NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
  28.         NioEventLoopGroup workerGroup = new NioEventLoopGroup();
  30.         try {
  31.             //创建服务器端的启动对象，配置参数
  32.             ServerBootstrap bootstrap = new ServerBootstrap();
  34.             //使用链式编程来进行设置
  35.             bootstrap.group(bossGroup, workerGroup)//设置两个线程组
  36.                         .channel(NioServerSocketChannel.class)//使用NioSocketChannel作为服务器的通道实现
  37.                         .option(ChannelOption.SO_BACKLOG, 128)//设置线程队列得到连接个数
  38.                         .childOption(ChannelOption.SO_KEEPALIVE, true)//设置保持活动连接状态
  39.                         .childHandler(new ChannelInitializer<SocketChannel>() {//创建一个通道测试对象（匿名对象）
  40.                             //给pipeline设置处理器
  41.                             @Override
  42.                             protected void initChannel(SocketChannel socketChannel) throws Exception {
  43.                                 socketChannel.pipeline().addLast(new NettyServerHandler());
  44.                             }
  45.                         });//给我们的workerGroup的EventLoop对应的管道设置处理器
  46.             System.out.println("...服务器 is ready...");
  48.             //绑定一个端口并且同步，生成了一个ChannelFuture对象
  49.             //启动服务器(并绑定端口)
  50.             ChannelFuture cf = bootstrap.bind(6668).sync();
  51.         } catch ( Exception e ) {
  52.             e.printStackTrace();
  53.         } finally {
  54.             bossGroup.shutdownGracefully();
  55.             workerGroup.shutdownGracefully();
  56.         }
  57.     }
  58. }

NettyServerHandler

  1. package com.wolfx.netty;
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.buffer.Unpooled;
  5. import io.netty.channel.Channel;
  6. import io.netty.channel.ChannelHandlerContext;
  7. import io.netty.channel.ChannelInboundHandlerAdapter;
  8. import io.netty.channel.ChannelPipeline;
  9. import io.netty.util.CharsetUtil;
  11. /**
  12.  * @description:
  13.  * 自定义Handler
  14.  * 说明
  15.  * 1、我们自定义一个Handler需要继续netty规定好的某个HandlerAdpter(规范)
  16.  * 2、这时我们自定义一个Handler，才能称为handler
  17.  * @author: sukang
  18.  * @date: 2020-10-21 15:30
  19.  */
  20. public class NettyServerHandler extends ChannelInboundHandlerAdapter {
  22.     //读取数据实际（这里我们可以读取客户端发送的消息）
  23.     //1、ChannelHandlerContext ctx:上下文对象，含有管道pipeline，通道channel，地址
  24.     //2、Object msg: 就是客户端发送的数据 默认Object
  25.     @Override
  26.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  27.         System.out.println("服务器读取线程 " + Thread.currentThread().getName());
  28.         System.out.println("server ctx =" + ctx);
  29.         System.out.println("看看 channel和 pipeline的关系");
  30.         Channel channel = ctx.channel();
  31.         ChannelPipeline pipeline = ctx.pipeline();//本质是一个双向链接，出站入站
  33.         //将msg转换一个ByteBuf
  34.         //ByteBuf 是Netty提供，不是NIO的ByteBuffer
  35.         ByteBuf buf = (ByteBuf) msg;
  36.         System.out.println("客户端发送消息是:"+buf.toString(CharsetUtil.UTF_8));
  37.         System.out.println("客户端地址:"+channel.remoteAddress());
  38.     }
  40.     //数据读取完毕
  41.     @Override
  42.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  43.         //writeAndFlush 是 write + flush
  44.         //将数据写入到缓存，并刷新
  45.         //一般讲， 我们对这个发送的数据进行编码
  46.         ctx.writeAndFlush(Unpooled.copiedBuffer("hello,客户端~(>^ω^<)喵",CharsetUtil.UTF_8));
  47.     }
  49.     //处理异常， 一般是需要关闭通道
  50.     @Override
  51.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  52.         ctx.close();
  53.     }
  54. }

NettyClient

  1. package com.wolfx.netty;
  3. import io.netty.bootstrap.Bootstrap;
  4. import io.netty.channel.ChannelFuture;
  5. import io.netty.channel.ChannelInitializer;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.SocketChannel;
  8. import io.netty.channel.socket.nio.NioSocketChannel;
  10. /**
  11.  * @description:
  12.  *      客户端
  13.  * @author: sukang
  14.  * @date: 2020-10-21 16:06
  15.  */
  16. public class NettyClient {
  17.     public static void main(String[] args) {
  18.         //客户端需要一个事件循环组
  19.         NioEventLoopGroup group = new NioEventLoopGroup();
  21.         try {
  22.             //创建客户端启动对象
  23.             //注意客户端使用的不是ServerBootstrap而是Bootstrap
  24.             Bootstrap bootstrap = new Bootstrap();
  26.             //设置相关参数
  27.             bootstrap.group(group)//设置线程组
  28.                      .channel(NioSocketChannel.class)//设置客户端通道的实现类(反射)
  29.                      .handler(new ChannelInitializer<SocketChannel>() {
  30.                          @Override
  31.                          protected void initChannel(SocketChannel socketChannel) throws Exception {
  32.                              socketChannel.pipeline().addLast(new NettyClientHandler());//加入自己的处理器
  33.                          }
  34.                      });
  35.             System.out.println("客户端 ok...");
  37.             //启动客户端去连接服务器端
  38.             //关于ChannelFuture要分析，涉及到netty的异步模型
  39.             ChannelFuture channelFuture = bootstrap.connect("127.0.0.1",6668).sync();
  40.             //给关闭通道进行监听
  41.             channelFuture.channel().closeFuture().sync();
  42.         } catch ( Exception e ) {
  43.             e.printStackTrace();
  44.         } finally {
  45.             group.shutdownGracefully();
  46.         }
  47.     }
  48. }

NettyClientHandler

  1. package com.wolfx.netty;
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.buffer.Unpooled;
  5. import io.netty.channel.ChannelHandlerContext;
  6. import io.netty.channel.ChannelInboundHandlerAdapter;
  7. import io.netty.util.CharsetUtil;
  9. /**
  10.  * @description:
  11.  * @author: sukang
  12.  * @date: 2020-10-21 16:16
  13.  */
  14. public class NettyClientHandler extends ChannelInboundHandlerAdapter {
  17.     //当通道就绪就会触发该方法
  19.     @Override
  20.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  21.         System.out.println("client"+ctx);
  22.         ctx.writeAndFlush(Unpooled.copiedBuffer("hello,server:(>^ω^<)喵", CharsetUtil.UTF_8));
  23.     }
  25.     //当通道有读取事件时，会触发
  26.     @Override
  27.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  28.         ByteBuf buf = (ByteBuf)msg;
  29.         System.out.println("服务器回复的消息:"+buf.toString(CharsetUtil.UTF_8));
  30.         System.out.println("服务器的地址："+ctx.channel().remoteAddress());
  31.     }
  33.     @Override
  34.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  35.         cause.printStackTrace();
  36.         ctx.close();
  37.     }
  38. }

2.12 异步模型

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

Future说明
1）表示异步的执行结果,可以通过它提供的方法来检测执行是否完成，比如检索计算等等。
2）ChannelFuture是一个接口：publicinterfaceChannelFutureextendsFuture我们可以添加监听器，当监听的事件发生时，就会通知到监听器。

工作原理示意图
在这里插入图片描述
说明:
1）在使用Netty进行编程时，拦截操作和转换出入站数据只需要您提供callback或利用future即可。这使得链式操作简单、高效,并有利于编写可重用的、通用的代码。
2）Netty框架的目标就是让你的业务逻辑从网络基础应用编码中分离出来、解脱出来。

Future-Listener机制
1）当Future对象刚刚创建时，处于非完成状态，调用者可以通过返回的ChannelFuture来获取操作执行的状态，注册监听函数来执行完成后的操作。
2）常见有如下操作

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

举例说明
演示：绑定端口是异步操作，当绑定操作处理完，将会调用相应的监听器处理逻辑

在这里插入图片描述

三、Netty核心模块组件

1、Bootstrap、ServerBootstrap

1）Bootstrap意思是引导，一个Netty应用通常由一个Bootstrap开始，主要作用是配置整个Netty程序，串联各个组件，Netty中Bootstrap类是客户端程序的启动引导类，ServerBootstrap是服务端启动引导类
2）常见的方法有
public ServerBootstrap group(EventLoopGroup parentGroup,EventLoopGroup childGroup)，该方法用于服务器端，用来设置两个EventLoop
public B group(EventLoopGroup group)，该方法用于客户端，用来设置一个EventLoop
public B channel(Class<?extendsC> 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)，该方法用于客户端，用来连接服务器端

2、Future、ChannelFuture

Netty中所有的IO操作都是异步的，不能立刻得知消息是否被正确处理。但是可以过一会等它执行完成或者直接注册一个监听，具体的实现就是通过Future和ChannelFutures，他们可以注册一个监听，当操作执行成功或失败时监听会自动触发注册的监听事件
常见的方法有
Channel channel()，返回当前正在进行IO操作的通道
ChannelFuture sync()，等待异步操作执行完毕

3、Channel

1）Netty网络通信的组件，能够用于执行网络I/O操作。
2）通过Channel可获得当前网络连接的通道的状态
3）通过Channel可获得网络连接的配置参数（例如接收缓冲区大小）
4）Channel提供异步的网络I/O操作(如建立连接，读写，绑定端口)，异步调用意味着任何I/O调用都将立即返回，并且不保证在调用结束时所请求的I/O操作已完成
5）调用立即返回一个ChannelFuture实例，通过注册监听器到ChannelFuture上，可以I/O操作成功、失败或取消时回调通知调用方
6）支持关联I/O操作与对应的处理程序
7）不同协议、不同的阻塞类型的连接都有不同的Channel类型与之对应，常用的Channel类型：
NioSocketChannel，异步的客户端TCPSocket连接。
NioServerSocketChannel，异步的服务器端TCPSocket连接。
NioDatagramChannel，异步的UDP连接。
NioSctpChannel，异步的客户端Sctp连接。
NioSctpServerChannel，异步的Sctp服务器端连接，这些通道涵盖了UDP和TCP网络IO以及文件IO。

4、Selector

1）Netty基于Selector对象实现I/O多路复用，通过Selector一个线程可以监听多个连接的Channel事件。
2）当向一个Selector中注册Channel后，Selector内部的机制就可以自动不断地查询(Select)这些注册的Channel是否有已就绪的I/O事件（例如可读，可写，网络连接完成等），这样程序就可以很简单地使用一个线程高效地管理多个Channel

5、ChannelHandler及其实现类

1）ChannelHandler是一个接口，处理I/O事件或拦截I/O操作，并将其转发到其ChannelPipeline(业务处理链)中的下一个处理程序。
2）ChannelHandler本身并没有提供很多方法，因为这个接口有许多的方法需要实现，方便使用期间，可以继承它的子类
3）ChannelHandler及其实现类一览图
在这里插入图片描述
4）我们经常需要自定义一个Handler类去继承ChannelInboundHandlerAdapter，然后通过重写相应方法实现业务逻辑，我们接下来看看一般都需要重写哪些方法
在这里插入图片描述

6、Pipeline和ChannelPipeline

ChannelPipeline是一个重点：
1）ChannelPipeline是一个Handler的集合，它负责处理和拦截inbound或者outbound的事件和操作，相当于一个贯穿Netty的链。(也可以这样理解：ChannelPipeline是保存ChannelHandler的List，用于处理或拦截Channel的入站事件和出站操作)
2）ChannelPipeline实现了一种高级形式的拦截过滤器模式，使用户可以完全控制事件的处理方式，以及Channel中各个的ChannelHandler如何相互交互
3）在Netty中每个Channel都有且仅有一个ChannelPipeline与之对应，它们的组成关系如下
在这里插入图片描述
4）常用方法
ChannelPipeline addFirst(ChannelHandler… handlers)，把一个业务处理类（handler）添加到链中的第一个位置
ChannelPipeline addLast(ChannelHandler… handlers)，把一个业务处理类（handler）添加到链中的最后一个位置

7、ChannelHandlerContext

1）保存Channel相关的所有上下文信息，同时关联一个ChannelHandler对象
2）即ChannelHandlerContext中包含一个具体的事件处理器ChannelHandler，同时ChannelHandlerContext中也绑定了对应的pipeline和Channel的信息，方便对ChannelHandler进行调用
3）常用方法
在这里插入图片描述

8、ChannelOption

1）Netty在创建Channel实例后,一般都需要设置ChannelOption参数。
2）ChannelOption参数如下:
在这里插入图片描述

9、EventLoopGroup和其实现类NioEventLoopGroup

1）EventLoopGroup是一组EventLoop的抽象，Netty为了更好的利用多核CPU资源，一般会有多个EventLoop同时工作，每个EventLoop维护着一个Selector实例。
2）EventLoopGroup提供next接口，可以从组里面按照一定规则获取其中一个EventLoop来处理任务。在Netty服务器端编程中，我们一般都需要提供两个EventLoopGroup，例如：BossEventLoopGroup和WorkerEventLoopGroup。
3）通常一个服务端口即一个ServerSocketChannel对应一个Selector和一个EventLoop线程。BossEventLoop负责接收客户端的连接并将SocketChannel交给WorkerEventLoopGroup来进行IO处理，如下图所示
在这里插入图片描述
4）常用方法
public NioEventLoopGroup()，构造方法
public Future<?> shutdownGracefully()，断开连接，关闭线程

10、Unpooled类

1）Netty提供一个专门用来操作缓冲区(即Netty的数据容器)的工具类
2）常用方法如下所示
在这里插入图片描述
3）举例说明Unpooled获取Netty的数据容器ByteBuf的基本使用【案例演示】
在这里插入图片描述
案例一

  1. package com.wolfx.netty;
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.buffer.Unpooled;
  6. /**
  7.  * @description:
  8.  * @author: sukang
  9.  * @date: 2020-10-22 18:32
  10.  */
  11. public class NettyByteBuf01 {
  13.     public static void main(String[] args) {
  14.         //创建一个ByteBuf
  15.         //说明
  16.         //1、创建对象，该对象包含一个数组arr，是一个byte[10]
  17.         //2、在netty的buffer中，不需要使用flip进行反转
  18.         //底层维护了readerindex和writerIndex
  19.         //3、通过readerindex和writerIndex和capacity，将buffer分成三个区域
  20.         //0---readerindex已经读取的区域
  21.         //readerindex---writerIndex，可读的区域
  22.         //writerIndex--capacity,可写的区域
  23.         ByteBuf buffer = Unpooled.buffer(10);
  25.         for (int i = 0; i < 10; i++) {
  26.             buffer.writeByte(i);
  27.         }
  29.         System.out.println("capacity =" + buffer.capacity());//10
  30.         //输出
  31.         for (int i = 0; i < buffer.capacity(); i++) {
  32.             System.out.println(buffer.readByte());
  33.         }
  34.         System.out.println("执行完毕");
  35.     }
  36. }

测试结果
在这里插入图片描述
案例二

  1. package com.wolfx.netty;
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.buffer.Unpooled;
  6. import java.nio.charset.Charset;
  8. /**
  9.  * @description:
  10.  * @author: sukang
  11.  * @date: 2020-10-22 18:47
  12.  */
  13. public class NettyByteBuf02 {
  14.     public static void main(String[] args) {
  16.         //创建ByteBuf
  17.         ByteBuf byteBuf = Unpooled.copiedBuffer("hello, world!", Charset.forName("utf-8"));
  19.         //使用相关的方法
  20.         if(byteBuf.hasArray()){//true
  21.             byte[] content = byteBuf.array();
  23.             //将content转成字符串
  24.             System.out.println(new String(content,Charset.forName("utf-8")));
  25.             System.out.println("byteBuf =" + byteBuf);
  26.             System.out.println(byteBuf.arrayOffset());//0
  27.             System.out.println(byteBuf.readerIndex());//0
  28.             System.out.println(byteBuf.writerIndex());//12
  29.             System.out.println(byteBuf.capacity());//36
  31.             System.out.println(byteBuf.getByte(0));//104
  32.             int len = byteBuf.readableBytes();//可读的字节数12
  33.             System.out.println("len="+len);
  35.             //使用for取出各个字节
  36.             for (int i = 0; i < len; i++) {
  37.                 System.out.println((char)byteBuf.getByte(i));
  38.             }
  40.             //按照某个范围读取
  41.             System.out.println(byteBuf.getCharSequence(0,4,Charset.forName("utf-8")));
  42.             System.out.println(byteBuf.getCharSequence(4,6,Charset.forName("utf-8")));
  43.         }
  44.     }
  45. }

测试结果
在这里插入图片描述

11、Netty应用实例-群聊系统

实例要求:
1）编写一个Netty群聊系统，实现服务器端和客户端之间的数据简单通讯（非阻塞）
2）实现多人群聊
3）服务器端：可以监测用户上线，离线，并实现消息转发功能
4）客户端：通过channel可以无阻塞发送消息给其它所有用户，同时可以接受其它用户发送的消息(有服务器转发得到)
5）目的：进一步理解Netty非阻塞网络编程机制
在这里插入图片描述
代码如下：
GroupChatServer

  1. package com.wolfx.netty;
  3. import io.netty.bootstrap.ServerBootstrap;
  4. import io.netty.channel.*;
  5. import io.netty.channel.nio.NioEventLoopGroup;
  6. import io.netty.channel.socket.SocketChannel;
  7. import io.netty.channel.socket.nio.NioServerSocketChannel;
  8. import io.netty.handler.codec.string.StringDecoder;
  10. /**
  11.  * @description:
  12.  * @author: sukang
  13.  * @date: 2020-10-22 19:21
  14.  */
  15. public class GroupChatServer {
  17.     private int port; //监听端口
  19.     public GroupChatServer(int port){
  20.         this.port = port;
  21.     }
  23.     //编写run方法，处理客户端的请求
  24.     public void run() {
  25.         //创建两个线程组
  26.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);
  27.         EventLoopGroup workerGroup = new NioEventLoopGroup();//8个
  29.         try {
  30.             ServerBootstrap bootstrap = new ServerBootstrap();
  32.             bootstrap.group(bossGroup, workerGroup)
  33.                     .channel(NioServerSocketChannel.class)
  34.                     .option(ChannelOption.SO_BACKLOG,128)
  35.                     .childOption(ChannelOption.SO_KEEPALIVE,true)
  36.                     .childHandler(new ChannelInitializer<SocketChannel>(){
  38.                         @Override
  39.                         protected void initChannel(SocketChannel socketChannel) throws Exception {
  40.                             //获取到pipeline
  41.                             ChannelPipeline pipeline = socketChannel.pipeline();
  42.                             //向pipeline加入解码器
  43.                             pipeline.addLast("decoder", new StringDecoder());
  44.                             //向pipeline加入编码器
  45.                             //加入自己的业务处理handler
  46.                             pipeline.addLast(new GroupChatServerHandler());
  47.                         }
  48.                     });
  50.             System.out.println("netty 服务器启动");
  51.             ChannelFuture channelFuture = bootstrap.bind(port).sync();
  53.             //监听关闭
  54.             channelFuture.channel().closeFuture().sync();
  55.         } catch ( Exception e ) {
  56.             e.printStackTrace();
  57.         } finally {
  58.             bossGroup.shutdownGracefully();
  59.             workerGroup.shutdownGracefully();
  60.         }
  61.     }
  63.     public static void main(String[] args) {
  64.         new GroupChatServer(7000).run();
  65.     }
  66. }

GroupChatServerHandler

  1. package com.wolfx.netty;
  3. import io.netty.channel.Channel;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.channel.SimpleChannelInboundHandler;
  6. import io.netty.channel.group.ChannelGroup;
  7. import io.netty.channel.group.DefaultChannelGroup;
  8. import io.netty.util.concurrent.GlobalEventExecutor;
  10. import java.text.SimpleDateFormat;
  12. /**
  13.  * @description:
  14.  * @author: sukang
  15.  * @date: 2020-10-22 19:53
  16.  */
  17. public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> {
  19.     //public static List<Channel> channels = new ArrayList<Channel>();
  20.     //使用一个hashmap管理
  21.     //public static Map<String,Channel> channels = new HashMap<String,Channel>();
  22.     //定义一个channle组，管理所有的channel
  23.     //GlobalEventExecutor.INSTANCE)是全局的事件执行器，是一个单例
  24.     private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
  25.     SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-ddHH:mm:ss");
  27.     //handlerAdded表示连接建立，一旦连接，第一个被执行
  28.     //将当前channel加入到channelGroup
  29.     @Override
  30.     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
  31.         Channel channel = ctx.channel();
  32.         //将该客户加入聊天的信息推送给其它在线的客户端
  33.         /*该方法会将channelGroup中所有的channel遍历，并发送消息，我们不需要自己遍历*/
  34.         channelGroup.writeAndFlush("[客户端]"+channel.remoteAddress()+"加入聊天"+sdf.format(new java.util.Date())+"\n");
  35.         channelGroup.add(channel);
  36.     }
  38.     //断开连接，将xx客户端离开信息推送给当前在线的客户
  39.     @Override
  40.     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
  41.         Channel channel = ctx.channel();
  42.         channelGroup.writeAndFlush("[客户端]" + channel.remoteAddress()+"离开了\n");
  43.         System.out.println("channelGroup size" + channelGroup.size());
  44.     }
  46.     //表示channel处于活动状态,提示xx上线
  47.     @Override
  48.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  49.         System.out.println(ctx.channel().remoteAddress()+"上线了~");
  50.     }
  52.     //表示channel处于不活动状态,提示xx离线了
  53.     @Override
  54.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  55.         System.out.println(ctx.channel().remoteAddress()+"离线了~");
  56.     }
  58.     //读取数据
  59.     @Override
  60.     protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
  61.         //获取到当前channel
  62.         final Channel channel = ctx.channel();
  63.         //这是我们遍历channelGroup, 根据不同的情况，会送不同的消息
  64.         channelGroup.forEach(ch -> {
  65.             if(channel != ch){//不是当前channel, 转发消息
  66.                 ch.writeAndFlush("[客户]"+channel.remoteAddress()+"发送了消息"+msg+"\n");
  67.             }else{//回显自己发送的消息给自己
  68.                 ch.writeAndFlush("[自己]发送了消息"+msg+"\n");
  69.             }
  70.         });
  71.     }
  73.     @Override
  74.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  75.         //关闭通道
  76.         ctx.close();
  77.     }
  78. }

GroupChatClient

  1. package com.wolfx.netty;
  3. import io.netty.bootstrap.Bootstrap;
  4. import io.netty.channel.Channel;
  5. import io.netty.channel.ChannelFuture;
  6. import io.netty.channel.ChannelInitializer;
  7. import io.netty.channel.ChannelPipeline;
  8. import io.netty.channel.nio.NioEventLoopGroup;
  9. import io.netty.channel.socket.SocketChannel;
  10. import io.netty.channel.socket.nio.NioSocketChannel;
  11. import io.netty.handler.codec.string.StringDecoder;
  12. import io.netty.handler.codec.string.StringEncoder;
  14. import java.util.Scanner;
  16. /**
  17.  * @description:
  18.  * @author: sukang
  19.  * @date: 2020-10-22 20:26
  20.  */
  21. public class GroupChatClient {
  23.     //属性
  24.     private final String host;
  25.     private final int port;
  27.     public GroupChatClient(String host, int port) {
  28.         this.host = host;
  29.         this.port = port;
  30.     }
  32.     public void run() {
  33.         NioEventLoopGroup group = new NioEventLoopGroup();
  35.         try {
  36.             Bootstrap bootstrap = new Bootstrap()
  37.                     .group(group)
  38.                     .channel(NioSocketChannel.class)
  39.                     .handler(new ChannelInitializer<SocketChannel>() {
  40.                         @Override
  41.                         protected void initChannel(SocketChannel ch) throws Exception {
  42.                             //得到pipeline
  43.                             ChannelPipeline pipeline = ch.pipeline();
  44.                             //加入相关hander
  45.                             pipeline.addLast("decoder",new StringDecoder());
  46.                             pipeline.addLast("encoder",new StringEncoder());
  47.                             //加入自定义的handler
  48.                             pipeline.addLast(new GroupChatClientHandler());
  49.                         }
  50.                     });
  52.             ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
  53.             //得到channel
  54.             Channel channel = channelFuture.channel();
  55.             System.out.println("-------"+channel.localAddress()+"--------");
  56.             //客户端需要输入信息，创建一个扫描器
  57.             Scanner scanner = new Scanner(System.in);
  58.             while (scanner.hasNextLine()){
  59.                 String msg = scanner.nextLine();
  60.                 //通过channel发送到服务器端
  61.                 channel.writeAndFlush(msg+"\r\n");
  62.             }
  63.         } catch ( Exception e ) {
  64.             e.printStackTrace();
  65.         } finally {
  66.             group.shutdownGracefully();
  67.         }
  68.     }
  70.     public static void main(String[] args) {
  71.         new GroupChatClient("127.0.0.1", 7000).run();
  72.     }
  73. }

GroupChatClientHandler

  1. package com.wolfx.netty;
  3. import io.netty.channel.ChannelHandlerContext;
  4. import io.netty.channel.SimpleChannelInboundHandler;
  6. /**
  7.  * @description:
  8.  * @author: sukang
  9.  * @date: 2020-10-22 20:41
  10.  */
  11. public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> {
  13.     @Override
  14.     protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
  15.         System.out.println(msg.trim());
  16.     }
  17. }

四、TCP粘包，拆包及解决方法

在进行Java NIO学习时，发现，如果客户端连续不断的向服务端发送数据包时，服务端接收的数据会出现两个数据包粘在一起的情况，这就是TCP协议中经常会遇到的粘包以及拆包的问题。
我们都知道TCP属于传输层的协议，传输层除了有TCP协议外还有UDP协议。那么UDP是否会发生粘包或拆包的现象呢？答案是不会。UDP是基于报文发送的，从UDP的帧结构可以看出，在UDP首部采用了16bit来指示UDP数据报文的长度，因此在应用层能很好的将不同的数据报文区分开，从而避免粘包和拆包的问题。而TCP是基于字节流的，虽然应用层和TCP传输层之间的数据交互是大小不等的数据块，但是TCP把这些数据块仅仅看成一连串无结构的字节流，没有边界；另外从TCP的帧结构也可以看出，在TCP的首部没有表示数据长度的字段，基于上面两点，在使用TCP传输数据时，才有粘包或者拆包现象发生的可能。

1、粘包、拆包表现形式

现在假设客户端向服务端连续发送了两个数据包，用packet1和packet2来表示，那么服务端收到的数据可以分为三种，现列举如下：
第一种情况，接收端正常收到两个数据包，即没有发生拆包和粘包的现象，此种情况不在本文的讨论范围内。
在这里插入图片描述
第二种情况，接收端只收到一个数据包，由于TCP是不会出现丢包的，所以这一个数据包中包含了发送端发送的两个数据包的信息，这种现象即为粘包。这种情况由于接收端不知道这两个数据包的界限，所以对于接收端来说很难处理。
在这里插入图片描述
第三种情况，这种情况有两种表现形式，如下图。接收端收到了两个数据包，但是这两个数据包要么是不完整的，要么就是多出来一块，这种情况即发生了拆包和粘包。这两种情况如果不加特殊处理，对于接收端同样是不好处理的。
在这里插入图片描述

2、粘包、拆包发生原因

发生TCP粘包或拆包有很多原因，现列出常见的几点：
1、要发送的数据大于TCP发送缓冲区剩余空间大小，将会发生拆包。
2、待发送数据大于MSS（最大报文长度），TCP在传输前将进行拆包。
3、要发送的数据小于TCP发送缓冲区的大小，TCP将多次写入缓冲区的数据一次发送出去，将会发生粘包。
4、接收数据端的应用层没有及时读取接收缓冲区中的数据，将发生粘包。

3、粘包、拆包解决办法

通过以上分析，我们清楚了粘包或拆包发生的原因，那么如何解决这个问题呢？解决问题的关键在于如何给每个数据包添加边界信息，常用的方法有如下几个：
1、发送端给每个数据包添加包首部，首部中应该至少包含数据包的长度，这样接收端在接收到数据后，通过读取包首部的长度字段，便知道每一个数据包的实际长度了。
2、发送端将每个数据包封装为固定长度（不够的可以通过补0填充），这样接收端每次从接收缓冲区中读取固定长度的数据就自然而然的把每个数据包拆分开来。
3、可以在数据包之间设置边界，如添加特殊符号，这样，接收端通过这个边界就可以将不同的数据包拆分开。

五、用Netty自己实现dubboRPC

1、RPC基本介绍

1）RPC（RemoteProcedureCall）—远程过程调用，是一个计算机通信协议。该协议允许运行于一台计算机的程序调用另一台计算机的子程序，而程序员无需额外地为这个交互作用编程
2）两个或多个应用程序都分布在不同的服务器上，它们之间的调用都像是本地方法调用一样(如图)
在这里插入图片描述
3）常见的RPC框架有:比较知名的如阿里的Dubbo、google的gRPC、Go语言的rpcx、Apache的thrift，Spring旗下的SpringCloud。
![在这里插入图片描述](https://img-blog.csdnimg.cn/20201024141910710.png?x-oss-process=image/watermark,type\_ZmFuZ3poZW5naGVpdGk,shadow\_10,text\_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N1Y2hhaGFlcmthbmc=,size\_16,color\_FFFFFF,t\_70\#pic\_center

2、RPC调用流程图

在这里插入图片描述

3、RPC 调用流程说明

1）服务消费方(client)以本地调用方式调用服务
2）client stub接收到调用后负责将方法、参数等封装成能够进行网络传输的消息体
3）client stub将消息进行编码并发送到服务端
4）server stub收到消息后进行解码
5）server stub根据解码结果调用本地的服务
6）本地服务执行并将结果返回给server stub
7）server stub将返回导入结果进行编码并发送至消费方
8）client stub接收到消息并进行解码
9）服务消费方(client)得到结果

小结：RPC的目标就是将2-8这些步骤都封装起来，用户无需关心这些细节，可以像调用本地方法一样即可完成远程服务调用

4、自己实现dubboRPC(基于Netty)

需求说明
1）dubbo底层使用了Netty作为网络通讯框架，要求用Netty实现一个简单的RPC框架
2）模仿dubbo，消费者和提供者约定接口和协议，消费者远程调用提供者的服务，提供者返回一个字符串，消费者打印提供者返回的数据。底层网络通信使用Netty4.1.20

设计说明
1）创建一个接口，定义抽象方法。用于消费者和提供者之间的约定。
2）创建一个提供者，该类需要监听消费者的请求，并按照约定返回数据。
3）创建一个消费者，该类需要透明的调用自己不存在的方法，内部需要使用Netty请求提供者返回数据
4）开发的分析图
在这里插入图片描述
代码实现

  1. package com.wolfx.dubbo.publicinterface;
  3. /**
  4.  * @description:
  5.  *      这个是接口， 是提供方和服务消费方都需要
  6.  * @author: sukang
  7.  * @date: 2020-10-23 15:32
  8.  */
  9. public interface HelloService {
  11.     String hello(String msg);
  12. }
  14. package com.wolfx.dubbo.provider;
  16. import com.wolfx.dubbo.publicinterface.HelloService;
  18. /**
  19.  * @description:
  20.  * @author: sukang
  21.  * @date: 2020-10-23 15:37
  22.  */
  23. public class HelloServiceImpl implements HelloService {
  25.     private static int count = 0;
  27.     //当有消费方调用改方法时，就返回一个结果
  28.     @Override
  29.     public String hello(String msg) {
  30.         System.out.println("收到客户端消息=" + msg);
  32.         //根据msg返回不同的结果
  33.         if(msg != null){
  34.             return "你好客户端,我已经收到你的消息 [" + msg + "]第" + (++count) + " 次";
  35.         }else{
  36.             return "你好客户端,我已经收到你的消息";
  37.         }
  38.     }
  39. }
  41. package com.wolfx.dubbo.provider;
  43. import com.wolfx.dubbo.netty.NettyServer;
  45. /**
  46.  * @description:
  47.  *      ServerBootstrap会启动一个服务提供者，就是NettyServer
  48.  * @author: sukang
  49.  * @date: 2020-10-23 15:52
  50.  */
  51. public class ServerBootstrap {
  53.     public static void main(String[] args) {
  54.         NettyServer.startServer("127.0.0.1",7000);
  55.     }
  56. }
  58. package com.wolfx.dubbo.netty;
  60. import io.netty.bootstrap.ServerBootstrap;
  61. import io.netty.channel.ChannelFuture;
  62. import io.netty.channel.ChannelInitializer;
  63. import io.netty.channel.ChannelPipeline;
  64. import io.netty.channel.EventLoopGroup;
  65. import io.netty.channel.nio.NioEventLoopGroup;
  66. import io.netty.channel.socket.SocketChannel;
  67. import io.netty.channel.socket.nio.NioServerSocketChannel;
  68. import io.netty.handler.codec.string.StringDecoder;
  69. import io.netty.handler.codec.string.StringEncoder;
  71. /**
  72.  * @description:
  73.  * @author: sukang
  74.  * @date: 2020-10-23 15:53
  75.  */
  76. public class NettyServer {
  78.     public static void startServer(String hostName,int port){
  79.         startServer0(hostName, port);
  80.     }
  82.     //编写一个方法， 完成对NettyServer的初始化和启动
  83.     private static void startServer0(String hostname, int port){
  84.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);
  85.         EventLoopGroup workerGroup = new NioEventLoopGroup();
  87.         try {
  88.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  89.             serverBootstrap.group(bossGroup, workerGroup)
  90.                     .channel(NioServerSocketChannel.class)
  91.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  92.                         @Override
  93.                         protected void initChannel(SocketChannel ch) throws Exception {
  94.                             ChannelPipeline pipeline = ch.pipeline();
  95.                             pipeline.addLast(new StringDecoder());
  96.                             pipeline.addLast(new StringEncoder());
  97.                             pipeline.addLast(new NettyServerHandler());//业务处理器
  98.                         }
  99.                     });
  101.             ChannelFuture channelFuture = serverBootstrap.bind(hostname, port).sync();
  102.             System.out.println("服务提供方开始提供服务~~");
  103.             channelFuture.channel().closeFuture().sync();
  105.         } catch ( Exception e ) {
  106.             e.printStackTrace();
  107.         } finally {
  108.             bossGroup.shutdownGracefully();
  109.             workerGroup.shutdownGracefully();
  110.         }
  112.     }
  113. }
  115. package com.wolfx.dubbo.netty;
  117. import com.wolfx.dubbo.customer.ClientBootstrap;
  118. import com.wolfx.dubbo.provider.HelloServiceImpl;
  119. import io.netty.channel.ChannelHandlerContext;
  120. import io.netty.channel.ChannelInboundHandlerAdapter;
  122. /**
  123.  * @description:
  124.  * @author: sukang
  125.  * @date: 2020-10-23 16:07
  126.  */
  127. public class NettyServerHandler extends ChannelInboundHandlerAdapter {
  129.     @Override
  130.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  131.         //获取客户端发送的消息， 并调用服务
  132.         System.out.println("msg =" + msg);
  134.         //客户端在调用服务器api时，我们需要定义一个协议
  135.         //比如我们要求 每次发送消息是都必须以某个字符串开头 "HelloService#hello#你好"
  136.         if (msg.toString().startsWith(ClientBootstrap.providerName)){
  137.             String result = new HelloServiceImpl().hello(msg.toString().substring(msg.toString().lastIndexOf("#") + 1));
  138.             ctx.writeAndFlush(result);
  139.         }
  140.     }
  142.     @Override
  143.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  144.         ctx.close();
  145.     }
  146. }
  148. package com.wolfx.dubbo.netty;
  150. import io.netty.channel.ChannelHandlerContext;
  151. import io.netty.channel.ChannelInboundHandlerAdapter;
  153. import java.util.concurrent.Callable;
  155. /**
  156.  * @description:
  157.  * @author: sukang
  158.  * @date: 2020-10-24 13:29
  159.  */
  160. public class NettyClientHandler extends ChannelInboundHandlerAdapter implements Callable {
  162.     private ChannelHandlerContext context;//上下文
  163.     private String result; //返回的结果
  164.     private String para; //客户端调用方法时，传入的参数
  166.     //与服务器的连接创建后，就会被调用，这个方法是第一个被调用（1）
  167.     @Override
  168.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  169.         System.out.println("channelActive被调用");
  170.         context = ctx; //因为我们在其他方法会使用ctx
  171.     }
  173.     //收到服务器的数据后，调用方法(4)
  174.     @Override
  175.     public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  176.         System.out.println(" channelRead 被调用");
  177.         result = msg.toString();
  178.         notify();//唤醒等待的线程
  179.     }
  181.     @Override
  182.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  183.         ctx.close();
  184.     }
  186.     //被代理对象调用，发送数据给服务器，->wait->等待被唤醒（channelRead）->返回结果(3) -> 5
  187.     @Override
  188.     public Object call() throws Exception {
  189.         System.out.println(" call 被调用 ");
  190.         context.writeAndFlush(para);
  192.         //进行wait
  193.         wait();//等待channelRead方法获取到服务器的结果后，唤醒
  194.         System.out.println("call2 被调用");
  196.         return result;
  197.     }
  199.     //(2)
  200.     void setPara(String para){
  201.         System.out.println("setPara");
  202.         this.para = para;
  203.     }
  204. }
  206. package com.wolfx.dubbo.netty;
  208. import io.netty.bootstrap.Bootstrap;
  209. import io.netty.channel.ChannelInitializer;
  210. import io.netty.channel.ChannelOption;
  211. import io.netty.channel.ChannelPipeline;
  212. import io.netty.channel.nio.NioEventLoopGroup;
  213. import io.netty.channel.socket.SocketChannel;
  214. import io.netty.channel.socket.nio.NioSocketChannel;
  215. import io.netty.handler.codec.string.StringDecoder;
  216. import io.netty.handler.codec.string.StringEncoder;
  218. import java.lang.reflect.Proxy;
  219. import java.util.concurrent.ExecutorService;
  220. import java.util.concurrent.Executors;
  222. /**
  223.  * @description:
  224.  * @author: sukang
  225.  * @date: 2020-10-24 13:46
  226.  */
  227. public class NettyClient {
  229.     //创建线程池
  230.     private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
  232.     private static NettyClientHandler client;
  233.     private int count = 0;
  235.     //编写方法使用代理模式，获取一个代理对象
  236.     public Object getBean(final Class<?> serviceClass, final String providerName){
  237.         return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
  238.                 new Class<?>[]{serviceClass},(proxy,method,args) ->{
  239.                     System.out.println("(proxy,method,args)进入...."+(++count)+"次");
  240.                     //{}部分的代码，客户端每调用一次hello,就会进入到该代码
  241.                     if(client == null){
  242.                         initClient();
  243.                     }
  245.                     //设置要发给服务器端的信息
  246.                     //providerName 协议头args[0]就是客户端调用apihello(???),参数
  247.                     client.setPara(providerName + args[0]);
  249.                     return executor.submit(client).get();
  250.                 });
  251.     }
  253.     //初始化客户端
  254.     private static void initClient(){
  255.         client = new NettyClientHandler();
  256.         //创建EventLoopGroup
  257.         NioEventLoopGroup group = new NioEventLoopGroup();
  258.         Bootstrap bootstrap = new Bootstrap();
  259.         bootstrap.group(group)
  260.                 .channel(NioSocketChannel.class)
  261.                 .option(ChannelOption.TCP_NODELAY,true)
  262.                     .handler(new ChannelInitializer<SocketChannel>(){
  264.                     @Override
  265.                     protected void initChannel(SocketChannel ch) throws Exception {
  266.                         ChannelPipeline pipeline = ch.pipeline();
  267.                         pipeline.addLast(new StringDecoder());
  268.                         pipeline.addLast(new StringEncoder());
  269.                         pipeline.addLast(client);
  270.                     }
  271.                 });
  273.         try {
  274.             bootstrap.connect("127.0.0.1",7000).sync();
  275.         } catch ( Exception e ) {
  276.             e.printStackTrace();
  277.         }
  278.     }
  279. }
  281. package com.wolfx.dubbo.customer;
  283. import com.wolfx.dubbo.netty.NettyClient;
  284. import com.wolfx.dubbo.publicinterface.HelloService;
  286. /**
  287.  * @description:
  288.  * @author: sukang
  289.  * @date: 2020-10-24 14:07
  290.  */
  291. public class ClientBootstrap {
  293.     //这里定义协议头
  294.     public static final String providerName  = "HelloService#hello#";
  296.     public static void main(String[] args) throws InterruptedException {
  297.         //创建一个消费者
  298.         NettyClient customer = new NettyClient();
  299.         //创建代理对象
  300.         HelloService service = (HelloService)customer.getBean(HelloService.class, providerName);
  302.         for (;;) {
  303.             Thread.sleep(2*1000);
  304.             //通过代理对象调用服务提供者的方法(服务)
  305.             String res = service.hello("你好 dubbo~");
  306.             System.out.println("调用的结果 res=" + res);
  307.         }
  308.     }
  309. }

发表评论

表情：
评论列表 （有 0 条评论，243人围观）

还没有评论，来说两句吧...

相关阅读

    相关 netty学习笔记

    Netty与原生Nio的区别 原生的nio，只简单封闭了ByteBuffer以及Channel，也就是网络IO操作以及对这些字节的缓存，字节的编码与解码、粘包与拆包

    野性酷女野性酷女/ 0 赞/ 357 阅读

    相关 netty学习笔记一

    netty是基于nio（非阻塞io）的高性能网络通信框架 什么是阻塞 ？ 一直等待在那里，知道有返回值才会处理 例如： i. Socket socket = serve

    蔚落蔚落/ 0 赞/ 323 阅读

    相关 netty学习笔记二

    Netty架构，特性、模块组件、线程模型(Boss线程、worker线程)以及源码、 netty架构图： ![70][] netty架构五部分： Extensible

    傷城~傷城~/ 0 赞/ 278 阅读