Java解讀NIO Socket非阻塞模式

NIO
有一個(gè)主要的類Selector,這個(gè)類似一個(gè)觀察者，只要我們把需要探知的socketchannel告訴Selector,我們接著做別的事情，當(dāng)有
事件發(fā)生時(shí)，他會(huì)通知我們，傳回一組SelectionKey,我們讀取這些Key,就會(huì)獲得我們剛剛注冊(cè)過(guò)的socketchannel,然后，我們從
這個(gè)Channel中讀取數(shù)據(jù)，放心，包準(zhǔn)能夠讀到，接著我們可以處理這些數(shù)據(jù)。

NIO主要原理和適用

NIO 有一個(gè)主要的類Selector,這個(gè)類似一個(gè)觀察者，只要我們把需要探知的socketchannel告訴Selector,我們接著做別的事情，當(dāng)有
事件發(fā)生時(shí)，他會(huì)通知我們，傳回一組SelectionKey,我們讀取這些Key,就會(huì)獲得我們剛剛注冊(cè)過(guò)的socketchannel,然后，我們從
這個(gè)Channel中讀取數(shù)據(jù)，放心，包準(zhǔn)能夠讀到，接著我們可以處理這些數(shù)據(jù)。

Selector內(nèi)部原理實(shí)際是在做一個(gè)對(duì)所注冊(cè)的channel的輪詢?cè)L問(wèn)，不斷的輪詢(目前就這一個(gè)算法)，一旦輪詢到一個(gè)channel有所注冊(cè)的事情發(fā)生，比如數(shù)據(jù)來(lái)了，他就會(huì)站起來(lái)報(bào)告，交出一把鑰匙，讓我們通過(guò)這把鑰匙來(lái)讀取這個(gè)channel的內(nèi)容。

jdk供的無(wú)阻塞I/O（NIO）有效解決了多線程服務(wù)器存在的線程開銷問(wèn)題，但在使用上略顯得復(fù)雜一些。在NIO中使用多線程，主要目的已不是為了應(yīng)對(duì)
每個(gè)客戶端請(qǐng)求而分配獨(dú)立的服務(wù)線程，而是通過(guò)多線程充分使用用多個(gè)CPU的處理能力和處理中的等待時(shí)間，達(dá)到提高服務(wù)能力的目的。

這段時(shí)間在研究NIO，寫篇博客來(lái)記住學(xué)過(guò)的東西。還是從最簡(jiǎn)單的Hello
World開始，client多線程請(qǐng)求server端，server接收client的名字，并返回Hello!
＋名字的字符格式給client。當(dāng)然實(shí)際應(yīng)用并不這么簡(jiǎn)單，實(shí)際可能是訪問(wèn)文件或者數(shù)據(jù)庫(kù)獲取信息返回給client。非阻塞的NIO有何神秘之處？

代 碼：

1）server端代碼

1. public class HelloWorldServer {
2. static int BLOCK = 1024;
3. static String name = “”;
4. protected Selector selector;
5. protected ByteBuffer clientBuffer = ByteBuffer.allocate(BLOCK);
6. protected CharsetDecoder decoder;
7. static CharsetEncoder encoder = Charset.forName(“GB2312”).newEncoder();
8. public HelloWorldServer(int port) throws IOException {
9. selector = this.getSelector(port);
10. Charset charset = Charset.forName(“GB2312”);
11. decoder = charset.newDecoder();
12. }
13. // 獲取Selector
14. protected Selector getSelector(int port) throws IOException {
15. ServerSocketChannel server = ServerSocketChannel.open();
16. Selector sel = Selector.open();
17. server.socket().bind(new InetSocketAddress(port));
18. server.configureBlocking(false);
19. server.register(sel, SelectionKey.OP_ACCEPT);
20. return sel;
21. }
22. // 監(jiān)聽端口
23. public void listen() {
24. try {
25. for (;;) {
26. selector.select();
27. Iterator iter = selector.selectedKeys().iterator();
28. while (iter.hasNext()) {
29. SelectionKey key = (SelectionKey) iter.next();
30. iter.remove();
31. process(key);
32. }
33. }
34. } catch (IOException e) {
35. e.printStackTrace();
36. }
37. }
38. // 處理事件
39. protected void process(SelectionKey key) throws IOException {
40. if (key.isAcceptable()) { // 接收請(qǐng)求
41. ServerSocketChannel server = (ServerSocketChannel) key.channel();
42. SocketChannel channel = server.accept();
43. //設(shè)置非阻塞模式
44. channel.configureBlocking(false);
45. channel.register(selector, SelectionKey.OP_READ);
46. } else if (key.isReadable()) { // 讀信息
47. SocketChannel channel = (SocketChannel) key.channel();
48. int count = channel.read(clientBuffer);
49. if (count > 0) {
50. clientBuffer.flip();
51. CharBuffer charBuffer = decoder.decode(clientBuffer);
52. name = charBuffer.toString();
53. // System.out.println(name);
54. SelectionKey sKey = channel.register(selector,
55. SelectionKey.OP_WRITE);
56. sKey.attach(name);
57. } else {
58. channel.close();
59. }
60. clientBuffer.clear();
61. } else if (key.isWritable()) { // 寫事件
62. SocketChannel channel = (SocketChannel) key.channel();
63. String name = (String) key.attachment();
64. ByteBuffer block = encoder.encode(CharBuffer
65. .wrap(“Hello !” + name));
66. channel.write(block);
67. //channel.close();
68. }
69. }
70. public static void main(String[] args) {
71. int port = 8888;
72. try {
73. HelloWorldServer server = new HelloWorldServer(port);
74. System.out.println(“listening on “ + port);
75. server.listen();
76. } catch (IOException e) {
77. e.printStackTrace();
78. }
79. }
80. }

2）client端代碼

1. public class HelloWorldClient {
2. static int SIZE = 10;
3. static InetSocketAddress ip = new InetSocketAddress(“localhost”, 8888);
4. static CharsetEncoder encoder = Charset.forName(“GB2312”).newEncoder();
5. static class Message implements Runnable {
6. protected String name;
7. String msg = “”;
8. public Message(String index) {
9. this.name = index;
10. }
11. public void run() {
12. try {
13. long start = System.currentTimeMillis();
14. //打開Socket通道
15. SocketChannel client = SocketChannel.open();
16. //設(shè)置為非阻塞模式
17. client.configureBlocking(false);
18. //打開選擇器
19. Selector selector = Selector.open();
20. //注冊(cè)連接服務(wù)端socket動(dòng)作
21. client.register(selector, SelectionKey.OP_CONNECT);
22. //連接
23. client.connect(ip);
24. //分配內(nèi)存
25. ByteBuffer buffer = ByteBuffer.allocate(8 * 1024);
26. int total = 0;
27. _FOR: for (;;) {
28. selector.select();
29. Iterator iter = selector.selectedKeys().iterator();
30. while (iter.hasNext()) {
31. SelectionKey key = (SelectionKey) iter.next();
32. iter.remove();
33. if (key.isConnectable()) {
34. SocketChannel channel = (SocketChannel) key
35. .channel();
36. if (channel.isConnectionPending())
37. channel.finishConnect();
38. channel
39. .write(encoder
40. .encode(CharBuffer.wrap(name)));
41. channel.register(selector, SelectionKey.OP_READ);
42. } else if (key.isReadable()) {
43. SocketChannel channel = (SocketChannel) key
44. .channel();
45. int count = channel.read(buffer);
46. if (count > 0) {
47. total += count;
48. buffer.flip();
49. while (buffer.remaining() > 0) {
50. byte b = buffer.get();
51. msg += (char) b;
52. }
53. buffer.clear();
54. } else {
55. client.close();
56. break _FOR;
57. }
58. }
59. }
60. }
61. double last = (System.currentTimeMillis() – start) * 1.0 / 1000;
62. System.out.println(msg + “used time :” + last + “s.”);
63. msg = “”;
64. } catch (IOException e) {
65. e.printStackTrace();
66. }
67. }
68. }
69. public static void main(String[] args) throws IOException {
70. String names[] = new String[SIZE];
71. for (int index = 0; index < SIZE; index++) {
72. names[index] = “jeff[“ + index + “]”;
73. new Thread(new Message(names[index])).start();
74. }
75. }
76. }