1、阻塞
- 阻塞模式下，相關方法都會導致執行緒暫停
- ServerSocketChannel.accept 會在沒有連接建立時讓執行緒暫停
- SocketChannel.read 會在通道中沒有資料可讀時讓執行緒暫停
- 阻塞的表現其實就是執行緒暫停了，暫停期間不會占用 cpu，但執行緒相當于閑置
- 單執行緒下，阻塞方法之間相互影響，幾乎不能正常作業，需要多執行緒支持
- 但多執行緒下，有新的問題，體現在以下方面
- 32 位 jvm 一個執行緒 320k，64 位 jvm 一個執行緒 1024k，如果連接數過多，必然導致 OOM，并且執行緒太多，反而會因為頻繁背景關系切換導致性能降低
- 可以采用執行緒池技術來減少執行緒數和執行緒背景關系切換，但治標不治本，如果有很多連接建立，但長時間 inactive，會阻塞執行緒池中所有執行緒，因此不適合長連接，只適合短連接
服務端代碼
public class Server {
public static void main(String[] args) {
// 創建緩沖區
ByteBuffer buffer = ByteBuffer.allocate(16);
// 獲得服務器通道
try(ServerSocketChannel server = ServerSocketChannel.open()) {
// 為服務器通道系結埠
server.bind(new InetSocketAddress(8080));
// 用戶存放連接的集合
ArrayList<SocketChannel> channels = new ArrayList<>();
// 回圈接收連接
while (true) {
System.out.println("before connecting...");
// 沒有連接時，會阻塞執行緒
SocketChannel socketChannel = server.accept();
System.out.println("after connecting...");
channels.add(socketChannel);
// 回圈遍歷集合中的連接
for(SocketChannel channel : channels) {
System.out.println("before reading");
// 處理通道中的資料
// 當通道中沒有資料可讀時，會阻塞執行緒
channel.read(buffer);
buffer.flip();
ByteBufferUtil.debugRead(buffer);
buffer.clear();
System.out.println("after reading");
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
客戶端代碼
public class Client {
public static void main(String[] args) {
try (SocketChannel socketChannel = SocketChannel.open()) {
// 建立連接
socketChannel.connect(new InetSocketAddress("localhost", 8080));
System.out.println("waiting...");
} catch (IOException e) {
e.printStackTrace();
}
}
}
運行結果
- 客戶端 - 服務器建立連接前：服務器端因 accept 阻塞
- 客戶端 - 服務器建立連接后，客戶端發送訊息前：服務器端因通道為空被阻塞
- 客戶端發送資料后，服務器處理通道中的資料，再次進入回圈時，再次被 accept 阻塞
- 之前的客戶端再次發送訊息，服務器端因為被 accept 阻塞，無法處理之前客戶端發送到通道中的資訊
2、非阻塞
- 可以通過 ServerSocketChannel 的 configureBlocking (false) 方法將 獲得連接設定為非阻塞的，此時若沒有連接，accept 會回傳 null
- 可以通過 SocketChannel 的 configureBlocking (false) 方法將從通道中 讀取資料設定為非阻塞的，若此時通道中沒有資料可讀，read 會回傳 - 1
服務器代碼如下
public class Server {
public static void main(String[] args) {
// 創建緩沖區
ByteBuffer buffer = ByteBuffer.allocate(16);
// 獲得服務器通道
try(ServerSocketChannel server = ServerSocketChannel.open()) {
// 設定為非阻塞模式，沒有連接時回傳null，不會阻塞執行緒
server.configureBlocking(false);
// 為服務器通道系結埠
server.bind(new InetSocketAddress(8080));
// 用戶存放連接的集合
ArrayList<SocketChannel> channels = new ArrayList<>();
// 回圈接收連接
while (true) {
SocketChannel socketChannel = server.accept();
// 通道不為空時才將連接放入到集合中
if (socketChannel != null) {
System.out.println("after connecting...");
channels.add(socketChannel);
}
// 回圈遍歷集合中的連接
for(SocketChannel channel : channels) {
// 處理通道中的資料
// 設定為非阻塞模式，若通道中沒有資料，會回傳0，不會阻塞執行緒
channel.configureBlocking(false);
int read = channel.read(buffer);
if(read > 0) {
buffer.flip();
ByteBufferUtil.debugRead(buffer);
buffer.clear();
System.out.println("after reading");
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
這樣寫存在一個問題，因為設定為了非阻塞，會一直執行 while (true) 中的代碼，CPU 一直處于忙碌狀態，會使得性能變低，所以實際情況中不使用這種方法處理請求
3、Selector
多路復用
單執行緒可以配合 Selector 完成對多個 Channel 可讀寫事件的監控，這稱之為多路復用
- 多路復用僅針對網路 IO，普通檔案 IO 無法利用多路復用
- 如果不用 Selector 的非阻塞模式，執行緒大部分時間都在做無用功，而 Selector 能夠保證
- 有可連接事件時才去連接
- 有可讀事件才去讀取
- 有可寫事件才去寫入
- 限于網路傳輸能力，Channel 未必時時可寫，一旦 Channel 可寫，會觸發 Selector 的可寫事件
4、使用及 Accpet 事件
要使用 Selector 實作多路復用，服務端代碼如下改進
public class SelectServer {
public static void main(String[] args) {
ByteBuffer buffer = ByteBuffer.allocate(16);
// 獲得服務器通道
try(ServerSocketChannel server = ServerSocketChannel.open()) {
server.bind(new InetSocketAddress(8080));
// 創建選擇器
Selector selector = Selector.open();
// 通道必須設定為非阻塞模式
server.configureBlocking(false);
// 將通道注冊到選擇器中，并設定感興趣的事件
server.register(selector, SelectionKey.OP_ACCEPT);
while (true) {
// 若沒有事件就緒，執行緒會被阻塞，反之不會被阻塞，從而避免了CPU空轉
// 回傳值為就緒的事件個數
int ready = selector.select();
System.out.println("selector ready counts : " + ready);
// 獲取所有事件
Set<SelectionKey> selectionKeys = selector.selectedKeys();
// 使用迭代器遍歷事件
Iterator<SelectionKey> iterator = selectionKeys.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
// 判斷key的型別
if(key.isAcceptable()) {
// 獲得key對應的channel
ServerSocketChannel channel = (ServerSocketChannel) key.channel();
System.out.println("before accepting...");
// 獲取連接并處理，而且是必須處理，否則需要取消
SocketChannel socketChannel = channel.accept();
System.out.println("after accepting...");
// 處理完畢后移除
iterator.remove();
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
步驟決議
- 獲得選擇器 Selector
Selector selector = Selector.open();
- 將通道設定為非阻塞模式，并注冊到選擇器中，并設定感興趣的事件
- channel 必須作業在非阻塞模式
- FileChannel 沒有非阻塞模式，因此不能配合 selector 一起使用
- 系結的事件型別可以有
- connect - 客戶端連接成功時觸發
- accept - 服務器端成功接受連接時觸發
- read - 資料可讀入時觸發，有因為接收能力弱，資料暫不能讀入的情況
- write - 資料可寫出時觸發，有因為發送能力弱，資料暫不能寫出的情況
// 通道必須設定為非阻塞模式
server.configureBlocking(false);
// 將通道注冊到選擇器中，并設定感興趣的實踐
server.register(selector, SelectionKey.OP_ACCEPT);
-
通過 Selector 監聽事件，并獲得就緒的通道個數，若沒有通道就緒，執行緒會被阻塞
-
阻塞直到系結事件發生
int count = selector.select(); -
阻塞直到系結事件發生，或是超時（時間單位為 ms）
int count = selector.select(long timeout); -
不會阻塞，也就是不管有沒有事件，立刻回傳，自己根據回傳值檢查是否有事件
int count = selector.selectNow();
-
-
獲取就緒事件并得到對應的通道，然后進行處理
// 獲取所有事件
Set<SelectionKey> selectionKeys = selector.selectedKeys();
// 使用迭代器遍歷事件
Iterator<SelectionKey> iterator = selectionKeys.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
// 判斷key的型別，此處為Accept型別
if(key.isAcceptable()) {
// 獲得key對應的channel
ServerSocketChannel channel = (ServerSocketChannel) key.channel();
// 獲取連接并處理，而且是必須處理，否則需要取消
SocketChannel socketChannel = channel.accept();
// 處理完畢后移除
iterator.remove();
}
}
事件發生后能否不處理
事件發生后，要么處理，要么取消（cancel），不能什么都不做，否則下次該事件仍會觸發，這是因為 nio 底層使用的是水平觸發
5、Read 事件
- 在 Accept 事件中，若有客戶端與服務器端建立了連接，需要將其對應的 SocketChannel 設定為非阻塞，并注冊到選擇其中
添加 Read 事件，觸發后進行讀取操作 - 添加 Read 事件，觸發后進行讀取操作
public class SelectServer {
public static void main(String[] args) {
ByteBuffer buffer = ByteBuffer.allocate(16);
// 獲得服務器通道
try(ServerSocketChannel server = ServerSocketChannel.open()) {
server.bind(new InetSocketAddress(8080));
// 創建選擇器
Selector selector = Selector.open();
// 通道必須設定為非阻塞模式
server.configureBlocking(false);
// 將通道注冊到選擇器中，并設定感興趣的實踐
server.register(selector, SelectionKey.OP_ACCEPT);
// 為serverKey設定感興趣的事件
while (true) {
// 若沒有事件就緒，執行緒會被阻塞，反之不會被阻塞，從而避免了CPU空轉
// 回傳值為就緒的事件個數
int ready = selector.select();
System.out.println("selector ready counts : " + ready);
// 獲取所有事件
Set<SelectionKey> selectionKeys = selector.selectedKeys();
// 使用迭代器遍歷事件
Iterator<SelectionKey> iterator = selectionKeys.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
// 判斷key的型別
if(key.isAcceptable()) {
// 獲得key對應的channel
ServerSocketChannel channel = (ServerSocketChannel) key.channel();
System.out.println("before accepting...");
// 獲取連接
SocketChannel socketChannel = channel.accept();
System.out.println("after accepting...");
// 設定為非阻塞模式，同時將連接的通道也注冊到選擇其中
socketChannel.configureBlocking(false);
socketChannel.register(selector, SelectionKey.OP_READ);
// 處理完畢后移除
iterator.remove();
} else if (key.isReadable()) {
SocketChannel channel = (SocketChannel) key.channel();
System.out.println("before reading...");
channel.read(buffer);
System.out.println("after reading...");
buffer.flip();
ByteBufferUtil.debugRead(buffer);
buffer.clear();
// 處理完畢后移除
iterator.remove();
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
洗掉事件
當處理完一個事件后，一定要呼叫迭代器的 remove 方法移除對應事件，否則會出現錯誤，原因如下
以我們上面的 Read 事件 的代碼為例
-
當呼叫了 server.register (selector, SelectionKey.OP_ACCEPT) 后，Selector 中維護了一個集合，用于存放 SelectionKey 以及其對應的通道
// WindowsSelectorImpl 中的 SelectionKeyImpl陣列 private SelectionKeyImpl[] channelArray = new SelectionKeyImpl[8];public class SelectionKeyImpl extends AbstractSelectionKey { // Key對應的通道 final SelChImpl channel; ... }
- 當選擇器中的通道對應的事件發生后，selecionKey 會被放到另一個集合中，但是 selecionKey 不會自動移除，所以需要我們在處理完一個事件后，通過迭代器手動移除其中的 selecionKey，否則會導致已被處理過的事件再次被處理，就會引發錯誤
斷開處理
當客戶端與服務器之間的連接斷開時，會給服務器端發送一個讀事件，對例外斷開和正常斷開需要加以不同的方式進行處理
-
正常斷開
-
正常斷開時，服務器端的 channel.read (buffer) 方法的回傳值為 - 1，所以當結束到回傳值為 - 1 時，需要呼叫 key 的 cancel 方法取消此事件，并在取消后移除該事件
int read = channel.read(buffer); // 斷開連接時，客戶端會向服務器發送一個寫事件，此時read的回傳值為-1 if(read == -1) { // 取消該事件的處理 key.cancel(); channel.close(); } else { ... } // 取消或者處理，都需要移除key iterator.remove();
-
-
例外斷開
- 例外斷開時，會拋出 IOException 例外， 在 try-catch 的 catch 塊中捕獲例外并呼叫 key 的 cancel 方法即可
訊息邊界
不處理訊息邊界存在的問題
將緩沖區的大小設定為 4 個位元組，發送 2 個漢字（你好），通過 decode 解碼并列印時，會出現亂碼
ByteBuffer buffer = ByteBuffer.allocate(4);
// 解碼并列印
System.out.println(StandardCharsets.UTF_8.decode(buffer));
你?
??
這是因為 UTF-8 字符集下，1 個漢字占用 3 個位元組，此時緩沖區大小為 4 個位元組，一次讀時間無法處理完通道中的所有資料，所以一共會觸發兩次讀事件，這就導致 你好 的 好 字被拆分為了前半部分和后半部分發送，解碼時就會出現問題
處理訊息邊界
傳輸的文本可能有以下三種情況
- 文本大于緩沖區大小
- 此時需要將緩沖區進行擴容
- 發生半包現象
- 發生粘包現象
解決思路大致有以下三種
-
固定訊息長度，資料包大小一樣，服務器按預定長度讀取，當發送的資料較少時，需要將資料進行填充，直到長度與訊息規定長度一致，缺點是浪費帶寬
-
另一種思路是按分隔符拆分，缺點是效率低，需要一個一個字符地去匹配分隔符
-
TLV 格式，即 Type 型別、Length 長度、Value 資料
（也就是在訊息開頭用一些空間存放后面資料的長度），如 HTTP 請求頭中的 Content-Type 與 Content-Length
，型別和長度已知的情況下，就可以方便獲取訊息大小，分配合適的 buffer，缺點是 buffer 需要提前分配，如果內容過大，則影響 server 吞吐量
- Http 1.1 是 TLV 格式
-
Http 2.0 是 LTV 格式
下文的訊息邊界處理方式為第二種：按分隔符拆分
附件與擴容
Channel 的 register 方法還有第三個引數：附件，可以向其中放入一個 Object 型別的物件，該物件會與登記的 Channel 以及其對應的 SelectionKey 系結，可以從 SelectionKey 獲取到對應通道的附件
public final SelectionKey register(Selector sel, int ops, Object att)
可通過 SelectionKey 的 attachment () 方法獲得附件
ByteBuffer buffer = (ByteBuffer) key.attachment();
我們需要在 Accept 事件發生后，將通道注冊到 Selector 中時，對每個通道添加一個 ByteBuffer 附件，讓每個通道發生讀事件時都使用自己的通道，避免與其他通道發生沖突而導致問題
// 設定為非阻塞模式，同時將連接的通道也注冊到選擇其中，同時設定附件
socketChannel.configureBlocking(false);
ByteBuffer buffer = ByteBuffer.allocate(16);
// 添加通道對應的Buffer附件
socketChannel.register(selector, SelectionKey.OP_READ, buffer);
當 Channel 中的資料大于緩沖區時，需要對緩沖區進行擴容操作，此代碼中的擴容的判定方法： Channel 呼叫 compact 方法后，的 position 與 limit 相等，說明緩沖區中的資料并未被讀取（容量太小），此時創建新的緩沖區，其大小擴大為兩倍，同時還要將舊緩沖區中的資料拷貝到新的緩沖區中，同時呼叫 SelectionKey 的 attach 方法將新的緩沖區作為新的附件放入 SelectionKey 中
// 如果緩沖區太小，就進行擴容
if (buffer.position() == buffer.limit()) {
ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity()*2);
// 將舊buffer中的內容放入新的buffer中
ewBuffer.put(buffer);
// 將新buffer作為附件放到key中
key.attach(newBuffer);
}
改造后的服務器代碼如下
public class SelectServer {
public static void main(String[] args) {
// 獲得服務器通道
try(ServerSocketChannel server = ServerSocketChannel.open()) {
server.bind(new InetSocketAddress(8080));
// 創建選擇器
Selector selector = Selector.open();
// 通道必須設定為非阻塞模式
server.configureBlocking(false);
// 將通道注冊到選擇器中，并設定感興趣的事件
server.register(selector, SelectionKey.OP_ACCEPT);
// 為serverKey設定感興趣的事件
while (true) {
// 若沒有事件就緒，執行緒會被阻塞，反之不會被阻塞，從而避免了CPU空轉
// 回傳值為就緒的事件個數
int ready = selector.select();
System.out.println("selector ready counts : " + ready);
// 獲取所有事件
Set<SelectionKey> selectionKeys = selector.selectedKeys();
// 使用迭代器遍歷事件
Iterator<SelectionKey> iterator = selectionKeys.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
// 判斷key的型別
if(key.isAcceptable()) {
// 獲得key對應的channel
ServerSocketChannel channel = (ServerSocketChannel) key.channel();
System.out.println("before accepting...");
// 獲取連接
SocketChannel socketChannel = channel.accept();
System.out.println("after accepting...");
// 設定為非阻塞模式，同時將連接的通道也注冊到選擇其中，同時設定附件
socketChannel.configureBlocking(false);
ByteBuffer buffer = ByteBuffer.allocate(16);
socketChannel.register(selector, SelectionKey.OP_READ, buffer);
} else if (key.isReadable()) {
SocketChannel channel = (SocketChannel) key.channel();
System.out.println("before reading...");
// 通過key獲得附件（buffer）
ByteBuffer buffer = (ByteBuffer) key.attachment();
int read = channel.read(buffer);
if(read == -1) {
key.cancel();
channel.close();
} else {
// 通過分隔符來分隔buffer中的資料
split(buffer);
// 如果緩沖區太小，就進行擴容
if (buffer.position() == buffer.limit()) {
ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity()*2);
// 將舊buffer中的內容放入新的buffer中
buffer.flip();
newBuffer.put(buffer);
// 將新buffer放到key中作為附件
key.attach(newBuffer);
}
}
System.out.println("after reading...");
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
private static void split(ByteBuffer buffer) {
buffer.flip();
for(int i = 0; i < buffer.limit(); i++) {
// 遍歷尋找分隔符
// get(i)不會移動position
if (buffer.get(i) == '\n') {
// 緩沖區長度
int length = i+1-buffer.position();
ByteBuffer target = ByteBuffer.allocate(length);
// 將前面的內容寫入target緩沖區
for(int j = 0; j < length; j++) {
// 將buffer中的資料寫入target中
target.put(buffer.get());
}
// 列印結果
ByteBufferUtil.debugAll(target);
}
}
// 切換為寫模式，但是緩沖區可能未讀完，這里需要使用compact
buffer.compact();
}
}
ByteBuffer 的大小分配
- 每個 channel 都需要記錄可能被切分的訊息，因為 ByteBuffer 不能被多個 channel 共同使用，因此需要為每個 channel 維護一個獨立的 ByteBuffer
- ByteBuffer 不能太大，比如一個 ByteBuffer 1Mb 的話，要支持百萬連接就要 1Tb 記憶體，因此需要設計大小可變的 ByteBuffer
- 分配思路可以參考
- 一種思路是首先分配一個較小的 buffer，例如 4k，如果發現資料不夠，再分配 8k 的 buffer，將 4k buffer 內容拷貝至 8k buffer，優點是訊息連續容易處理，缺點是資料拷貝耗費性能
- 另一種思路是用多個陣列組成 buffer，一個陣列不夠，把多出來的內容寫入新的陣列，與前面的區別是訊息存盤不連續決議復雜，優點是避免了拷貝引起的性能損耗
6、Write 事件
服務器通過 Buffer 向通道中寫入資料時，可能因為通道容量小于 Buffer 中的資料大小，導致無法一次性將 Buffer 中的資料全部寫入到 Channel 中，這時便需要分多次寫入，具體步驟如下
-
執行一次寫操作，向將 buffer 中的內容寫入到 SocketChannel 中，然后判斷 Buffer 中是否還有資料
-
若 Buffer 中還有資料，則需要將 SockerChannel 注冊到 Seletor 中，并關注寫事件，同時將未寫完的 Buffer 作為附件一起放入到 SelectionKey 中
int write = socket.write(buffer);
// 通道中可能無法放入緩沖區中的所有資料
if (buffer.hasRemaining()) {
// 注冊到Selector中，關注可寫事件，并將buffer添加到key的附件中
socket.configureBlocking(false);
socket.register(selector, SelectionKey.OP_WRITE, buffer);
}
- 添加寫事件的相關操作 key.isWritable()，對 Buffer 再次進行寫操作
- 每次寫后需要判斷 Buffer 中是否還有資料（是否寫完），若寫完，需要移除 SelecionKey 中的 Buffer 附件，避免其占用過多記憶體，同時還需移除對寫事件的關注
SocketChannel socket = (SocketChannel) key.channel();
// 獲得buffer
ByteBuffer buffer = (ByteBuffer) key.attachment();
// 執行寫操作
int write = socket.write(buffer);
System.out.println(write);
// 如果已經完成了寫操作，需要移除key中的附件，同時不再對寫事件感興趣
if (!buffer.hasRemaining()) {
key.attach(null);
key.interestOps(0);
}
整體代碼如下
public class WriteServer {
public static void main(String[] args) {
try(ServerSocketChannel server = ServerSocketChannel.open()) {
server.bind(new InetSocketAddress(8080));
server.configureBlocking(false);
Selector selector = Selector.open();
server.register(selector, SelectionKey.OP_ACCEPT);
while (true) {
selector.select();
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectionKeys.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
// 處理后就移除事件
iterator.remove();
if (key.isAcceptable()) {
// 獲得客戶端的通道
SocketChannel socket = server.accept();
// 寫入資料
StringBuilder builder = new StringBuilder();
for(int i = 0; i < 500000000; i++) {
builder.append("a");
}
ByteBuffer buffer = StandardCharsets.UTF_8.encode(builder.toString());
// 先執行一次Buffer->Channel的寫入，如果未寫完，就添加一個可寫事件
int write = socket.write(buffer);
System.out.println(write);
// 通道中可能無法放入緩沖區中的所有資料
if (buffer.hasRemaining()) {
// 注冊到Selector中，關注可寫事件，并將buffer添加到key的附件中
socket.configureBlocking(false);
socket.register(selector, SelectionKey.OP_WRITE, buffer);
}
} else if (key.isWritable()) {
SocketChannel socket = (SocketChannel) key.channel();
// 獲得buffer
ByteBuffer buffer = (ByteBuffer) key.attachment();
// 執行寫操作
int write = socket.write(buffer);
System.out.println(write);
// 如果已經完成了寫操作，需要移除key中的附件，同時不再對寫事件感興趣
if (!buffer.hasRemaining()) {
key.attach(null);
key.interestOps(0);
}
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
7、優化
多執行緒優化
充分利用多核 CPU，分兩組選擇器
- 單執行緒配一個選擇器（Boss），專門處理 accept 事件
- 創建 cpu 核心數的執行緒（Worker），每個執行緒配一個選擇器，輪流處理 read 事件
實作思路
-
創建一個負責處理 Accept 事件的 Boss 執行緒，與多個負責處理 Read 事件的 Worker 執行緒
-
Boss 執行緒執行的操作
-
接受并處理 Accepet 事件，當 Accept 事件發生后，呼叫 Worker 的 register (SocketChannel socket) 方法，讓 Worker 去處理 Read 事件，其中需要根據標識 robin 去判斷將任務分配給哪個 Worker
// 創建固定數量的Worker Worker[] workers = new Worker[4]; // 用于負載均衡的原子整數 AtomicInteger robin = new AtomicInteger(0); // 負載均衡，輪詢分配Worker workers[robin.getAndIncrement()% workers.length].register(socket); -
register (SocketChannel socket) 方法會通過同步佇列完成 Boss 執行緒與 Worker 執行緒之間的通信，讓 SocketChannel 的注冊任務被 Worker 執行緒執行，添加任務后需要呼叫 selector.wakeup () 來喚醒被阻塞的 Selector
public void register(final SocketChannel socket) throws IOException { // 只啟動一次 if (!started) { // 初始化操作 } // 向同步佇列中添加SocketChannel的注冊事件 // 在Worker執行緒中執行注冊事件 queue.add(new Runnable() { @Override public void run() { try { socket.register(selector, SelectionKey.OP_READ); } catch (IOException e) { e.printStackTrace(); } } }); // 喚醒被阻塞的Selector // select類似LockSupport中的park，wakeup的原理類似LockSupport中的unpark selector.wakeup(); }
-
-
Worker 執行緒執行的操作
- 從同步佇列中獲取注冊任務，并處理 Read 事件
實作代碼
public class ThreadsServer {
public static void main(String[] args) {
try (ServerSocketChannel server = ServerSocketChannel.open()) {
// 當前執行緒為Boss執行緒
Thread.currentThread().setName("Boss");
server.bind(new InetSocketAddress(8080));
// 負責輪詢Accept事件的Selector
Selector boss = Selector.open();
server.configureBlocking(false);
server.register(boss, SelectionKey.OP_ACCEPT);
// 創建固定數量的Worker
Worker[] workers = new Worker[4];
// 用于負載均衡的原子整數
AtomicInteger robin = new AtomicInteger(0);
for(int i = 0; i < workers.length; i++) {
workers[i] = new Worker("worker-"+i);
}
while (true) {
boss.select();
Set<SelectionKey> selectionKeys = boss.selectedKeys();
Iterator<SelectionKey> iterator = selectionKeys.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
// BossSelector負責Accept事件
if (key.isAcceptable()) {
// 建立連接
SocketChannel socket = server.accept();
System.out.println("connected... ");
socket.configureBlocking(false);
// socket注冊到Worker的Selector中
System.out.println("before read...");
// 負載均衡，輪詢分配Worker
workers[robin.getAndIncrement()% workers.length].register(socket);
System.out.println("after read...");
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
static class Worker implements Runnable {
private Thread thread;
private volatile Selector selector;
private String name;
private volatile boolean started = false;
/**
* 同步佇列，用于Boss執行緒與Worker執行緒之間的通信
*/
private ConcurrentLinkedQueue<Runnable> queue;
public Worker(String name) {
this.name = name;
}
public void register(final SocketChannel socket) throws IOException {
// 只啟動一次
if (!started) {
thread = new Thread(this, name);
selector = Selector.open();
queue = new ConcurrentLinkedQueue<>();
thread.start();
started = true;
}
// 向同步佇列中添加SocketChannel的注冊事件
// 在Worker執行緒中執行注冊事件
queue.add(new Runnable() {
@Override
public void run() {
try {
socket.register(selector, SelectionKey.OP_READ);
} catch (IOException e) {
e.printStackTrace();
}
}
});
// 喚醒被阻塞的Selector
// select類似LockSupport中的park，wakeup的原理類似LockSupport中的unpark
selector.wakeup();
}
@Override
public void run() {
while (true) {
try {
selector.select();
// 通過同步佇列獲得任務并運行
Runnable task = queue.poll();
if (task != null) {
// 獲得任務，執行注冊操作
task.run();
}
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectionKeys.iterator();
while(iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
// Worker只負責Read事件
if (key.isReadable()) {
// 簡化處理，省略細節
SocketChannel socket = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(16);
socket.read(buffer);
buffer.flip();
ByteBufferUtil.debugAll(buffer);
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
本文由
傳智教育博學谷教研團隊發布，如果本文對您有幫助，歡迎
關注和點贊；如果您有任何建議也可留言評論或私信，您的支持是我堅持創作的動力，轉載請注明出處！
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/540916.html
標籤：其他