Java網路程式設計——第十章 非阻塞I/O
阿新 • • 發佈：2019-01-10
使用非阻塞IO方式實現chargen收發協議
客戶端
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.SocketChannel;
import java.nio.channels.WritableByteChannel;
public class ChargenClient {
public static final int DEFAULT_PORT = 19;
public static final String HOST= "localhost";
public static void main(String[] args) {
try {
SocketAddress address = new InetSocketAddress(HOST, DEFAULT_PORT);
SocketChannel client = SocketChannel.open(address);
// 注意client的通道是阻塞模式，修改為非阻塞模式，則通過
//client.configureBlocking(false);
ByteBuffer buffer = ByteBuffer.allocate(74);
// 使用通道的方式將緩衝區的資料寫入標準輸出
// Channels.newChannel是構建一個寫入目標流的通道
WritableByteChannel out = Channels.newChannel(System.out);
while (client.read(buffer) != -1) {
// 迴繞緩衝區，該方法將緩衝區準備為資料傳出狀態，輸出通道從資料開頭而不是末尾開始
// 迴繞緩衝區保持緩衝區資料不變，只是準備寫入而不是讀取
buffer.flip();
out.write(buffer);
// 要重用現有緩衝區，再次讀取之前需要清空緩衝區；該方法不會改變緩衝區資料
// 只是簡單重置索引
buffer.clear();
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
服務端
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
public class ChargenServer {
public static final int DEFAULT_PORT = 19;
// 伺服器端單執行緒處理多客戶端連線
public static void main(String[] args) {
int port = DEFAULT_PORT;
System.out.println("listening for connections on port: " + port);
// 可列印ASCII共95個，由於每列印一行需要左移一個字元，這裡生成92*2個只是為了列印方便
byte[] rotation = new byte[95 * 2];
for (byte i = ' '; i <= '~'; i++) {
rotation[i - ' '] = i;
rotation[i + 95 - ' '] = i;
}
ServerSocketChannel serverChannel;
// Selector只接受非阻塞通道
Selector selector;
try {
serverChannel = ServerSocketChannel.open();
// 要繫結埠，需要先用socket方法獲取ServerSocket的對等端peer
// 然後使用bind繫結埠
ServerSocket ss = serverChannel.socket();
InetSocketAddress address = new InetSocketAddress(DEFAULT_PORT);
ss.bind(address);
// 配置serverChannel為非阻塞模式，如果沒有連線則accpet方法立即返回null， 如果不設定，accept方法將一直阻塞直到有連線進入
serverChannel.configureBlocking(false);
// 使用選擇器迭代處理準備好的連線
selector = Selector.open();
// 向選擇器註冊對此通道的監視，對於chargen，只關心伺服器Socket是否準備好接收一個新連線
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
} catch (IOException e) {
e.printStackTrace();
return;
}
while (true) {
try {
// 檢查是否有可操作的通道準備好接受IO操作
selector.select();
} catch (IOException e) {
e.printStackTrace();
break;
}
// 獲取就緒通道的的SelectionKey的集合
Set<SelectionKey> readyKey = selector.selectedKeys();
// 迭代處理所有的key
Iterator<SelectionKey> iterator = readyKey.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
// 已經處理的key需要從集合中刪除，標記該鍵已經處理
// 若下次該key還再次準備好，還將繼續出現在Set中
iterator.remove();
try {
// 測試該key是否已經準備好接收一個新的socket連線，即此時就緒的是伺服器通道，接收一個新的Socket通道，將其新增到選擇器
if (key.isAcceptable()) {
// 獲取該key建立的channel
ServerSocketChannel server = (ServerSocketChannel) key.channel();
// 接受一個到此server通道的連線
SocketChannel client = server.accept();
System.out.println("accepted from " + client);
// 配置client的通道為非阻塞模式
client.configureBlocking(false);
SelectionKey key2 = client.register(selector, SelectionKey.OP_WRITE);
ByteBuffer buffer = ByteBuffer.allocate(74);
buffer.put(rotation, 0, 72);
buffer.put((byte) '\r');
buffer.put((byte) '\n');
// 迴繞緩衝區
buffer.flip();
key2.attach(buffer);
} else if (key.isWritable()) {// 測試該key是否準備好寫操作，即此時就緒的是Socket通道，向緩衝區寫入資料
SocketChannel client = (SocketChannel) key.channel();
ByteBuffer buffer = (ByteBuffer) key.attachment();
if (!buffer.hasRemaining()) {
// 用下一行重新填充緩衝區
buffer.rewind();
// 得到上一次首字元
int first = buffer.get();
// 準備改變緩衝區資料
buffer.rewind();
// 尋找rotation中新的首字元位置
int position = first - ' ' + 1;
// 將資料從rotation複製到緩衝區
buffer.put(rotation, position, 72);
// 在緩衝區末尾存放一個行分隔符
buffer.put((byte) '\r');
buffer.put((byte) '\n');
// 迴繞緩衝區，準備寫入
buffer.flip();
}
client.write(buffer);
}
// 沒有就緒通道，選擇器等待就緒通道
} catch (IOException e) {
key.cancel();
try {
key.channel().close();
} catch (IOException ex) {
}
}
}
}
}
}
緩衝區
在NIO中，所有IO都必須緩衝；同時新模型中不再直接操作輸入流、輸出流寫，而是從緩衝區讀寫資料，注意緩衝區底層可能是位元組陣列，也可能是其他實現
流與通道的區別
1、流基於位元組，按順序一個位元組一個位元組的傳送，處於效能考慮也可以傳送位元組陣列；通道基於塊，傳送緩衝區中的資料塊，在讀寫通道位元組之前，資料必須已經儲存在緩衝區中，且一次讀寫一個緩衝區的資料
2、通道和緩衝區支援同一物件的讀寫
緩衝區型別
Java基本資料型別中，除了boolean外，所有基本型別都有對應的緩衝區，如float對應FloatBuffer，其超類是Buffer，Buffer中的方法為通用方法
緩衝區4個關鍵引數，無論緩衝區型別，都有相同的方法來獲取和設定這些值
1、位置position，即緩衝區中將讀取或者寫入的下一個位置，該值從0開始，最大值為緩衝區大小
public final int position()
public final Buffer position(int newPosition)
2、容量capacity，緩衝區儲存資料最大數目，容量在建立時設定，且設定後不可改變
public final int capacity()
3、限度limit，緩衝區中可訪問資料的末尾位置，只要不改變限度，就無法讀寫超過這個位置的資料，即使緩衝區容量更大
public final int limit()
public final Buffer limit(int newLimit)
4、標記mark，緩衝區中客戶端指定的標記，使得緩衝區能夠記住一個位置並在之後將其返回，緩衝區標記在mark()呼叫之前是未定義的，呼叫時則設定mark=position，reset()方法一樣，設定mark=position，如果標記未定義時呼叫reset()，則丟擲InvalidMaekException，一些緩衝區函式會拋棄已經設定的標記，如rewind()、clear()、flip()總是拋棄標記，如果設定的值比當前的標記小，則拋棄標記
public final Buffer mark()
public final Buffer reset()
注意： 0<= mark<=position<=limit<= capacity
公共buffer中的其他方法
public final Bufferclear()，位置設定為0，限度設定為容量，“清空”緩衝區，注意clear()並沒有刪除資料，只是重置了位置引數，這些資料依然可以使用絕對的get方法或者再改變限度和位置進行讀取
public final Buffer rewind()，位置設定為0，但不改變限度，即允許重新讀取緩衝區
public final Buffer flip()，將限度設定為當前位置，位置設定為0，在希望排空剛剛填充的緩衝區時可以使用這個方法
public final int remaining() 返回position和limit之間的元素數目
public fianl boolean hasRemaining()，如果remaining()返回的數大於0，則返回true
建立緩衝區
緩衝區類的層次是基於繼承的，而非基於多型，在使用緩衝區前，首先需要確定的是需要的緩衝區型別，然後使用對應的工程方法建立特定的實現類；空的緩衝區使用allocate建立，常用於輸入；預填充緩衝區由wrap建立，常用於輸出
allocate方法只用於建立固定容量的空緩衝區，底層基於陣列實現，可以通過array()和arrayOffset()訪問該底層陣列（後備陣列），但需要注意的是，這實際上暴露了buffer的私有資料，對該陣列的任何更改都將反映到緩衝區中，因此在獲取陣列後，就不應該在寫緩衝區
allocateDirect方法（僅用於ByteBuffer），底層使用直接記憶體在網絡卡、核心記憶體等位置建立緩衝區，底層不是陣列，且使用array()將丟擲UnSupportedOperationException；需要注意的是，直接分配應用於持續時間較短的緩衝區，其實現細節依賴JVM，不是必須不應使用
wrap方法，用於包裝已有的陣列形成緩衝區，此時緩衝區包含了陣列（充當了後備陣列）的引用，類似的對此後備陣列的改變會反映到緩衝區，因此在陣列操作完成前，不應使用wrap
填充put與排空get
對於使用相對的put和get，都會影響position，put會使position增大，超出容量丟擲BufferOverflowException，get會使position減小，可能丟擲BufferUnderflowException
對於使用絕對的put和get，即使用index，此時不會影響position，超出範圍則丟擲IndexOutOfBoundsException
批量方法，即以資料塊的方式填充或排空緩衝區，該方法會影響position，對應的put和get都是從當前position填充或排空資料，put在緩衝區沒有足夠空間容納陣列內容則丟擲BufferOverflowException，get在緩衝區沒有足夠資料填充陣列則丟擲BufferUnderflowException
批量方法(一下為ByteBuffer的bulk method，對於IntBuffer，則引數型別為int[]等)
public ByteBuffer get(byte[] det, int offset, int length)
public ByteBuffer get(byte[] det)
public ByteBuffer put(byte[] det, int offset, int length)
public ByteBuffer put(byte[] det)
資料轉換（僅用於ByteBuffer