NIO代碼 · JAVA

[TOC] # 編解碼類CharsetHelper ~~~ package com.nio; import java.nio.ByteBuffer; import java.nio.CharBuffer; import java.nio.charset.CharacterCodingException; import java.nio.charset.Charset; import java.nio.charset.CharsetDecoder; import java.nio.charset.CharsetEncoder; public class CharsetHelper { private static final String UTF_8 = "UTF-8"; private static CharsetEncoder encoder = Charset.forName(UTF_8).newEncoder(); private static CharsetDecoder decoder = Charset.forName(UTF_8).newDecoder(); public static ByteBuffer encode(CharBuffer in) throws CharacterCodingException { return encoder.encode(in); } public static CharBuffer decode(ByteBuffer in) throws CharacterCodingException{ return decoder.decode(in); } } ~~~ # 服務端類NioServer ~~~ package com.nio; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.CharBuffer; import java.nio.channels.*; import java.util.Iterator; public class NioServer { //和操作系統交互的緩存 private ByteBuffer readBuffer; //輪詢器 private Selector selector; public static void main(String[] args) { NioServer server = new NioServer(); server.init(); System.out.println("server started --> 8383"); server.listen(); } private void init() { //分配個緩存 readBuffer = ByteBuffer.allocate(1024); ServerSocketChannel serverSocketChannel; try { //創建一個socket channel; channel是nio中對通信通道的抽象,不分入站出站方向 serverSocketChannel = ServerSocketChannel.open(); //設置通道為非阻塞的方式 serverSocketChannel.configureBlocking(false); //將通道綁定在服務器的ip地址和某個端口上 serverSocketChannel.socket().bind(new InetSocketAddress(8383)); //打開一個多路復用器 selector = Selector.open(); //將上面創建好的socket channel注冊到selector多路復用器上 //對于復用端來說,一定要先注冊一個OP_ACCEPT事件用來響應客戶端的連接請求 //將上述的通道管理器和通道綁定，并為該通道注冊OP_ACCEPT事件 //注冊事件后，當該事件到達時，selector.select()會返回（一個key），如果該事件沒到達selector.select()會一直阻塞 serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); }catch (Exception e) { e.printStackTrace(); } } private void listen() { while(true) { try { //去詢問一次selector選擇器 //這是一個阻塞方法，一直等待直到有數據可讀，返回值是key的數量（可以有多個） selector.select(); //拿到事件的key //如果channel有數據了，將生成的key訪入keys集合中,得到這個keys集合的迭代器 Iterator<SelectionKey> ite = selector.selectedKeys().iterator(); //使用迭代器遍歷集合 while (ite.hasNext()) { //遍歷到一個事件key,得到集合中的一個key實例 SelectionKey key = ite.next(); //確保不重復處理 //拿到當前key實例之后記得在迭代器中將這個元素刪除，非常重要，否則會出錯 ite.remove(); //處理事件 handlekey(key); } }catch (Exception e) { e.printStackTrace(); } } } //處理事件 private void handlekey(SelectionKey key) { SocketChannel channel = null; try { //這個key是可接受連接的嗎 if (key.isAcceptable()) { ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel(); //接收連接請求 channel = serverChannel.accept(); channel.configureBlocking(false); //對讀的事件感興趣 channel.register(selector,SelectionKey.OP_READ); //這個事件是可讀的嗎 }else if(key.isReadable()) { channel = (SocketChannel) key.channel(); //先清空一下buffer,因為要用,以防是老的數據 readBuffer.clear(); /** * 當客戶端channel關閉后,會不斷收到read事件,但沒有消息,即read方法返回-1 * 所以這時服務器端也需要關閉channel,避免無限無效的處理 * 把消息read到readBuffer中 */ int count = channel.read(readBuffer); if (count > 0) { // 一定需要調用flip函數，否則讀取錯誤數據 // 簡單來說，flip操作就是讓讀寫指針、limit指針復位到正確的位置 readBuffer.flip(); /* * 使用CharBuffer配合取出正確的數據; * String question = new String(readBuffer.array());可能會出錯， * 因為前面readBuffer.clear();并未真正清理數據只是重置緩沖區的position, * limit, mark，而readBuffer.array()會返回整個緩沖區的內容。 * decode方法只取readBuffer的position到limit數據。 * 例如，上一次讀取到緩沖區的是"where", clear后position為0，limit為 1024， * 再次讀取“bye"到緩沖區后，position為3，limit不變， * flip后position為0，limit為3，前三個字符被覆蓋了，但"re"還存在緩沖區中，所以 new * String(readBuffer.array()) 返回 "byere", * 而decode(readBuffer)返回"bye"。 */ CharBuffer charBuffer = CharsetHelper.decode(readBuffer); String question = charBuffer.toString(); // 根據客戶端的請求，調用相應的業務方法獲取業務結果 String answer = getAnswer(question); channel.write(CharsetHelper.encode(CharBuffer.wrap(answer))); } else { // 這里關閉channel，因為客戶端已經關閉channel或者異常了 channel.close(); } } }catch (Exception e) { e.printStackTrace(); } } private String getAnswer(String question) { String answer = null; if ("who".equals(question)) { answer = "我是鳳姐\n"; } else if ("what".equals(question)) { answer = "我是來幫你解悶的\n"; } else if ("where".equals(question)) { answer = "我來自外太空\n"; } else if ("hi".equals(question)) { answer = "hello\n"; } else if ("bye".equals(question)) { answer = "88\n"; } else { answer = "請輸入 who，或者what，或者where"; } return answer; } } ~~~ # 客戶端NioClient ~~~ package com.nio; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.CharBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.SocketChannel; import java.util.Iterator; import java.util.Random; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; public class NioClient implements Runnable{ private BlockingQueue<String> words; private Random random; public static void main(String[] args) { // 多個線程發起Socket客戶端連接請求 for(int i=0; i<10; i++){ NioClient c = new NioClient(); c.init(); new Thread(c).start(); } } @Override public void run() { SocketChannel channel = null; Selector selector = null; try { channel = SocketChannel.open(); channel.configureBlocking(false); // 主動請求連接 channel.connect(new InetSocketAddress("localhost", 8383)); selector = Selector.open(); channel.register(selector, SelectionKey.OP_CONNECT); boolean isOver = false; while(! isOver){ selector.select(); Iterator<SelectionKey> ite = selector.selectedKeys().iterator(); while(ite.hasNext()){ SelectionKey key = ite.next(); ite.remove(); if(key.isConnectable()){ if(channel.isConnectionPending()){ if(channel.finishConnect()){ //只有當連接成功后才能注冊OP_READ事件 key.interestOps(SelectionKey.OP_READ); channel.write(CharsetHelper.encode(CharBuffer.wrap(getWord()))); sleep(); } else{ key.cancel(); } } } else if(key.isReadable()){ ByteBuffer byteBuffer = ByteBuffer.allocate(128); channel.read(byteBuffer); byteBuffer.flip(); CharBuffer charBuffer = CharsetHelper.decode(byteBuffer); String answer = charBuffer.toString(); System.out.println(Thread.currentThread().getId() + "---" + answer); String word = getWord(); if(word != null){ channel.write(CharsetHelper.encode(CharBuffer.wrap(word))); } else{ isOver = true; } sleep(); } } } } catch (IOException e) { // TODO } finally{ if(channel != null){ try { channel.close(); } catch (IOException e) { e.printStackTrace(); } } if(selector != null){ try { selector.close(); } catch (IOException e) { e.printStackTrace(); } } } } private void init() { words = new ArrayBlockingQueue<String>(5); try { words.put("hi"); words.put("who"); words.put("what"); words.put("where"); words.put("bye"); } catch (InterruptedException e) { e.printStackTrace(); } random = new Random(); } private String getWord(){ return words.poll(); } private void sleep() { try { TimeUnit.SECONDS.sleep(random.nextInt(3)); } catch (InterruptedException e) { e.printStackTrace(); } } private void sleep(long l) { try { TimeUnit.SECONDS.sleep(l); } catch (InterruptedException e) { e.printStackTrace(); } } } ~~~