Netty-04-线程模型

字数统计: 6.5k字   |   阅读时长≈ 29分钟

Netty-04-线程模型

前言

原生NIO存在的问题

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

Netty优点

  • Netty对JDK自带的NIO的API进行了封装，解决了上述问题。
    • 设计优雅：适用于各种传输类型的统一 API 阻塞和非阻塞 Socket；基于灵活且可扩展的事件模型，可以清晰地分离关注点；高度可定制的线程模型 - 单线程，一个或多个线程池.
    • 安全：完整的 SSL/TLS 和 StartTLS 支持
    • 高性能、吞吐量更高：延迟更低；减少资源消耗；最小化不必要的内存复制。

1. I/O线程模型

目前存在的线程模型主要有：

  • 传统阻塞I/O服务模型
  • Reactor模式

根据Reactor的数量和处理资源池线程的数量不同，有如下3种典型的实现

  • 单Reactor单线程
  • 单Reactor多线程
  • 主从Reactor多线程
  • Netty线程模型主要基于主从Reactor多线程模型做了一定的改进，其中主从Reactor多线程模型有多个Reactor。

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

mark

图解说明：黄色的框表示对象，蓝色的框表示线程、白色的框表示方法（API）。之后的图相同。

1.1.1 模型分析

模型特点：

  • 采用阻塞IO模式获取输入的数据
  • 每个链接都需要独立的线程完成数据的输入，业务处理、数据返回。

问题分析：

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

1.1.2 代码实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
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;

public class test {
    public static void main(String[] args) throws IOException {
        // 1. 创建一个线程池
        // 2. 如果有客户端前来链接，就创建一个线程与之通信（单独的一个方法）
        ExecutorService executorService = Executors.newCachedThreadPool();

        // 创建ServerSocket
        ServerSocket serverSocket = new ServerSocket(6666);
        System.out.println("服务器启动了");

        while (true){
            // 监听等待客户端的连接
            final Socket socket = serverSocket.accept();
            System.out.println("连接到一个客户端请求");
            // 创建一个线程与之通信
            executorService.execute(()->{
                // 重写Runnable 方法，与客户端进行同i性能
                handler(socket);
            });
        }
    }

    // 编写一个handler方法，和客户端通讯，主要进行数据的读取和业务的处理
    public static void handler(Socket socket){
        try {
            byte[] bytes = new byte[1024];

            // 通过socket获取输入流
            InputStream inputStream = socket.getInputStream();
            // 循环的读取客户端发送来的数据
            while (true){
                int read = inputStream.read(bytes);
                if (read != -1){
                    System.out.println(new String(bytes, 0, read)); // 输出客户端发送的数据
                }else{
                    break;
                }
            }
        }catch (IOException e){
            e.printStackTrace();
        }finally {
            System.out.println("关闭和client的链接");
            try{
                socket.close();
            }catch (IOException e){
                e.printStackTrace();
            }
        }
    }
}

1.2 Reactor 模型概述

针对传统阻塞I/O服务模型的2个缺点，解决方案如下：

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

I/O复用结合线程池，就是Reactor模式基本设计思想，如图所示：

mark

  • Reactor 模式，通过一个或者多个输入同时传递给服务处理器模式（基于事件驱动）
  • 服务端端程序处理传入的多个请求，并将它们同步分派到相应的处理线程，因此Reactor模式也叫做Dispatcher 模式
  • Reactor模式使用的IO多路复用监听事件，收到事件后，分发到某个线程或者进程，这点就是网络服务高并发处理的关键。

Reactor模式中的核心组成部分：

  • Reactor：Reactor在一个单独的线程中运行，负责监听和分发事件，分发给适当的处理程序来对IO事件作出反应。
  • Handlers：处理程序执行I/O事件要完成的实际事件，类似于客户想要与之交谈的公司中的实际官员。Reactor通过调度适当的处理程序来响应I/O事件，处理程序执行非阻塞操作。

1.3 单Reactor 单线程模式

mark

方案说明：

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

mark

1.3.1 模型分析

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

1.3.2 代码实现

这里面我为了简便，我将Reactor和Acceptor和Handler三个对象搞成了方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
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.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

public class test {
    private Selector selector;
    private ServerSocketChannel serverSocketChannel;
    private int PORT = 6666;

    // 初始化操作
    public test() {
        try {
            selector = Selector.open();
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.bind(new InetSocketAddress(PORT));
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    // 对客户端进行监听事件
    public void listen(){
        try {
            while (true){
                int count = selector.select();
                // 返回值大于0表示有客户端产生事件
                if (count > 0){
                    // 取出产生事件的channel
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    // 对事件遍历
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while (iterator.hasNext()){
                        SelectionKey next = iterator.next();
                        // 将key进行分发
                        dispatch(next);
                        iterator.remove();
                    }
                }
            }
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    // 分发
    public void dispatch(SelectionKey key){
        if (key.isAcceptable()){
            // 如果是连接事件
            accept(key);
        }else{
            // 其他事件
            handler(key);
        }
    }

    // 建立新的连接
    private void accept(SelectionKey key){
        try {
            SocketChannel socketChannel = serverSocketChannel.accept();
            socketChannel.configureBlocking(false);
            socketChannel.register(selector,SelectionKey.OP_READ);
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    // 对请求进行处理，接收消息--业务处理--返回消息
    private void handler(SelectionKey key){
        SocketChannel channel = null;

        try{
            channel = (SocketChannel) key.channel();
            ByteBuffer buffer = ByteBuffer.allocate(3);
            StringBuilder msg = new StringBuilder();

            while (channel.read(buffer) > 0){
                msg.append(new String(buffer.array()));
                buffer.clear();
            }

            System.out.println("接收到消息：" + msg.toString());

            // 发送消息
            String ok = "OK";
            buffer.put(ok.getBytes());
            //这个flip非常重要哦，是将position置0，limit置于position的位置，
            // 以便下面代码进行写入操作能够正确写入buffer中的所有数据
            buffer.flip();
            channel.write(buffer);
            buffer.clear();
        }catch (IOException e){
            try {
                System.out.println(channel.getRemoteAddress() + "离线了");
                //取消该通道的注册并关闭通道，这里非常重要，没有这一步的话当客户端断开连接就会不断抛出IOException
                //是因为，select会一直产生该事件。
                key.cancel();
                channel.close();
            }catch (IOException ex){
                ex.printStackTrace();
            }
        }
    }

    // 调用
    public static void main(String[] args) {
        test test = new test();
        test.listen();
    }
}

这里有更牛逼更完整的Reactor单线程模型的代码案例：https://www.cnblogs.com/hama1993/p/10611229.html

1.4 单Reactor 多线程模式

mark

方案说明：

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

1.4.1 模型分析

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

1.4.2 代码实现

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

    private SelectionKey selectionKey;

    public ReadHandler(SelectionKey selectionKey) {
        this.selectionKey = selectionKey;
    }


    public void run() {
        try {
            SocketChannel sc = (SocketChannel) selectionKey.channel();
            //处理读请求
            doRead(sc);
            //处理完读请求，将通道注册为写
            Selector selector = selectionKey.selector();
            SelectionKey sk = sc.register(selector, SelectionKey.OP_WRITE);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    private void doRead(SocketChannel ssc) {
        System.out.println("读取数据，然后做一些数据处理");
    }
}
  • 在ReadHandler中，doRead(sc)，方法是同步的，而一般读到一个请求要经历解析请求信息，然后交到对应的service处理业务等等操作，非常耗时，此时将doRead进行异步处理，将能提高服务器资源的使用，提高性能。
  • 所以单Reactor单线程模型的实现，就是将单线程模型中的Handler中的数据处理异步处理，并引入线程池管理这些线程，这样dispatch将专注于事件分发，而各自handler专注于各自的事件处理。
  • 单Reactor多线程
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/**
 * 单Reactor多线程模型
 */
public class OneReactorMultiThreadMode {

    public static void main(String[] args) {
        /**
         * 初始化一个线程池，然后启动MultiReacor
         */
        ThreadPool.getPool().init(3);
        new MultiReactor(8089).run();

    }

}
  • MultiReactor
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class MultiReactor implements Runnable{

    private Selector selector;

    private ServerSocketChannel servChannel;

    private volatile boolean stop;

    public MultiReactor(int port) {
        try {
            selector = Selector.open();
            servChannel = ServerSocketChannel.open();
            servChannel.configureBlocking(false);
            servChannel.socket().bind(new InetSocketAddress(port), 1024);
            SelectionKey sk = servChannel.register(selector, SelectionKey.OP_ACCEPT);
            stop=false;
            System.out.println("The time server is start in port : " + port);
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
        }
    }
 @Override
    public void run() {
        while (!stop) {
            try {
                selector.select(1000);
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> it = selectedKeys.iterator();
                SelectionKey key = null;
                while (it.hasNext()) {
                    key = it.next();
                    it.remove();
                    try {
                        disptach(key);
                    } catch (Exception e) {
                        if (key != null) {
                            key.cancel();
                            if (key.channel() != null)
                                key.channel().close();
                        }
                    }
                }
            } catch (Throwable t) {
                t.printStackTrace();
            }
        }
    }

    private void disptach(SelectionKey key) {
        if(key.isValid()){
            if(key.isAcceptable()){
                new MultiAcceptor(key).run();
            }
            if (key.isReadable()){
                new MultiReadHandler(key).run();
            }
            if(key.isWritable()){
                new MultiWriteHandler(key).run();
            }
        }
    }
}
  • MultiAcceptor 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
class MultiAcceptor {

    private SelectionKey selectionKey;

    public MultiAcceptor(SelectionKey selectionKey) {
        this.selectionKey = selectionKey;
    }

    public void run() {
        try {
            ServerSocketChannel ssc = (ServerSocketChannel) selectionKey.channel();
            SocketChannel sc = ssc.accept();
            sc.configureBlocking(false);
            Selector selector = selectionKey.selector();
            SelectionKey sk = sc.register(selector, SelectionKey.OP_READ);
        }catch (Exception e){
            e.printStackTrace();
        }

    }
}
  • MultiReadHandler
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class MultiReadHandler {

    private SelectionKey selectionKey;

    public MultiReadHandler(SelectionKey selectionKey) {
        this.selectionKey = selectionKey;
    }


    
    public void run() {
        try {
            final SocketChannel sc = (SocketChannel) selectionKey.channel();
            //使用线程池，异步处理读请求
            ThreadPool.getPool().submit(new Runnable() {
                @Override
                public void run() {
                    doRead(sc);
                }
            });

            //处理完读请求，将通道注册为写
            Selector selector = selectionKey.selector();
            SelectionKey sk = sc.register(selector, SelectionKey.OP_WRITE);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    private void doRead(SocketChannel ssc) {
        System.out.println("读取数据，然后做一些数据处理");
    }
}
  • 线程池
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
/**
 * 示例-线程池
 */
class ThreadPool{

    public static final ThreadPool pool = new ThreadPool();

    private boolean init=false;

    private  static ExecutorService executorService;

    private ThreadPool(){};

    public synchronized void init(int size){
        if(!init){
            executorService=Executors.newFixedThreadPool(size);
            init=true;
        }else {
            System.out.println("the thread pool had inited");
        }

    }

    public static ThreadPool getPool(){
       return pool;
    }

    public  void submit(Runnable runnable){
        if(init){
            executorService.submit(runnable);
        }else {
            throw new RuntimeException("the thread pool is not inited");
        }
    }
}

1.5 主从Reactor模式

mark

mark

方案说明

  • Reactor主线程 MainReactor 对象就只注册一个用于监听连接请求的ServerSocketChannel，通过select 监听连接事件, 收到事件后，通过Acceptor 处理连接事件
  • 当 Acceptor 处理连接事件后，MainReactor 通过accept获取新的连接，并将连接注册到SubReactor
  • subreactor 将连接加入到连接队列进行监听,并创建handler进行各种事件处理
  • 当有新事件发生时， subreactor 就会调用对应的handler处理
  • handler 通过read 读取数据，分发给后面的worker 线程处理
  • worker 线程池分配独立的worker 线程进行业务处理，并返回结果
  • handler收到响应的结果后，再通过send将结果返回给client
  • Reactor主线程可以对应多个Reactor子线程, 即MainRecator可以关联多个SubReactor

1.5.1 模型分析

  • 优点：父线程与子线程的数据交互简单职责明确，父线程只需要接收新连接，子线程完成后续的业务处理。
  • 优点：父线程与子线程的数据交互简单，Reactor 主线程只需要把新连接传给子线程，子线程无需返回数据
  • 缺点：编程复杂度较高
  • 结合实例：这种模型在许多项目中广泛使用，包括 Nginx 主从 Reactor 多进程模型，Memcached 主从多线程，Netty 主从多线程模型的支持

1.5.2 代码实现

  • 主从Reactor多线程模型，将使用两个selector，一个主selector专门维护accept事件，当接收到accept事件，将该连接交给从selector，从selector维护该连接的read和write事件。
  • 主从selector的方式，将连接和数据处理完全分开维护，将大大提高并发量。
  • [参考博客][https://www.cnblogs.com/eason-ou/p/11912010.html]
1
2
3
4
5
6
7
8
9
10
11
12
13
/**
 * 主从Reactor多线程模型
 */
public class MainAndSubReactorMultiThreadMode {

    public static void main(String[] args) {
        /**
         * 初始化一个线程池，然后创建一个主Reactor，并加入一个从Reactor.
         */
        ThreadPool.getPool().init(3);
        new MainReactor(8089).addSub(new SubReactor()).run();
    }
}
  • 主Reactor
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
/**
 * 主Reactor
 */
class MainReactor{


    /**
     * 维护一个从Reactor
     */
    private SubReactor subReactor;

    private int port;


    private Selector selector;

    private ServerSocketChannel servChannel;

    private volatile boolean stop;

    public MainReactor(int port) {
        try {
            selector = Selector.open();
            servChannel = ServerSocketChannel.open();
            servChannel.configureBlocking(false);
            servChannel.socket().bind(new InetSocketAddress(port), 1024);
            SelectionKey sk = servChannel.register(selector, SelectionKey.OP_ACCEPT);
            stop=false;
            this.port=port;
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
        }
    }
/**
     * 添加子Reactor
     * @param subReactor
     * @return
     */
    public MainReactor addSub(SubReactor subReactor){
        this.subReactor=subReactor;
        this.subReactor.run();
        return this;
    }

    public void run() {
        System.out.println("主reactor开始启动了,监听端口："+port+".......");
        while (!stop) {
            try {
                selector.select(1000);
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> it = selectedKeys.iterator();
                SelectionKey key = null;
                while (it.hasNext()) {
                    key = it.next();
                    it.remove();
                    try {
                        disptach(key);
                    } catch (Exception e) {
                        if (key != null) {
                            key.cancel();
                            if (key.channel() != null)
                                key.channel().close();
                        }
                    }
                }
            } catch (Throwable t) {
                t.printStackTrace();
            }
        }
    }
private void disptach(SelectionKey key) {
        if(key.isValid()){
            /**
             * 主Reactor只关心Accept事件
             */
            if(key.isAcceptable()){
                new MultiAcceptor(key).addSub(this.subReactor).run();
            }

            //如果未使用了从Reactor
            if(this.subReactor==null){
                if (key.isReadable()){
                new MultiReadHandler(key).run();
            }
            if(key.isWritable()){
                new MultiWriteHandler(key).run();
            }
            }

        }
    }
}
  • 从Reactor
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
/**
 * 从Reactor
 */
class SubReactor {

    private Selector selector;

    private volatile boolean stop;

    public SubReactor() {
        try {
            selector = SelectorProvider.provider().openSelector();
            stop=false;
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
        }
    }

    /**
     * 将主Reactor中的Channel注册到从Reactor中的selector
     * @param sc
     */
    public void register(SocketChannel sc){
        try {
            sc.register(selector,SelectionKey.OP_READ|SelectionKey.OP_WRITE);
        } catch (ClosedChannelException e) {
            e.printStackTrace();
        }
    }
public void run() {
        ThreadPool.getPool().submit(new Runnable() {
            @Override
            public void run() {
                System.out.println("从reactor开始启动了。。。。。");
                while (!stop) {
                    try {
                        selector.select(1000);
                        Set<SelectionKey> selectedKeys = selector.selectedKeys();
                        Iterator<SelectionKey> it = selectedKeys.iterator();
                        SelectionKey key = null;
                        while (it.hasNext()) {
                            key = it.next();
                            it.remove();
                            try {
                                disptach(key);
                            } catch (Exception e) {
                                if (key != null) {
                                    key.cancel();
                                    if (key.channel() != null)
                                        key.channel().close();
                                }
                            }
                        }
                    } catch (Throwable t) {
                        t.printStackTrace();
                    }
                }
            }
        });

    }
private void disptach(SelectionKey key) {
        /**
         * 从Reactor只关心读和写事件
         */
        if(key.isValid()){

            if (key.isReadable()){
                new MultiReadHandler(key).run();
            }

            if(key.isWritable()){
                new MultiWriteHandler(key).run();
            }
        }
    }
}
  • acceptor
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class MultiAcceptor  {

    private SubReactor subReactor;

    private SelectionKey selectionKey;

    public MultiAcceptor(SelectionKey selectionKey) {
        this.selectionKey = selectionKey;
    }

    public MultiAcceptor addSub(SubReactor subReactor){
        this.subReactor=subReactor;
        return this;
    }
public void run() {
        try {
            ServerSocketChannel ssc = (ServerSocketChannel) selectionKey.channel();
            SocketChannel sc = ssc.accept();
            sc.configureBlocking(false);
            if(subReactor==null){
                Selector selector = selectionKey.selector();
                SelectionKey sk = sc.register(selector, SelectionKey.OP_READ);
            }else {
                System.out.println("accept");
                subReactor.register(sc);
            }

        }catch (Exception e){
            e.printStackTrace();
        }

    }
}
  • ReadHandler
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class MultiReadHandler {

    private SelectionKey selectionKey;

    public MultiReadHandler(SelectionKey selectionKey) {
        this.selectionKey = selectionKey;
    }



    public void run() {
        try {
            final SocketChannel sc = (SocketChannel) selectionKey.channel();
            //使用线程池，异步处理读请求
            ThreadPool.getPool().submit(new Runnable() {
                @Override
                public void run() {
                    doRead(sc);
                }
            });

            //处理完读请求，将通道注册为写
            Selector selector = selectionKey.selector();
            SelectionKey sk = sc.register(selector, SelectionKey.OP_WRITE);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    private void doRead(SocketChannel ssc) {
        System.out.println("读取数据，然后做一些数据处理");
    }
}
  • WriteHandler
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class MultiWriteHandler {

    private SelectionKey selectionKey;

    public MultiWriteHandler(SelectionKey selectionKey) {
        this.selectionKey = selectionKey;
    }



    private void doWrite(SocketChannel sc) {
        System.out.println("处理写。。。");
    }

    public void run() {
        try {
            final SocketChannel sc = (SocketChannel) selectionKey.channel();
            //使用线程池，异步处理写请求
            ThreadPool.getPool().submit(new Runnable() {
                @Override
                public void run() {
                    doWrite(sc);
                }
            });
            //写完后，将通道注册为读
            Selector selector = selectionKey.selector();
            SelectionKey sk = sc.register(selector, SelectionKey.OP_READ);
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}
  • 线程池
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
/**
 * 示例-线程池
 */
class ThreadPool{

    public static final ThreadPool pool = new ThreadPool();

    private boolean init=false;

    private  static ExecutorService executorService;

    private ThreadPool(){};

    public synchronized void init(int size){
        if(!init){
            executorService=Executors.newFixedThreadPool(size);
            init=true;
        }else {
            System.out.println("the thread pool had inited");
        }

    }

    public static ThreadPool getPool(){
       return pool;
    }

    public  void submit(Runnable runnable){
        if(init){
            executorService.submit(runnable);
        }else {
            throw new RuntimeException("the thread pool is not inited");
        }
    }
}

2. Netty 线程模型

2.1 主从Reactor进阶

  • Netty主要是基于主从Reactor多线程模式做了一定的改进，其中主从Reactor都有单一的一个变成了多个。下面是简单的改进图。

mark

  • 如图所示，增加了BossGroup来维护多个主Reactor，主Reactor还是只关注连接的Accept；增加了WorkGroup来维护多个从Reactor，从Reactor将接收到的请求交给Handler进行处理。
  • 在主Reactor中接收到Accept事件，获取到对应的SocketChannel，Netty会将它进一步封装成NIOSocketChannel对象，这个封装后的对象还包含了该Channel对应的SelectionKey、通信地址等详细信息
  • Netty会将装个封装后的Channel对象注册到WorkerGroup中的从Reactor中。
  • 当WorkerGroup中的从Reactor监听到事件后，就会将之交给与此Reactor对应的Handler进行处理。

稍微详细图如下

mark

  • Netty将Selector以及Selector相关的事件及任务封装了NioEventLoop，这样BossGroup就可以通过管理NioEventLoop去管理各个Selector。

  • 同时，Netty模型中主要存在两个大的线程池组BossGroup和WorkerGroup，用于管理主Reactor线程和从Reactor线程。

2.2 Netty 模型

mark

图中详细解释

  • Netty抽象出两组线程池，

    • BossGroup专门负责接收客户端的连接，
    • WorkerGroup专门负责网络的读写

  • BossGroup和WorkerGroup类型的本质都是NioEventLoopGroup类型。

  • NioEventLoopGroup相当于一个线程管理器（类似于ExecutorServevice），它下面维护很多个NioEventLoop线程。

    • 在初始化这两个线程组的时候，默认会在每个Group中生成CPU*2个NioEventLoop线程
    • 当n个连接来了，Group默认会按照连接请求的顺序分别将这些连接分给各个NioEventLoop去处理。
    • 同时Group还负责管理EventLoop的生命周期。

  • NioEventLoop表示一个不断循环的执行处理任务的线程

    • 它维护了一个线程和任务队列
    • 每个NioEventLoop都包含一个Selector，用于监听绑定在它上面的socket通讯。
    • 每个NioEventLoop相当于Selector，负责处理多个Channel上的事件
    • 每增加一个请求连接，NioEventLoopGroup就将这个请求依次分发给它下面的
    • NioEventLoop处理。

  • 每个Boss NioEventLoop循环执行的步骤有3步：

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

  • 每个Worker NioEventLoop循环执行的步骤：

    • 轮询read，write事件
    • 处理I/O事件，即read，write事件，在对应的NioSocketChannel中进行处理
    • 处理任务队列的任务，即runAllTasks

  • 每个 Worker NioEventLoop处理业务时，会使用pipeline（管道）

    • pipeline中维护了一个ChannelHandlerContext链表
    • ChannelHandlerContext则保存了Channel相关的所有上下文信息，同时关联一个ChannelHandler对象。
    • 如图所示，Channel和pipeline一一对应，ChannelHandler和ChannelHandlerContext一一对应。

mark

2.3 ChannelHandler

  • ChannelHandler是一个接口，负责处理或拦截I/O操作，并将其转发到Pipeline中的下一个处理Handler进行处理。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
                                               I/O Request
                                          via Channel or
                                      ChannelHandlerContext
                                                    |
+---------------------------------------------------+---------------+
|                           ChannelPipeline         |               |
|                                                  \|/              |
|    +---------------------+            +-----------+----------+    |
|    | Inbound Handler  N  |            | Outbound Handler  1  |    |
|    +----------+----------+            +-----------+----------+    |
|              /|\                                  |               |
|               |                                  \|/              |
|    +----------+----------+            +-----------+----------+    |
|    | Inbound Handler N-1 |            | Outbound Handler  2  |    |
|    +----------+----------+            +-----------+----------+    |
|              /|\                                  .               |
|               .                                   .               |
| ChannelHandlerContext.fireIN_EVT() ChannelHandlerContext.OUT_EVT()|
|        [ method call]                       [method call]         |
|               .                                   .               |
|               .                                  \|/              |
|    +----------+----------+            +-----------+----------+    |
|    | Inbound Handler  2  |            | Outbound Handler M-1 |    |
|    +----------+----------+            +-----------+----------+    |
|              /|\                                  |               |
|               |                                  \|/              |
|    +----------+----------+            +-----------+----------+    |
|    | Inbound Handler  1  |            | Outbound Handler  M  |    |
|    +----------+----------+            +-----------+----------+    |
|              /|\                                  |               |
+---------------+-----------------------------------+---------------+
                |                                  \|/
+---------------+-----------------------------------+---------------+
|               |                                   |               |
|       [ Socket.read() ]                    [ Socket.write() ]     |
|                                                                   |
|  Netty Internal I/O Threads (Transport Implementation)            |
+-------------------------------------------------------------------+

3. Netty 入门代码实例

  1. 服务器端
  • 服务端启动器代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
package NettyPro;

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

import java.io.IOException;

public class NettyServer {
    public static void main(String[] args) throws InterruptedException {
        //创建BossGroup 和 WorkerGroup
        //1、创建两个线程组，bossGroup 和 workerGroup
        //2、bossGroup 只是处理连接请求，真正的和客户端业务处理，会交给 workerGroup 完成
        //3、两个都是无限循环
        //4、bossGroup 和 workerGroup 含有的子线程（NioEventLoop）个数为实际 cpu 核数 * 2
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup worderGroup = new NioEventLoopGroup();

        try {
            //创建服务器端的启动对象，配置参数
            ServerBootstrap bootstrap = new ServerBootstrap();

            //使用链式编程来进行设置，配置
            bootstrap.group(bossGroup, worderGroup) //设置两个线程组
                    .channel(NioServerSocketChannel.class) //使用 NioServerSocketChannel 作为服务器的通道实现
                    .option(ChannelOption.SO_BACKLOG, 128) //设置线程队列得到连接个数
                    .childOption(ChannelOption.SO_KEEPALIVE, true) //设置保持活动连接状态
                    .childHandler(new ChannelInitializer<SocketChannel>() { //为accept channel的pipeline预添加的handler
                        //给 pipeline 添加处理器，每当有连接accept时，就会运行到此处。
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new NettyServerHandler());
                        }
                    }); //给我们的 workerGroup 的 EventLoop 对应的管道设置处理器

            System.out.println("........服务器 is ready......");
            //绑定一个端口并且同步，生成了一个ChannelFuture 对象
            //启动服务器（并绑定端口）
            ChannelFuture future = bootstrap.bind(6668).sync();

            //对关闭通道进行监听
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            worderGroup.shutdownGracefully();
        }
    }
}
  • 服务器端处理器代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
package NettyPro;

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

import java.nio.charset.Charset;

public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    /**
     *读取客户端发送过来的消息
     * @param ctx 上下文对象，含有 管道pipeline，通道channel，地址
     * @param msg 就是客户端发送的数据，默认Object
     */

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("服务器读取线程：" + Thread.currentThread().getName());
        System.out.println("server ctx = " + ctx);
        // 看看Channel 和 PipelLine的瓜西
        Channel channel = ctx.channel();
        ChannelPipeline pipeline = ctx.pipeline();

        // 将msg 转换成一个ByteBuf ，比较NIO的ByteBuffer 性能的提高
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("客户端发来的消息是 + " + buf.toString(CharsetUtil.UTF_8));
        System.out.println("客户端的地址是 + " + ctx.channel().remoteAddress());
    }

    // 数据读取完毕
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        // 它是 write + flush ，将数据写入到缓存buffer，并将buffer中的数据flush进入通道
        // 一般来讲，我们对这个发送的数据进行编码
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello 客户端！",CharsetUtil.UTF_8));
    }

    // 处理异常，一般是关闭通道
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}
  1. 客户端
  • 客户端的启动器代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
package NettyPro;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class NettyClient {
    public static void main(String[] args) throws Exception{
        // 客户端需要一个事件循环组
        NioEventLoopGroup group = new NioEventLoopGroup();

        // 创建启动器
        try {
            // 创建客户端启动对象
            // 注意： 客户端使用的不是ServerBootStrap 而是 Bootstrap
            Bootstrap bootstrap = new Bootstrap();

            // 设置相关参数
            bootstrap.group(group) // 设置线程组
                    .channel(NioSocketChannel.class) // 设置客户端通道的是实现类(使用反射实现)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new NettyClientHandler()); // 加入自己的处理器
                        }
                    });

            System.out.println("----- 客户端 is Ready----");

            // 启动客户端去连接服务器端
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6668).sync();
            // 给监听的通道关闭
            channelFuture.channel().closeFuture().sync();
        }finally {
            group.shutdownGracefully();
        }
    }
}
  • 客户端的处理器代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
package NettyPro;

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;

public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    /**
     * 当通道就绪就会触发
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("client is + " + 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 buf = (ByteBuf) msg;
        System.out.println("服务器回复的消息是：" + buf.toString(CharsetUtil.UTF_8));
        System.out.println("服务器的地址是 + " + ctx.channel().remoteAddress());
    }

    // 发生异常的关闭处理
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

参考博客 ： https://www.cnblogs.com/eason-ou/p/11912010.html

​ https://blog.csdn.net/Danny_idea/article/details/89286195

打赏
  • 版权声明： 本博客所有文章除特别声明外，均采用 Apache License 2.0 许可协议。转载请注明出处！
  • © 2019-2022 Zhuuu
  • PV: UV:

请我喝杯咖啡吧~

支付宝
微信