JDK1.4的NIO有效解決了原有流式IO存在的線程開銷的問題，在NIO中使用多線程，主要目的已不是為了應對每個客戶端請求而分配獨立的服務線程，而是通過多線程充分使用用多個CPU的處理能力和處理中的等待時間，達到提高服務能力的目的。

 

 

JDK1.4的NIO有效解決了原有流式IO存在的線程開銷的問題，在NIO中使用多線程，主要目的已不是為了應對每個客戶端請求而分配獨立的服務線程，而是通過多線程充分使用用多個CPU的處理能力和處理中的等待時間，達到提高服務能力的目的。

線程模型

NIO的選擇器采用了多路復用（Multiplexing）技術，可在一個選擇器上處理多個套接字，通過獲取讀寫通道來進行IO操作。由于網(wǎng)絡帶寬等原因，在通道的讀、寫操作中是容易出現(xiàn)等待的，所以在讀、寫操作中引入多線程，對性能提高明顯，而且可以提高客戶端的感知服務質量。所以本文的模型將主要通過使用讀、寫線程池來提高與客戶端的數(shù)據(jù)交換能力。

同時整個服務端的流程處理，建立于事件機制上。在 [接受連接－＞讀－＞業(yè)務處理－＞寫 －＞關閉連接
]這個過程中，觸發(fā)器將觸發(fā)相應事件，由事件處理器對相應事件分別響應，完成服務器端的業(yè)務處理。
下面我們就來詳細看一下這個模型的各個組成部分。

相關事件定義 在這個模型中，我們定義了一些基本的事件：

（1）onAccept：

當服務端收到客戶端連接請求時，觸發(fā)該事件。通過該事件我們可以知道有新的客戶端呼入。該事件可用來控制服務端的負載。例如，服務器可設定同時只為一定數(shù)量客戶端提供服務，當同時請求數(shù)超出數(shù)量時，可在響應該事件時直接拋出異常，以拒絕新的連接。

（2）onAccepted：

當客戶端請求被服務器接受后觸發(fā)該事件。該事件表明一個新的客戶端與服務器正式建立連接。

（3）onRead：

當客戶端發(fā)來數(shù)據(jù)，并已被服務器控制線程正確讀取時，觸發(fā)該事件。該事件通知各事件處理器可以對客戶端發(fā)來的數(shù)據(jù)進行實際處理了。需要注意的是，在本模型中，客戶端的數(shù)據(jù)讀取是由控制線程交由讀線程完成的，事件處理器不需要在該事件中進行專門的讀操作，而只需將控制線程傳來的數(shù)據(jù)進行直接處理即可。

（4）onWrite：

當客戶端可以開始接受服務端發(fā)送數(shù)據(jù)時觸發(fā)該事件，通過該事件，我們可以向客戶端發(fā)送回應數(shù)據(jù)。在本模型中，事件處理器只需要在該事件中設置 。

（5）onClosed：

當客戶端與服務器斷開連接時觸發(fā)該事件。

（6）onError：

當客戶端與服務器從連接開始到最后斷開連接期間發(fā)生錯誤時觸發(fā)該事件。通過該事件我們可以知道有什么錯誤發(fā)生。

事件回調機制的實現(xiàn)

在這個模型中，事件采用廣播方式，也就是所有在冊的事件處理器都能獲得事件通知。這樣可以將不同性質的業(yè)務處理，分別用不同的處理器實現(xiàn)，使每個處理器的業(yè)務功能盡可能單一。

如下圖：整個事件模型由監(jiān)聽器、事件適配器、事件觸發(fā)器、事件處理器組成。

（事件模型）

1.監(jiān)聽器（Serverlistener）：

這是一個事件接口，定義需監(jiān)聽的服務器事件，如果您需要定義更多的事件，可在這里進行擴展。

1. public interface Serverlistener
2. {
3. public void onError(String error);
4. public void onAccept() throws Exception;
5. public void onAccepted(Request request) throws Exception;
6. public void onRead(Request request) throws Exception;
7. public void onWrite(Request request, Response response) throws Exception;
8. public void onClosed(Request request) throws Exception;
9. }

2. 事件適配器（EventAdapter）：

對Serverlistener接口實現(xiàn)一個適配器(EventAdapter)，這樣的好處是最終的事件處理器可以只處理所關心的事件。

1. public abstract class EventAdapter
2. implements Serverlistener
3. {
4. public EventAdapter() {}
5. public void onError(String error) {}
6. public void onAccept() throws Exception {}
7. public void onAccepted(Request request) throws Exception {}
8. public void onRead(Request request) throws Exception {}
9. public void onWrite(Request request, Response response) throws Exception {}
10. public void onClosed(Request request) throws Exception {}
11. }

3. 事件觸發(fā)器（Notifier）：

用于在適當?shù)臅r候通過觸發(fā)服務器事件，通知在冊的事件處理器對事件做出響應。觸發(fā)器以Singleton模式實現(xiàn)，統(tǒng)一控制整個服務器端的事件，避免造成混亂。

1. public class Notifier
2. {
3. private static Arraylist listeners = null;
4. private static Notifier instance = null;
5. private Notifier()
6. {
7. listeners = new Arraylist();
8. }
9. /**
10. * 獲取事件觸發(fā)器
11. * @return 返回事件觸發(fā)器
12. */
13. public static synchronized Notifier
14. getNotifier()
15. {
16. if (instance == null)
17. {
18. instance = new Notifier();
19. return instance;
20. }
21. else
22. {
23. return instance;
24. }
25. }
26. /**
27. * 添加事件監(jiān)聽器
28. * @param l 監(jiān)聽器
29. */
30. public void addlistener(Serverlistener l)
31. {
32. synchronized (listeners)
33. {
34. if (!listeners.contains(l))
35. {
36. listeners.add(l);
37. }
38. }
39. }
40. public void fireOnAccept()
41. throws Exception
42. {
43. for (int i = listeners.size() – 1;
44. i >= 0; i–)
45. {
46. ( (Serverlistener) listeners.
47. get(i)).onAccept();
48. }
49. }
50. // other fire method
51. }

4. 事件處理器（Handler）：

繼承事件適配器，對感興趣的事件進行響應處理，實現(xiàn)業(yè)務處理。以下是一個簡單的事件處理器實現(xiàn)，它響應onRead事件，在終端打印出從客戶端讀取的數(shù)據(jù)。

1. public class ServerHandler
2. extends EventAdapter
3. {
4. public ServerHandler() {}
5. public void onRead(Request request)
6. throws Exception
7. {
8. System.out.println(“Received: “ +
9. new String(data));
10. }
11. }

5. 事件處理器的注冊。

為了能讓事件處理器獲得服務線程的事件通知，事件處理器需在觸發(fā)器中注冊。

1. ServerHandler handler = new ServerHandler();
2. Notifier.addlistener(handler);

實現(xiàn)NIO多線程服務器

NIO多線程服務器主要由主控服務線程、讀線程和寫線程組成。

1. 主控服務線程（Server）：

主控線程將創(chuàng)建讀、寫線程池，實現(xiàn)監(jiān)聽、接受客戶端請求，同時將讀、寫通道提交由相應的讀線程（Reader）和寫服務線程(Writer），由讀寫線程分別完成對客戶端數(shù)據(jù)的讀取和對客戶端的回應操作。

1. public class Server implements Runnable
2. {
3. private static List wpool = new LinkedList();
4. private static Selector selector;
5. private ServerSocketChannel sschannel;
6. private InetSocketAddress address;
7. protected Notifier notifier;
8. private int port;
9. private static int MAX_THREADS = 4;
10. /**
11. * Creat the main thread
12. * @param port server port
13. * @throws java.lang.Exception
14. */
15. public Server(int port) throws Exception
16. {
17. this.port = port;
18. // event dispatcher
19. notifier = Notifier.getNotifier();
20. // create the thread pool for reading and writing
21. for (int i = 0; i < MAX_THREADS; i++)
22. {
23. Thread r = new Reader();
24. Thread w = new Writer();
25. r.start();
26. w.start();
27. }
28. // create nonblocking socket
29. selector = Selector.open();
30. sschannel = ServerSocketChannel.open();
31. sschannel.configureBlocking(false);
32. address = new InetSocketAddress(port);
33. ServerSocket ss = sschannel.socket();
34. ss.bind(address);
35. sschannel.register(selector, SelectionKey.OP_ACCEPT);
36. }
37. public void run()
38. {
39. System.out.println(“Server started “);
40. System.out.println(“Server listening on port: “ + port);
41. while (true)
42. {
43. try
44. {
45. int num = 0;
46. num = selector.select();
47. if (num > 0)
48. {
49. Set selectedKeys = selector.selectedKeys();
50. Iterator it = selectedKeys.iterator();
51. while (it.hasNext())
52. {
53. SelectionKey key = (SelectionKey) it.next();
54. it.remove();
55. if ((key.readyOps() & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT)
56. {
57. // Accept the new connection
58. ServerSocketChannel ssc =
59. (ServerSocketChannel) key.channel();
60. notifier.fireOnAccept();
61. SocketChannel sc = ssc.accept();
62. sc.configureBlocking(false);
63. Request request = new Request(sc);
64. notifier.fireOnAccepted(request);
65. sc.register(selector, SelectionKey.OP_READ,request);
66. }
67. else if ((key.readyOps() & SelectionKey.OP_READ) == SelectionKey.OP_READ)
68. {
69. Reader.processRequest(key);
70. key.cancel();
71. }
72. else if ((key.readyOps() & SelectionKey.OP_WRITE) == SelectionKey.OP_WRITE)
73. {
74. Writer.processRequest(key);
75. key.cancel();
76. }
77. }
78. }
79. //this selector’s wakeup method is invoked
80. else
81. {
82. //register new channel for writing to selector
83. addRegister();
84. }
85. }
86. catch (Exception e)
87. {
88. notifier.fireOnError(“Error occured in Server: “
89. + e.getMessage());
90. continue;
91. }
92. }
93. }
94. private void addRegister()
95. {
96. synchronized (wpool)
97. {
98. while (!wpool.isEmpty())
99. {
100. SelectionKey key = (SelectionKey) wpool.remove(0);
101. SocketChannel schannel = (SocketChannel) key.channel();
102. try
103. {
104. schannel.register(selector, SelectionKey.OP_WRITE, key
105. .attachment());
106. }
107. catch (Exception e)
108. {
109. try
110. {
111. schannel.finishConnect();
112. schannel.close();
113. schannel.socket().close();
114. notifier.fireOnClosed((Request) key.attachment());
115. }
116. catch (Exception e1)
117. {
118. }
119. notifier.fireOnError(“Error occured in addRegister: “
120. + e.getMessage());
121. }
122. }
123. }
124. }
125. public static void processWriteRequest(SelectionKey key)
126. {
127. synchronized (wpool)
128. {
129. wpool.add(wpool.size(), key);
130. wpool.notifyAll();
131. }
132. selector.wakeup();
133. }
134. }

2. 讀線程（Reader）：

使用線程池技術，通過多個線程讀取客戶端數(shù)據(jù)，以充分利用網(wǎng)絡數(shù)據(jù)傳輸?shù)臅r間，提高讀取效率。

1. public class Reader extends Thread
2. {
3. public void run()
4. {
5. while (true)
6. {
7. try
8. {
9. SelectionKey key;
10. synchronized (pool)
11. {
12. while (pool.isEmpty())
13. {
14. pool.wait();
15. }
16. key = (SelectionKey) pool.remove(0);
17. }
18. // 讀取客戶端數(shù)據(jù)，并觸發(fā)onRead事件
19. read(key);
20. }
21. catch (Exception e)
22. {
23. continue;
24. }
25. }
26. }
27. }

3. 寫線程（Writer）：

和讀操作一樣，使用線程池，負責將服務器端的數(shù)據(jù)發(fā)送回客戶端。

1. public final class Writer extends Thread
2. {
3. public void run()
4. {
5. while (true)
6. {
7. try
8. {
9. SelectionKey key;
10. synchronized (pool)
11. {
12. while (pool.isEmpty())
13. {
14. pool.wait();
15. }
16. key = (SelectionKey) pool.remove(0);
17. }
18. write(key);
19. }
20. catch (Exception e)
21. {
22. continue;
23. }
24. }
25. }
26. }

具體應用

NIO多線程模型的實現(xiàn)告一段落，現(xiàn)在我們可以暫且將NIO的各個API和煩瑣的調用方法拋于腦后，專心于我們的實際應用中。

我們用一個簡單的TimeServer（時間查詢服務器）來看看該模型能帶來多么簡潔的開發(fā)方式。

在這個TimeServer中，將提供兩種語言（中文、英文）的時間查詢服務。我們將讀取客戶端的查詢命令（GB/EN），并回應相應語言格式的當前時間。在應答客戶的請求的同時，服務器將進行日志記錄。做為示例，對日志記錄，我們只是簡單地將客戶端的訪問時間和IP地址輸出到服務器的終端上。

1. 實現(xiàn)時間查詢服務的事件處理器（TimeHandler）：

1. public class TimeHandler extends EventAdapter
2. {
3. public TimeHandler() {}
4. public void onWrite(Request request, Response response) throws Exception
5. {
6. String command = new String(request.getDataInput());
7. String time = null;
8. Date date = new Date();
9. if (command.equals(“GB”))
10. {
11. DateFormat cnDate = DateFormat.getDateTimeInstance(DateFormat.FulL,
12. DateFormat.FulL, Locale.CHINA);
13. time = cnDate.format(date);
14. }
15. else
16. {
17. DateFormat enDate = DateFormat.getDateTimeInstance(DateFormat.FulL,
18. DateFormat.FulL, Locale.US);
19. time = enDate.format(date);
20. }
21. response.send(time.getBytes());
22. }
23. }

2. 實現(xiàn)日志記錄服務的事件處理器（LogHandler）：

1. public class LogHandler extends EventAdapter
2. {
3. public LogHandler() {}
4. public void onClosed(Request request)
5. throws Exception
6. {
7. String log = new Date().toString() + ” from “ + request.getAddress().toString();
8. System.out.println(log);
9. }
10. public void onError(String error)
11. {
12. System.out.println(“Error: “ + error);
13. }
14. }

3. 啟動程序：

1. public class Start
2. {
3. public static void main(String[] args)
4. {
5. try
6. {
7. LogHandler loger = new LogHandler();
8. TimeHandler timer = new TimeHandler();
9. Notifier notifier = Notifier.getNotifier();
10. notifier.addlistener(loger);
11. notifier.addlistener(timer);
12. System.out.println(“Server starting “);
13. Server server = new Server(5100);
14. Thread tServer = new Thread(server);
15. tServer.start();
16. }
17. catch (Exception e)
18. {
19. System.out.println(“Server error: “ + e.getMessage());
20. System.exit(-1);
21. }
22. }
23. }

小  結

通過例子我們可以看到，基于事件回調的NIO多線程服務器模型，提供了清晰直觀的實現(xiàn)方式，可讓開發(fā)者從NIO及多線程的技術細節(jié)中擺脫出來，集中精力關注具體的業(yè)務實現(xiàn)。