4.網(wǎng)絡(luò)編程

4.1.非阻塞 VS 阻塞

在網(wǎng)絡(luò)編程中，**阻塞（Blocking）和非阻塞（Non-blocking）**是兩種不同的編程模型，描述了程序在進(jìn)行網(wǎng)絡(luò)通信時(shí)的行為方式。

1. 阻塞（Blocking）：
   • 在阻塞模型中，當(dāng)程序發(fā)起一個(gè)網(wǎng)絡(luò)請求時(shí)，它會一直等待直到操作完成或者發(fā)生錯(cuò)誤。
   • 在網(wǎng)絡(luò)通信過程中，如果數(shù)據(jù)沒有到達(dá)，或者連接還沒有建立，程序會被掛起，直到數(shù)據(jù)到達(dá)或者連接建立完成。
   • 在阻塞模型中，通常一個(gè)線程只處理一個(gè)連接，因此需要為每個(gè)連接創(chuàng)建一個(gè)新的線程，這會增加系統(tǒng)開銷，尤其在高并發(fā)環(huán)境下，可能導(dǎo)致資源耗盡和性能下降。
2. 非阻塞（Non-blocking）：
   • 在非阻塞模型中，程序可以在發(fā)起網(wǎng)絡(luò)請求后立即返回，不必等待操作完成。
   • 如果數(shù)據(jù)沒有到達(dá)或者連接尚未建立，程序不會被掛起，而是會立即返回一個(gè)狀態(tài)，告訴調(diào)用者當(dāng)前操作尚未完成。
   • 在非阻塞模型中，程序可以不斷輪詢網(wǎng)絡(luò)狀態(tài)，不斷嘗試進(jìn)行數(shù)據(jù)讀取或者連接操作，直到操作完成或者發(fā)生錯(cuò)誤。
   • 通過使用非阻塞模型，一個(gè)線程可以同時(shí)處理多個(gè)連接，避免了為每個(gè)連接創(chuàng)建新線程的開銷，提高了系統(tǒng)的性能和資源利用率。

在實(shí)際的網(wǎng)絡(luò)編程中，可以根據(jù)具體的需求和系統(tǒng)性能要求選擇合適的編程模型。阻塞模型通常更加簡單直觀，適用于連接數(shù)較少且并發(fā)要求不高的場景；而非阻塞模型更加靈活，適用于需要處理大量并發(fā)連接的高性能網(wǎng)絡(luò)應(yīng)用。

4.1.1.阻塞

阻塞模式 一個(gè)線程，可能會影響別的線程運(yùn)行

accept 會影響 read , read 也會影響 accept

Server

/**** Server 服務(wù)端* @author 13723* @version 1.0* 2024/2/20 13:11*/
public class Server {private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());public static void main(String[] args) throws IOException {// 使用NIO來理解阻塞模式 單線程進(jìn)行處理ByteBuffer buffer = ByteBuffer.allocate(16);// 1.創(chuàng)建一個(gè)ServerSocketChannel  創(chuàng)建一個(gè)服務(wù)器ServerSocketChannel ssc = ServerSocketChannel.open();// 2.綁定監(jiān)聽端口ssc.bind(new InetSocketAddress(9000));// 3.建立一個(gè)練級的集合List<SocketChannel> socketChannelList = new ArrayList<SocketChannel>();while (true){logger.error("--------------- connection  start ----------------");// 3.accept 建立和客戶端之間的連接，說白了就是和客戶端之間進(jìn)行通信// 這里會的方法會阻塞，線程會停止運(yùn)行 (這里會等一個(gè)新的連接，如果沒有新的連接建立會一直阻塞在這里)SocketChannel sc = ssc.accept();logger.error("--------------- connection  {} ----------------",sc);socketChannelList.add(sc);// 5.介紹客戶端發(fā)送的數(shù)據(jù)for (SocketChannel socketChannel : socketChannelList) {logger.error("--------------- before read  ----------------");socketChannel.read(buffer);// 切換為讀模式buffer.flip();ByteBufferUtil.debugAll(buffer);// 切換為寫模式 重新接收新的數(shù)據(jù)buffer.clear();logger.error("--------------- after read  ----------------");}}}
}

Client

/*** 客戶端* @author 13723* @version 1.0* 2024/2/20 13:24*/
public class Client {private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());public static void main(String[] args) throws IOException {// 1.創(chuàng)建客戶端連接SocketChannel sc = SocketChannel.open();// 2.設(shè)置連接信息sc.connect(new InetSocketAddress("localhost",9000));// 等待logger.error("--------------- waiting ---------------");}
}

啟動的時(shí)候，Server正常進(jìn)行啟動，Client 以debug的方式進(jìn)行啟動

4.1.2.非阻塞

server

缺點(diǎn)：很明顯，當(dāng)我們沒有數(shù)據(jù)的時(shí)候

? accept 和 read 還再循環(huán)

public class Server {private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());public static void main(String[] args) throws IOException {// 使用NIO來理解阻塞模式 單線程進(jìn)行處理ByteBuffer buffer = ByteBuffer.allocate(16);// 1.創(chuàng)建一個(gè)ServerSocketChannel  創(chuàng)建一個(gè)服務(wù)器ServerSocketChannel ssc = ServerSocketChannel.open();// TODO 設(shè)置為非阻塞模式ssc.configureBlocking(false);// 2.綁定監(jiān)聽端口ssc.bind(new InetSocketAddress(9000));// 3.建立一個(gè)練級的集合List<SocketChannel> socketChannelList = new ArrayList<SocketChannel>();while (true){// logger.error("--------------- connection  start ----------------");// 3.accept 建立和客戶端之間的連接，說白了就是和客戶端之間進(jìn)行通信// 切換成非阻塞模式了，如果沒有連接建立 返回的時(shí)一個(gè)null值SocketChannel sc = ssc.accept();if (sc != null){logger.error("--------------- connection  {} ----------------",sc);sc.configureBlocking(false);socketChannelList.add(sc);}// 5.介紹客戶端發(fā)送的數(shù)據(jù)for (SocketChannel socketChannel : socketChannelList) {// logger.error("--------------- before read  ----------------");// 編程非阻塞，但是線程仍然會繼續(xù)運(yùn)行 如果沒有讀取到數(shù)據(jù) read會返回0int read = socketChannel.read(buffer);if (read > 0){// 切換為讀模式buffer.flip();ByteBufferUtil.debugAll(buffer);// 切換為寫模式 重新接收新的數(shù)據(jù)buffer.clear();logger.error("--------------- after read  ----------------");}}}}
}

client

客戶端代碼 和上面一樣沒有做額外改動

4.2.Selector

介紹選擇器Selector之前，先介紹一個(gè)概念 IO事件

IO事件

• IO事件表示通道內(nèi)的某種IO操作已經(jīng)準(zhǔn)備就緒

例如：在Server Scoket通道上發(fā)生的一個(gè)IO事件，代表一個(gè)新的連接已經(jīng)準(zhǔn)備好，這個(gè)事件就叫做接收就緒事件?；蛘哒f，一個(gè)通道內(nèi)如果有數(shù)據(jù)可以讀取，就會發(fā)生一個(gè)IO事件，代表該連接數(shù)據(jù)已經(jīng)準(zhǔn)備好，這個(gè)事件就叫做讀就緒事件

JavaNIO將NIO事件做了簡化，只定義了四個(gè)事件，他們用SelectionKey的4個(gè)常量來表示

• SelectionKey.OP_CONNECT
  • 表示連接就緒事件，用于表示客戶端連接建立后觸發(fā)的事件。客戶端的 SocketChannel 關(guān)注此事件，以便在連接建立后執(zhí)行相應(yīng)的操作。
• SelectionKey.OP_ACCEPT
  • 表示接受連接就緒事件，用于表示服務(wù)器端有連接請求時(shí)觸發(fā)的事件。服務(wù)器端的 ServerSocketChannel 關(guān)注此事件，以便在有新的連接請求時(shí)執(zhí)行相應(yīng)的操作。
• SelectionKey.OP_READ
  • 表示讀就緒事件，用于表示通道中有數(shù)據(jù)可以讀取的事件。通常由 SocketChannel 關(guān)注，以便在通道中有數(shù)據(jù)可讀時(shí)執(zhí)行相應(yīng)的讀取操作。
• SelectionKey.OP_WRITE
  • 表示寫就緒事件，用于表示通道可以寫入數(shù)據(jù)的事件。通常由 SocketChannel 關(guān)注，以便在通道可寫入數(shù)據(jù)時(shí)執(zhí)行相應(yīng)的寫入操作

管理多個(gè)channel , 可以發(fā)現(xiàn)channel是否有事件發(fā)生，有事件發(fā)生再去執(zhí)行 防止cpu空轉(zhuǎn)造成系統(tǒng)資源浪費(fèi)

/**** Server 服務(wù)端* @author 13723* @version 1.0* 2024/2/20 13:11*/
public class Server {private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());public static void main(String[] args) throws IOException {// 創(chuàng)建一個(gè)Selector對象Selector selector = Selector.open();ByteBuffer buffer = ByteBuffer.allocate(16);// 創(chuàng)建channelServerSocketChannel ssc = ServerSocketChannel.open();ssc.configureBlocking(false);// 建立selector和Channel之間的連接（將Channel注冊到Selector中）// SelectionKey 將來事件發(fā)生后，通過它，可以知道哪種事件，是那個(gè)Channel發(fā)生的事件// 0 表示不關(guān)注任何事件SelectionKey sscKey = ssc.register(selector, 0, null);// ** 事件有四種類型// ?? accept 會在有連接請求時(shí)觸發(fā) （SelectionKey關(guān)注）// ?? connect 客戶端連接建立后觸發(fā)的事件// ?? read 可讀事件 （SocketChannel關(guān)注）// ?? write 可寫事件（SocketChannel關(guān)注）// 設(shè)置具體的事件 (設(shè)置只關(guān)注 accept事件);sscKey.interestOps(SelectionKey.OP_ACCEPT);// 2.綁定監(jiān)聽端口ssc.bind(new InetSocketAddress(9000));while (true){// 3.調(diào)用selector的select方法(沒有事件發(fā)生，那么還是阻塞的)// !! 注意Selector在編程時(shí)，未處理時(shí)，不會阻塞會一直進(jìn)行執(zhí)行(要么處理，要么取消 不能不管)selector.select();// 4.處理事件(selectionKeys 中所有的可用的事件)Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();// 遍歷的時(shí)候 想要?jiǎng)h除 必須使用迭代器遍歷while (iterator.hasNext()){SelectionKey key = iterator.next();logger.error("key : {}",key);// 拿到對應(yīng)channelServerSocketChannel channel = (ServerSocketChannel)key.channel();// 建立連接SocketChannel accept = channel.accept();logger.error("accept : {}",accept);// 一個(gè)處理是accept 還可以進(jìn)行取消// key.cancel();}}}
}

4.2.1處理read

讀取數(shù)據(jù) 每個(gè)channel 里面 針對不同的事件類型 又創(chuàng)建了不同的channel進(jìn)行維護(hù)

public class Server {private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());public static void main(String[] args) throws IOException {// 創(chuàng)建一個(gè)Selector對象Selector selector = Selector.open();// 創(chuàng)建channelServerSocketChannel ssc = ServerSocketChannel.open();ssc.configureBlocking(false);// 建立selector和Channel之間的連接（將Channel注冊到Selector中）// SelectionKey 將來事件發(fā)生后，通過它，可以知道哪種事件，是那個(gè)Channel發(fā)生的事件// 0 表示不關(guān)注任何事件SelectionKey sscKey = ssc.register(selector, 0, null);// ** 事件有四種類型// ?? accept 會在有連接請求時(shí)觸發(fā) （SelectionKey關(guān)注）// ?? connect 客戶端連接建立后觸發(fā)的事件// ?? read 可讀事件 （SocketChannel關(guān)注）// ?? write 可寫事件（SocketChannel關(guān)注）// 設(shè)置具體的事件 (設(shè)置只關(guān)注 accept事件);sscKey.interestOps(SelectionKey.OP_ACCEPT);// 2.綁定監(jiān)聽端口ssc.bind(new InetSocketAddress(9000));while (true){// 3.調(diào)用selector的select方法(沒有事件發(fā)生，那么還是阻塞的)// !! 注意Selector在編程時(shí)，未處理時(shí)，不會阻塞會一直進(jìn)行執(zhí)行(要么處理，要么取消 不能不管)selector.select();// 4.處理事件(selectionKeys 中所有的可用的事件)Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();// 遍歷的時(shí)候 想要?jiǎng)h除 必須使用迭代器遍歷while (iterator.hasNext()){SelectionKey key = iterator.next();logger.error("key : {}",key);// 5.區(qū)分事件類型if (key.isAcceptable()) {// accept事件// 拿到對應(yīng)channelServerSocketChannel channel = (ServerSocketChannel)key.channel();// 建立連接SocketChannel sc = channel.accept();// 設(shè)置channel為非阻塞的sc.configureBlocking(false);// 將管理權(quán)交給selector（負(fù)責(zé)管理當(dāng)前處理的channel）SelectionKey scKey = sc.register(selector, 0, null);// 注意 這里是readscKey.interestOps(SelectionKey.OP_READ);logger.error("sc : {}",sc);}else if (key.isReadable()){// 讀取數(shù)據(jù)的事件SocketChannel channel = (SocketChannel) key.channel();ByteBuffer buffer = ByteBuffer.allocate(16);channel.read(buffer);// 切換為讀模式buffer.flip();ByteBufferUtil.debugRead(buffer);}}}}
}

4.2.2.用完key之后為什么要remove

重點(diǎn)：SelectedKey 只會往里面添加 key ，但是不會進(jìn)行刪除（也就是事件處理完成后，會標(biāo)記成處理，但是不會刪除）

// TODO 刪除key 一定要?jiǎng)h除
// SelectedKey 只會往里面添加 key ，但是不會進(jìn)行刪除（也就是事件處理完成后，會標(biāo)記成處理，但是不會刪除）
// 不然下次進(jìn)來還是上一個(gè)Key上一個(gè)Key是沒有事件，所有會報(bào)空指針
// 這就是這里要使用迭代器的原因，迭代器可以邊遍歷邊刪除，forEach不行
iterator.remove();

4.2.3.處理客戶端斷開

else if (key.isReadable()){try {// 讀取數(shù)據(jù)的事件SocketChannel channel = (SocketChannel) key.channel();ByteBuffer buffer = ByteBuffer.allocate(16);// 如果是正常斷開。那么read返回的是-1 因?yàn)槊看螖嚅_都會觸發(fā)一次讀事件int read = channel.read(buffer);if (read == -1){// 刪除keykey.cancel();}else {// 切換為讀模式buffer.flip();ByteBufferUtil.debugRead(buffer);// TODO 刪除key 一定要?jiǎng)h除// SelectedKey 只會往里面添加 key ，但是不會進(jìn)行刪除（也就是事件處理完成后，會標(biāo)記成處理，但是不會刪除）// 不然下次進(jìn)來還是上一個(gè)Key上一個(gè)Key是沒有事件，所有會報(bào)空指針iterator.remove();}}catch (Exception e){// 客戶端關(guān)閉了,這里需要將key從SelectedKey集合中真正的刪除e.printStackTrace();key.cancel();}
}

正常斷開

// 客戶端 需要手動調(diào)用close
public class Client {private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());public static void main(String[] args) throws IOException {// 1.創(chuàng)建客戶端連接SocketChannel sc = SocketChannel.open();// 2.設(shè)置連接信息sc.connect(new InetSocketAddress("localhost",9000));// 等待logger.error("--------------- waiting ---------------");// 正常斷開 不寫就是異常斷開sc.close();}
}

異常斷開（強(qiáng)制斷開）

4.2.4.處理消息邊界

當(dāng)客戶端發(fā)動的服務(wù)端的中文信息過長時(shí)，就可能會出現(xiàn)亂碼的情況

• 一種思路是，固定消息的長度，數(shù)據(jù)包的大小一樣，服務(wù)器按照預(yù)定長度讀取，缺點(diǎn)是浪費(fèi)帶寬
• 另一種思路是按照分隔符拆分，缺點(diǎn)是效率低下
• TLV格式，Type類型，Length長度，Value數(shù)據(jù)，類型和長度已知情況下，就可以方便獲取消息大小，分配合適的buffer,缺點(diǎn)是buffer需要提前分配，如果內(nèi)容過大，則會影響server吞吐量
  • HTTP 1.1 是LTV格式
  • HTTP 2.0 是LTV格式

server

每次將ByteBuffer作為參數(shù)進(jìn)行傳遞 也就是 通過

? 服務(wù)端 注冊
? ByteBuffer buffer = ByteBuffer.allocate(15);
? SelectionKey scKey = sc.register(selector, 0, buffer);

? 客戶端 獲取 重新設(shè)置

? ByteBuffer buffer = (ByteBuffer) key.attachment();

? key.attach(newByteBuffer);

public class Server {private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());public static void main(String[] args) throws IOException {// 創(chuàng)建一個(gè)Selector對象Selector selector = Selector.open();// 創(chuàng)建channelServerSocketChannel ssc = ServerSocketChannel.open();ssc.configureBlocking(false);// 建立selector和Channel之間的連接（將Channel注冊到Selector中）// SelectionKey 將來事件發(fā)生后，通過它，可以知道哪種事件，是那個(gè)Channel發(fā)生的事件// 0 表示不關(guān)注任何事件SelectionKey sscKey = ssc.register(selector, 0, null);// ** 事件有四種類型// ?? accept 會在有連接請求時(shí)觸發(fā) （SelectionKey關(guān)注）// ?? connect 客戶端連接建立后觸發(fā)的事件// ?? read 可讀事件 （SocketChannel關(guān)注）// ?? write 可寫事件（SocketChannel關(guān)注）// 設(shè)置具體的事件 (設(shè)置只關(guān)注 accept事件);sscKey.interestOps(SelectionKey.OP_ACCEPT);// 2.綁定監(jiān)聽端口ssc.bind(new InetSocketAddress(9000));while (true){// 3.調(diào)用selector的select方法(沒有事件發(fā)生，那么還是阻塞的)// !! 注意Selector在編程時(shí)，未處理時(shí)，不會阻塞會一直進(jìn)行執(zhí)行(要么處理，要么取消 不能不管)selector.select();// 4.處理事件(selectionKeys 中所有的可用的事件)Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();// 遍歷的時(shí)候 想要?jiǎng)h除 必須使用迭代器遍歷while (iterator.hasNext()){SelectionKey key = iterator.next();logger.error("key : {}",key);// 5.區(qū)分事件類型if (key.isAcceptable()) {// accept事件// 拿到對應(yīng)channelServerSocketChannel channel = (ServerSocketChannel)key.channel();// 建立連接SocketChannel sc = channel.accept();// 設(shè)置channel為非阻塞的sc.configureBlocking(false);// 將管理權(quán)交給selector（負(fù)責(zé)管理當(dāng)前處理的channel）//!! 1.將ByteBuffer 注冊到SelectionKey中，這樣保證每個(gè)人SelectionKey都有一個(gè)獨(dú)有的ByteBuff//!! 這種稱為附件 attachment//!! buffer不在作為局部變量了ByteBuffer buffer = ByteBuffer.allocate(15);SelectionKey scKey = sc.register(selector, 0, buffer);// 注意 這里是readscKey.interestOps(SelectionKey.OP_READ);logger.error("sc : {}",sc);// TODO 刪除key 一定要?jiǎng)h除// SelectedKey 只會往里面添加 key ，但是不會進(jìn)行刪除（也就是事件處理完成后，會標(biāo)記成處理，但是不會刪除）// 不然下次進(jìn)來還是上一個(gè)Key上一個(gè)Key是沒有事件，所有會報(bào)空指針iterator.remove();}else if (key.isReadable()){try {// 讀取數(shù)據(jù)的事件SocketChannel channel = (SocketChannel) key.channel();// !!2.從讀事件中 拿到附件ByteBuffer buffer = (ByteBuffer) key.attachment();// 為了保證每個(gè)Channel都有一個(gè)獨(dú)有的ByteBuffer// 如果是正常斷開。那么read返回的是-1 因?yàn)槊看螖嚅_都會觸發(fā)一次讀事件int read = channel.read(buffer);if (read == -1){// 刪除keykey.cancel();}else {// 切換為讀模式buffer.flip();split(buffer);// !!3.判斷一次是否讀取完全// 如果position 和 limit一樣 說明沒有讀取完成，需要擴(kuò)容if (buffer.position() == buffer.limit()){ByteBuffer newByteBuffer = ByteBuffer.allocate(buffer.capacity() * 2);// 新的bytebuffer 是舊的兩倍 將舊的ByteBuffer內(nèi)容設(shè)置的到新的中buffer.flip();newByteBuffer.put(buffer);// 新的buffer替換原來的bufferkey.attach(newByteBuffer);}iterator.remove();}}catch (Exception e){// 客戶端關(guān)閉了,這里需要將key從SelectedKey集合中真正的刪除e.printStackTrace();key.cancel();}}}}}private static void split(ByteBuffer source) {// 找到一個(gè)完整消息， \nfor (int i = 0; i < source.limit(); i++) {if (source.get(i)  == '\n') {// 計(jì)算消息的長度 (換行符合 + 1 - 起始索引(就是ByteBuffer的Position))int length = i + 1 - source.position();// 找到一個(gè)完整消息了（get(i)不會移動指針）ByteBuffer target = ByteBuffer.allocate(length);// 從source讀取，向target寫for (int j = 0; j < length; j++) {target.put(source.get());}// 打印拆出來的信息ByteBufferUtil.debugAll(target);}}// 因?yàn)榭赡苓€有沒有讀取完成的數(shù)據(jù)，比如一半的數(shù)據(jù)，留給下次讀取source.compact();}}

4.2.5.ByteBuffer大小的分配

• 每個(gè)channel都需要記錄可能被切分的消息，因?yàn)锽yteBuffer不能被多個(gè)channel共同使用，因此需要為每個(gè)channel維護(hù)—個(gè)獨(dú)立的 ByteBuffer
• ByteBuffer 不能太大，比如一個(gè)ByteBuffer 1Mb的話，要支持百萬連接就要1Tb內(nèi)存，因此需要設(shè)計(jì)大小可變的 ByteBuffer
  • 一種思路是首先分配一個(gè)較小的buffer，例如4k，如果發(fā)現(xiàn)數(shù)據(jù)不夠，再分配8k的buffer，將4kbuffer內(nèi)容拷貝至8k buffer，優(yōu)點(diǎn)是消息連續(xù)容易處理，缺點(diǎn)是數(shù)據(jù)拷貝耗費(fèi)性能，參考實(shí)現(xiàn)http://tutorials,jenkov.com/java-performance/resizable-array.html
  • 另一種思路是用多個(gè)數(shù)組組成buffer，一個(gè)數(shù)組不夠，把多出來的內(nèi)容寫入新的數(shù)組，與前面的區(qū)別是消息存儲不連續(xù)解析復(fù)雜，優(yōu)點(diǎn)是避免了拷貝引起的性能損耗

4.2.6.寫入內(nèi)容過多問題

服務(wù)端一次向客戶端寫入太大的數(shù)據(jù)

服務(wù)端

public class WriteServer {private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());public static void main(String[] args) throws IOException {ServerSocketChannel ssc = ServerSocketChannel.open();ssc.configureBlocking(false);Selector selector = Selector.open();// 直接關(guān)注 accept事件ssc.register(selector, SelectionKey.OP_ACCEPT);ssc.bind(new InetSocketAddress(9000));while (true){selector.select();Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();while (iterator.hasNext()){SelectionKey key = iterator.next();iterator.remove();if (key.isAcceptable()){SocketChannel sc = ssc.accept();sc.configureBlocking(false);// 向客戶端發(fā)送大量數(shù)據(jù)StringBuffer sb = new StringBuffer();for (int i = 0; i < 3000000; i++) {sb.append("a");}ByteBuffer buffer = Charset.defaultCharset().encode(sb.toString());// 通過Channel寫入數(shù)據(jù) 并不能保證一次將所有數(shù)據(jù)寫入到 客戶端// 返回值代表實(shí)際寫入的字節(jié)數(shù)while (buffer.hasRemaining()){int write = sc.write(buffer);logger.error("實(shí)際寫入的字節(jié)數(shù)：{}",write);}}}}}
}

客戶端

public class WriteClient {private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());public static void main(String[] args) throws IOException {SocketChannel sc = SocketChannel.open();sc.connect(new InetSocketAddress("localhost",9000));// 3.接收數(shù)據(jù)int count = 0;while (true){ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);count += sc.read(buffer);logger.error("接收的字節(jié)數(shù)：{}",count);buffer.compact();}}
}

問題

改進(jìn)

思路就是：先嘗試寫一次 如果一次沒寫完，那么就在關(guān)聯(lián)一個(gè)SelectionKey，繼續(xù)寫，就不用while循環(huán)一直在那里嘗試寫了，注意的是，SelectionKey 是可以 進(jìn)行相加的，比如 既可以讀 也可以 ，通過附件 attach傳遞沒有發(fā)送完的數(shù)據(jù)。

注意 讀取完成后 記得把數(shù)據(jù)釋放掉。

while (true){selector.select();Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();while (iterator.hasNext()){SelectionKey key = iterator.next();iterator.remove();if (key.isAcceptable()){SocketChannel sc = ssc.accept();sc.configureBlocking(false);SelectionKey scKey = sc.register(selector, 0, null);// !! 這里可能原來的是讀取事件scKey.interestOps(SelectionKey.OP_READ);// 向客戶端發(fā)送大量數(shù)據(jù)StringBuffer sb = new StringBuffer();for (int i = 0; i < 30000000; i++) {sb.append("a");}ByteBuffer buffer = Charset.defaultCharset().encode(sb.toString());// 通過Channel寫入數(shù)據(jù) 并不能保證一次將所有數(shù)據(jù)寫入到 客戶端// 返回值代表實(shí)際寫入的字節(jié)數(shù)int write = sc.write(buffer);logger.error("實(shí)際寫入的字節(jié)數(shù)：{}",write);// 先嘗試寫了一次，然后觀察是否還有剩余內(nèi)容if(buffer.hasRemaining()){// 關(guān)注一個(gè)寫事件// !! 這里又加了一個(gè)寫事件，為了防止把原先的事件覆蓋，所以這里需要加上原來事件// 讀事件 1  寫事件 4  加一起 等于5 說明 又關(guān)注讀又關(guān)注寫scKey.interestOps(scKey.interestOps() + SelectionKey.OP_WRITE);// 要把未寫完的數(shù)據(jù) 放到SelectionKey中scKey.attach(buffer);}} else if (key.isWritable()) {// 把上一次 buffer取出來， 關(guān)注的socketChannel拿出來ByteBuffer buffer = (ByteBuffer) key.attachment();SocketChannel sc = (SocketChannel) key.channel();// 繼續(xù)寫(數(shù)據(jù)量很多 就會反復(fù)進(jìn)入可寫事件)int write = sc.write(buffer);logger.error("實(shí)際寫入的字節(jié)數(shù)：{}",write);// 寫完清理附件if (!buffer.hasRemaining()){// 內(nèi)容寫完了 清楚buffer 可寫事件 也不需要進(jìn)行關(guān)聯(lián)了key.attach(null);key.interestOps(key.interestOps() - SelectionKey.OP_WRITE);}}}}