1. IO

1.1 BIO

1.1.1 介绍

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

1.1.2 实例

BIO服务器

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
package com.kag.bio;

import org.apache.log4j.Logger;
import java.io.IOException;
import java.io.InputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
* @Author: KHighness
* @Date: 2020/9/13 11:07
* @Description: BIO服务器
*/

public class BIOServer {

private Logger logger = Logger.getLogger(BIOServer.class);

/**
* 启动服务器
* @throws IOException
*/
public void start() throws IOException {

/**
* 线程池机制
* 思路
* 1、创建一个线程池
* 2、如果有多个线程池联机呃,就创建一个线程,与之通讯
*/

ExecutorService newCachedThreadPool = Executors.newCachedThreadPool();

// 创建ServerSocket
ServerSocket serverSocket = new ServerSocket(6666);
logger.info("NIO 服务器启动 ");

while (true) {
// 监听,等待客户端接受
final Socket socket = serverSocket.accept();
logger.info("新增一个客户端连接");
// 创建一个线程,与之通讯
newCachedThreadPool.execute(new Runnable() {
public void run() {
handler(socket);
}
});
}
}

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

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

1.1.3 配置

1.1.4 运行

1
telnet 127.0.0.1 6666

1.1.5 不足

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

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

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

1.2 NIO

1.2.1 介绍

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

1.2.2 比较

BIO VS NIO

  • BIO以流的方式处理数据,而NIO以块的方式处理数据,块I/O的效率比流I/O高很多

  • BIO是阻塞的,NIO则是非阻塞的

  • BIO基于字节流和字符流进行操作,而NIO基于Channel(通道)和Buffer(缓冲区)进行操作,数据总是从通道读取到缓冲区中,或者从缓冲区写入到同道中人。Selectors(选择器)用于监听多个通道的事件(比如:连接请求、数据到达等),因此使用单个线程就可以监听多个客户端通道

1.2.3 组件

三大组件:SelectorChanneBuffer

关系图

说明

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

说明

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

常用Buffer子类:

  • ByteBuffer
  • ShortBuffer
  • CharBuffer
  • IntBuffer
  • LongBuffer
  • DoubleBuffer
  • FloatBuffer

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

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

mark <= position <= limit <= capacity

1.2.3.2 Channel

说明

通道:NIO的通道类似于流,但有些区别

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

常用的Channel类:

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

FileChannel的主要IO操作

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

说明

选择器:

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

类及相关方法

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

说明

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

1.2.3 编程

将数据写入到本地文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public static void write(String s) throws IOException {
// 创建输出流 ——> channel
FileOutputStream fileOutputStream = new FileOutputStream("D:\\Java\\Test\\K1.txt");

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

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

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

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

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

从本地文件读取数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public static void read(String path) throws IOException {
// 创建文件的输入流
File file = new File(path);
FileInputStream fileInputStream = new FileInputStream(file);

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

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

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

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

拷贝文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public static void copy(String source, String destination) throws IOException {
FileInputStream fileInputStream = new FileInputStream(source);
FileChannel fileChannel1 = fileInputStream.getChannel();

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

ByteBuffer byteBuffer = ByteBuffer.allocate(1024);

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

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

使用transferFrom完成拷贝

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public static void copyImage(String fromPath, String toPath) throws IOException {
// 创建相关流
FileInputStream fileInputStream = new FileInputStream(fromPath);
FileOutputStream fileOutputStream = new FileOutputStream(toPath);

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

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

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

buffer数组完成读写操作

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
// Scattering: 将数据写入到buffer时,可以采用buffer数组,依次写入[分散]
// Gathering: 从buffer读取数据时,可以采用buffer数组,依次读取[聚集]
public class ScatteringAndGatheringTest {

public static void demo() throws Exception{
// 使用ServerSocketChannel 和 SocketChannel 网络
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
InetSocketAddress inetSocketAddress = new InetSocketAddress(7000);

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

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

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

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

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

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

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

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

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

1.2.4 实例

群聊系统Demo

编码步骤:

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

NIOServer

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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
package com.kag.nio.groupchat;

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

/**
* @Author: KHighness
* @Date: 2020/9/19 22:59
* @Description:
*/

public class GroupChatServer {

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

private Selector selector;

private ServerSocketChannel listenChannel;

private final int PORT = 3333;

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

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

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

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

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

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

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

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

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

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

NIOClient

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
package com.kag.nio.groupchat;

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

/**
* @Author: KHighness
* @Date: 2020/9/20 0:53
* @Description:
*/

public class GroupChatClient {

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

private final String HOST = "127.0.0.1";

private final int PORT = 3333;

private Selector selector;

private SocketChannel socketChannel;

private String username;

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

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

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

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

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

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

1.2.5 零拷贝

说明

  • 零拷贝,是从OS(操作系统)的角度来说的。因为内核缓冲区之间,没有数据是重复的

  • 零拷贝不仅仅带来更少的数据复制,还能带来其他的性能优势,更少的CPU缓存伪共享以及无CPU校验和计算

mmap

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

mmap和sendFile的区别

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

2. 概述

官网: Netty.io

1.1 介绍

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

1.2 设计

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

1.3 性能

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

1.4 架构

3. Reactor

2.1 介绍

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

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

2.2 组成

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

2.3 分类

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

2.2.1 单Reactor单线程

图示

分析

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

2.2.2 单Reactor多线程

图示

方案说明

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

分析

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

2.2.3 主从Reactor多线程

图示

方案说明

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

分析

  1. 优点:父线程与子线程的职责明确,父线程只需要接收新连接,子线程完成后续的业务处理

  2. 优点:父线程与子线程的数据交互简单,Reactor主线程只需要把新连接传给子线程,子线程无需返回数据

  3. 缺点:编程复杂度较高

2.4 优点

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

4. 架构

3.1 图示

3.2 说明

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

3.3 代码示例

NettyServer.java

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
package com.kag.netty.simple;

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

/**
* @author: KHighness
* @date: 2020/9/21 21:53
* @apiNote:
*/

public class NettyServer {

public static void main(String[] args) throws Exception{

/*
* 创建BossGroup 和 WorkerGroup
* 说明:
* 1、创建两个线程组:bossGroup 和 workerGroup
* 2、bossGroup处理连接请求
* 3、workerGroup处理和客户端的业务
* 4、两个都是无限循环
*/
EventLoopGroup bossGroup = new NioEventLoopGroup();
NioEventLoopGroup workerGroup = new NioEventLoopGroup();

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

// 使用链式编程来进行设置
bootstrap.group(workerGroup, workerGroup) // 设置两个线程组
.channel(NioServerSocketChannel.class) // 使用NioSocketChannel作为服务器的通道实现
.option(ChannelOption.SO_BACKLOG, 128) // 设置线程队列得到连接个数
.childOption(ChannelOption.SO_KEEPALIVE, true) // 设置保持活动连接状态
.childHandler(new ChannelInitializer<SocketChannel>() { // 创建一个通道测试对象
// 给Pipeline设置处理器
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new NettyServerHandler());
}
}); // 给workerGroup的EventLoop对应的管道设置处理器

System.out.println("服务器已就绪···");

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

// 对关闭通道进行监听
channelFuture.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}

NettyHandler.java

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
package com.kag.netty.simple;

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

/**
* @author: KHighness
* @date: 2020/9/21 22:20
* @apiNote:
*/

/**
* 说明:
* 自定义一个Handler需要继承Netty规定好的某个HandlerAdapter
*/
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

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

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

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

NettyClient.java

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
package com.kag.netty.simple;

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

/**
* @author: KHighness
* @date: 2020/9/21 22:20
* @apiNote:
*/

/**
* 说明:
* 自定义一个Handler需要继承Netty规定好的某个HandlerAdapter
*/
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

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

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

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

NettyClientHandler.java

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
package com.kag.netty.simple;

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

/**
* @author: KHighness
* @date: 2020/9/22 19:09
* @apiNote:
*/

public class NettyClientHandler extends ChannelInboundHandlerAdapter {

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

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

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

3.4 任务队列

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

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

3.5 异步模型

3.5.1 基本介绍

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

3.5.2 Future说明

  1. 表示异步的执行结果,可以通过它提供的方法来检测执行是否完成

  2. ChannelFuture是一个接口: public interface ChannelFuture extends Future

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

3.5.3 Future-Listener机制

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

  2. 常见操作:

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

例如:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// 绑定一个端口并且同步
// 绑定端口并启动服务器
ChannelFuture CF = bootstrap.bind(3333).sync();

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

3.6 Http服务

TestServer.java

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
package com.kag.netty.http;

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

/**
* @author: KHighness
* @date: 2020/9/26 10:34
* @apiNote:
*/

public class TestServer {
public static void main(String[] args) {

EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();

try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new TestServerInitializer());

ChannelFuture CF = serverBootstrap.bind(3333).sync();

CF.channel().closeFuture().sync();

} catch (InterruptedException e) {
System.out.println(e.getMessage());
} finally{
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}

TestServerInitializer.java

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
package com.kag.netty.http;

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

/**
* @author: KHighness
* @date: 2020/9/26 10:39
* @apiNote:
*/

public class TestServerInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
// 向管道加入处理器

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

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

TestHttpServerHandler.java

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
package com.kag.netty.http;

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

/**
* @author: KHighness
* @date: 2020/9/26 10:39
* @apiNote:
*/

public class TestServerInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
// 向管道加入处理器

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

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

5. 核心

4.1 BootStrap、ServerBootStrap

说明

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

常用方法

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

4.2 Future、ChannelFuture

说明

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

常用方法

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

4.3 Channel

说明

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

常见Channel类型

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

4.4 Selector

说明

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

4.5 ChannelHandler及其实现类

说明

  • ChannelHandler是一个接口,处理I/O事件或拦截I/O操作,并将其转发到ChannelPipeline(业务处理链)中的下一个处理程序

  • ChannelHandler本身并没有提供很多方法,因为这个接口有许多的方法需要实现,方便使用期间,可以继承它的子类

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

    一般需要重写的方法

    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
    public class ChannelInboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelInboundHandler {

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

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

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

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

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

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

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

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

    //通道发生异常事件
    @Skip
    @Override
    @SuppressWarnings("deprecation")
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
    throws Exception {
    ctx.fireExceptionCaught(cause);
    }
    }
    • ChannelInboundHandler 用于处理入站I/O事件
    • ChannelOutboundHandler 用于处理出战I/O事件

    适配器

  • ChannelInboundHandlerAdapter 用于处理入站 I/O 事件

  • ChannelOutboundHandlerAdapter 用于处理出站 I/O 操作

  • ChannelDuplexHandler 用于处理入站和出站事件

4.6 Pipeline、Channel

说明

  • ChannelPipeline是一个Handler的集合,它负责处理和拦截inboundoutbound的事件和操作,相当于一个贯穿Netty的链(通俗的讲:ChannelPipeline是保存ChannelHandlerlist,用于处理或拦截Channel的入站事件和出战操作)

  • ChanenelPipline实现了一种高级形式的拦截过滤器模式,使用户可以完全控制事件的处理方式,以及Channel中各个的ChannelHandler如何相互交互

  • Netty中每个Channel都有且仅有一个ChannelPipeline与之对应,他们的组成关系如下

  • 一个Channel包含一个ChannelPipeline,而ChannelPipeline中又维护了一个由ChannelHandlerContext组成的双向链表,并且每个ChannelHandlerContext中又关联着一个ChannelHandler

  • 入站事件和出战事件在一个双向链表中,入站事件会从链表head往后传递到最后一个入站的handler,出站事件会从链表tail往前传递到最前一个出站的handler,两种类型的handler互不干扰

常用方法

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

4.7 ChannelHandlerContext

说明

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

常见方法

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

4.8 ChannelOption

说明

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

参数如下:

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

4.9 EventLoopGroup和其实现类NioEventLoopGroup

说明

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

常用方法

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

4.10 Unpooled

说明

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

常用方法

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

4.11 群聊系统

要求

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

GroupChatServer.java

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
package com.kag.netty.groupChat;

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

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

public class GroupChatServer {

private int port;

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

public void run() {

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

ServerBootstrap serverBootstrap = new ServerBootstrap();

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

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

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

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

GroupChatServerHandler.java

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
package com.kag.netty.groupChat;

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

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

/**
* @author: KHighness
* @date: 2020/10/2 8:33
* @apiNote:
*/

public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> {

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

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

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

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

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

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

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

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

}

GroupChatClient.java

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
package com.kag.netty.groupChat;

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

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

/**
* @author: KHighness
* @date: 2020/10/2 9:22
* @apiNote:
*/

public class GroupChatClient {

private final String host;
private final int port;

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

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

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

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

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

GroupChatClientHandler.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
package com.kag.netty.groupChat;

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

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

public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> {

@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
System.out.println(s.trim());
}
}

4.12 心跳检测

要求

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

MyServer.java

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
package com.kag.netty.heartbeat;

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

import java.util.concurrent.TimeUnit;

/**
* @author: KHighness
* @date: 2020/10/2 10:25
* @apiNote:
*/

public class MyServer {
public static void main(String[] args) throws InterruptedException {
NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
NioEventLoopGroup workerGroup = new NioEventLoopGroup();

try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO)) //为BossGroup中的请求添加日志处理Handler
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
/**
* 加入一个 netty 提供的 IdleStateHandler
* 说明
* 1、IdleStateHandler 是 netty 提供的检测空闲状态的处理器
* 2、long readerIdleTime:表示多长时间没有读,就会发送一个心跳检测包检测是否还是连接的状态
* 3、long writerIdleTime:表示多长时间没有写,就会发送一个心跳检测包检测是否还是连接的状态
* 4、long allIdleTime:表示多长时间没有读写,就会发送一个心跳检测包检测是否还是连接的状态
* 5、当 IdleStateEvent 触发后,就会传递给管道的下一个 Handler,通过调用(触发)下一个Handler的 userEventTriggered,在该方法区处理这个事件。
*/
pipeline.addLast(new IdleStateHandler(3, 5, 7, TimeUnit.SECONDS));

//加入一个对空闲检测进一步处理的Handler(自定义)
pipeline.addLast(new MyServerHandler());
}
});
//启动服务器,设置为同步模式
ChannelFuture channelFuture = serverBootstrap.bind(3333).sync();
channelFuture.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}

MyServerHandler.java

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
package com.kag.netty.heartbeat;

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

/**
* @author: KHighness
* @date: 2020/10/2 10:48
* @apiNote:
*/

public class MyServerHandler extends ChannelInboundHandlerAdapter {

/**
*
* @param ctx 上下文
* @param evt 事件
* @throws Exception
*/
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent) {
// 将evt向下转型
IdleStateEvent event = (IdleStateEvent) evt;
String eventType = null;
switch (event.state()) {
case READER_IDLE:
eventType = "读空闲";
break;
case WRITER_IDLE:
eventType = "写空闲";
break;
case ALL_IDLE:
eventType = "读写空闲";
break;
}
System.out.println(ctx.channel().remoteAddress() + "——>超时:" + eventType);

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

4.13 简单总结

服务器

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

客户端

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

常见Handler

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

Handler常用方法

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