网络框架基本上会有如下几个基本结构:

  • Read request
  • Decode request
  • Process service
  • Encode reply
  • Send reply

经典的服务端设计:

代码实现:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class Server implements Runnable {
public void run() {
try {
ServerSocket ss = new ServerSocket(PORT);
while (!Thread.interrupted())
new Thread(new Handler(ss.accept())).start(); //创建新线程来handle
// or, single-threaded, or a thread pool
} catch (IOException ex) { /* ... */ }
}

static class Handler implements Runnable {
final Socket socket;
Handler(Socket s) { socket = s; }
public void run() {
try {
byte[] input = new byte[MAX_INPUT];
socket.getInputStream().read(input);
byte[] output = process(input);
socket.getOutputStream().write(output);
} catch (IOException ex) { /* ... */ }
}
private byte[] process(byte[] cmd) { /* ... */ }
}
}

对于每一个请求都分发给一个线程,每个线程中都独自处理上面的流程。

这种模型由于 IO 在阻塞时会一直等待,因此在用户负载增加时,性能下降的非常快。

server 导致阻塞的原因:

  1. serversocket的accept方法,阻塞等待client连接,直到client连接成功。

  2. 线程从socket inputstream读入数据,会进入阻塞状态,直到全部数据读完。

  3. 线程向socket outputstream写入数据,会阻塞直到全部数据写完。

client 导致阻塞的原因:

  1. client建立连接时会阻塞,直到连接成功。

  2. 线程从 socket 输入流读入数据,如果没有足够数据读完会进入阻塞状态,直到有数据或者读到输入流末尾。

  3. 线程从 socket 输出流写入数据,直到输出所有数据。

  4. socket.setsolinger() 设置 socket 的延迟时间,当 socket 关闭时,会进入阻塞状态,直到全部数据都发送完或者超时。

可伸缩性的设计目标

  • 当负载增加完美的降级方案
  • 增加资源的利用率
  • 也满足可用性和性能目标
    • 短延迟
    • 满足高峰要求
    • 高可用
  • 通常分治法是最有效的解决方案

分治发设计

  • 把执行分成小的任务(每一个任务没有阻塞的执行一个 action)
  • 当任务是启用状态的时候执行它(通常 IO 事件作为触发器)
  • java nio 以及有基本的支持
    • Non-blocking 读写
    • IO 事件触发

改进:采用基于事件驱动的设计,当有事件触发时,才会调用处理器进行数据处理。

java.nio Support

  • Channels:连接文件,socket 并且支持非阻塞的读
  • Buffers:Channels 读和写的缓存
  • Selectors:通知Channels的IO事件集合
  • SelectionKeys:记录IO事件状态和绑定的Channel和Selector

代码如下:

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
class Reactor implements Runnable {
final Selector selector;
final ServerSocketChannel serverSocket;
Reactor(int port) throws IOException { //Reactor初始化
selector = Selector.open();
serverSocket = ServerSocketChannel.open();
serverSocket.socket().bind(new InetSocketAddress(port));
serverSocket.configureBlocking(false); //非阻塞
SelectionKey sk = serverSocket.register(selector, SelectionKey.OP_ACCEPT); //分步处理,第一步,接收accept事件
sk.attach(new Acceptor()); //attach callback object, Acceptor
}

public void run() {
try {
while (!Thread.interrupted()) {
selector.select();
Set selected = selector.selectedKeys();
Iterator it = selected.iterator();
while (it.hasNext())
dispatch((SelectionKey)(it.next()); //Reactor负责dispatch收到的事件
selected.clear();
}
} catch (IOException ex) { /* ... */ }
}

void dispatch(SelectionKey k) {
Runnable r = (Runnable)(k.attachment()); //调用之前注册的callback对象
if (r != null)
r.run();//注意这里面使用的是run方法,而不是创建Thread的start
}

class Acceptor implements Runnable { // inner class
public void run() {
try {
SocketChannel c = serverSocket.accept();
if (c != null)
new Handler(selector, c);
}
catch(IOException ex) { /* ... */ }
}
}
}

final class Handler implements Runnable {
final SocketChannel socket;
final SelectionKey sk;
ByteBuffer input = ByteBuffer.allocate(MAXIN);
ByteBuffer output = ByteBuffer.allocate(MAXOUT);
static final int READING = 0, SENDING = 1;
int state = READING;

Handler(Selector sel, SocketChannel c) throws IOException {
socket = c; c.configureBlocking(false);
// Optionally try first read now
sk = socket.register(sel, 0);
sk.attach(this); //将Handler作为callback对象
sk.interestOps(SelectionKey.OP_READ); //第二步,接收Read事件
sel.wakeup();
}
boolean inputIsComplete() { /* ... */ }
boolean outputIsComplete() { /* ... */ }
void process() { /* ... */ }

public void run() {
try {
if (state == READING) read();
else if (state == SENDING) send();
} catch (IOException ex) { /* ... */ }
}

void read() throws IOException {
socket.read(input);
if (inputIsComplete()) {
process();
state = SENDING;
// Normally also do first write now
sk.interestOps(SelectionKey.OP_WRITE); //第三步,接收write事件
}
}
void send() throws IOException {
socket.write(output);
if (outputIsComplete()) sk.cancel(); //write完就结束了, 关闭select key
}
}

//上面 的实现用Handler来同时处理Read和Write事件, 所以里面出现状态判断
//我们可以用State-Object pattern来更优雅的实现
class Handler { // ...
public void run() { // initial state is reader
socket.read(input);
if (inputIsComplete()) {
process();
sk.attach(new Sender()); //状态迁移, Read后变成write, 用Sender作为新的callback对象
sk.interest(SelectionKey.OP_WRITE);
sk.selector().wakeup();
}
}
class Sender implements Runnable {
public void run(){ // ...
socket.write(output);
if (outputIsComplete()) sk.cancel();
}
}
}

这里用到了 Reactor 模式。

关于 Reactor 模式的一些概念:

Reactor:负责响应 IO 事件,当检测到一个新的事件,将其发送给相应的Handler去处理。

Handler:负责处理非阻塞的行为,标识系统管理的资源;同时将 handler 与事件绑定。

Reactor为单个线程,需要处理 accept 连接,同时发送请求到处理器中。

由于只有单个线程,所以处理器中的业务需要能够快速处理完。

改进:使用多线程处理业务逻辑。

代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class Handler implements Runnable {
// uses util.concurrent thread pool
static PooledExecutor pool = new PooledExecutor(...);
static final int PROCESSING = 3;
// ...
synchronized void read() { // ...
socket.read(input);
if (inputIsComplete()) {
state = PROCESSING;
pool.execute(new Processer()); //使用线程pool异步执行
}
}

synchronized void processAndHandOff() {
process();
state = SENDING; // or rebind attachment
sk.interest(SelectionKey.OP_WRITE); //process完,开始等待write事件
}

class Processer implements Runnable {
public void run() { processAndHandOff(); }
}
}

从上面代码以及示意图可以看出来,把业务逻辑的执行放在了线程池里面了,这个就可以避免了,业务逻辑等待过程的阻塞。

但Reactor仍为单个线程。

继续改进:对于多个CPU的机器,为充分利用系统资源,将Reactor拆分为两部分。

1
2
3
4
5
6
7
8
9
10
Selector[] selectors; //subReactors集合, 一个selector代表一个subReactor
int next = 0;
class Acceptor { // ...
public synchronized void run() { ...
Socket connection = serverSocket.accept(); //主selector负责accept
if (connection != null)
new Handler(selectors[next], connection); //选个subReactor去负责接收到的connection
if (++next == selectors.length) next = 0;
}
}

mainReactor 负责监听连接,accept 连接给 subReactor 处理,为什么要单独分一个 Reactor 来处理监听呢?因为像 TCP 这样需要经过3次握手才能建立连接,这个建立连接的过程也是要耗时间和资源的,单独分一个 Reactor 来处理,可以提高性能。

【参考资料】

  1. http://gee.cs.oswego.edu/dl/cpjslides/nio.pdf

  2. http://www.cnblogs.com/luxiaoxun/p/4331110.html

—EOF—