如何利用CountDownLatch實現(xiàn)基于netty的BIO

今天就跟大家聊聊有關(guān)如何利用CountDownLatch實現(xiàn)基于netty的BIO，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結(jié)了以下內(nèi)容，希望大家根據(jù)這篇文章可以有所收獲。

Netty是一個NIO客戶端-服務(wù)器框架，在NIO模式下，線程直接返回結(jié)果，當(dāng)緩沖區(qū)準(zhǔn)備好數(shù)據(jù)之后再異步線程回調(diào)客戶端消息處理。不會像BIO一樣阻塞線程（I/O）并持續(xù)等待被調(diào)用方準(zhǔn)備好響應(yīng)數(shù)據(jù)才返回。

但是在某些使用socket通訊的場景下，使用了NIO的netty，但是調(diào)用方必須在當(dāng)前線程等待服務(wù)返回業(yè)務(wù)處理狀態(tài)之后才能進(jìn)行后續(xù)的操作，那么就需要用到BIO模式，那么除了使用java原生的BIO，這里介紹一種使用CountDownLatch+Netty來模擬BIO的實現(xiàn)。

下面介紹下實現(xiàn)過程和偽代碼，文章末尾提供完整示例下載地址。代碼僅供學(xué)習(xí)。

• 環(huán)境：netty4.1.27、eclipse，jdk1.8；

• 流程設(shè)計

• 原理說明

• 1、在客戶端發(fā)送數(shù)據(jù)給服務(wù)端時（本實例服務(wù)端采用java原生socket簡單實現(xiàn)），通過uuid生成本次發(fā)送的請求的唯一標(biāo)識（sn），并將此sn附在發(fā)送數(shù)據(jù)的頭部位置（服務(wù)端返回時需要帶上此sn），同時客戶端產(chǎn)生一個CountDownLatch和當(dāng)前sn綁定。

• 2、發(fā)送成功后，通過CountDownLatch.await阻塞當(dāng)前線程。

• 3、服務(wù)端接收數(shù)據(jù)并處理相關(guān)完后，封裝含客戶端sn的返回數(shù)據(jù)發(fā)送給客戶端，客戶端接收到服務(wù)端返回的信息后，根據(jù)返回的sn獲取到對應(yīng)的CountDownLatch，并調(diào)用countDown使得上一步阻塞的線程能夠繼續(xù)進(jìn)行后續(xù)業(yè)務(wù)操作。

• 關(guān)鍵代碼片段

• 發(fā)送消息

 /*** 發(fā)送消息* @param msg*/public void sendMsg(String ip,int port,String msg) {  try {     //1 連接到服務(wù)端，如果長連接可以修改此邏輯    connectToServer(ip, port);    //2生成阻塞計數(shù)器    String sn = UUID.randomUUID().toString();    Receiver.addWait(sn);     //3 發(fā)送數(shù)據(jù)    client.channel().writeAndFlush(sn+"@"+msg+"END").sync();//加end防止粘包    LOG.info("發(fā)送消息成功,msg:"+msg);     //4 等待接接收到消息后，計數(shù)器countdown。    Receiver.block(sn);//可以設(shè)置等待時間，以防止線程長時間阻塞。    //5處理服務(wù)端返回    handlerServerAnswer(sn);     } catch (Exception e) {      LOG.error("處理消息失敗.",e);    }}

• 接收消息

/**  * 處理服務(wù)端返回消息 */@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {   try {     String dataFromServer = (String) msg;     String[] snData = dataFromServer.split("@");     String sn = snData[0];     String data = snData[1];     response.put(sn, data);//等待業(yè)務(wù)邏輯消費     if(!waitLatch.containsKey(sn)) {       LOG.error("非法sn："+sn);       return;     }     waitLatch.get(sn).countDown();//取消阻塞，讓發(fā)送線程后續(xù)邏輯運行。     waitLatch.remove(sn);    } catch (Exception e) {      LOG.error("接收到非法數(shù)據(jù):"+msg);    } }

• 服務(wù)端代碼

/*** 模擬服務(wù)端* @param args* @throws Exception*/public static void main( String[] args ) throws Exception  {     serverSocket = new ServerSocket(PORT);     LOG.info("服務(wù)監(jiān)聽啟動，端口："+PORT);     while(true) {       try {        Socket socket = serverSocket.accept();        InputStream in = socket.getInputStream();        int len;        StringBuffer sb = new StringBuffer();        byte[] bytes = new byte[1024];        while((len = in.read(bytes))!=-1) {          String bytesData = new String(bytes,0,len,"utf-8");          sb.append(bytesData);          if(bytesData.endsWith("END")) {            //處理業(yè)務(wù)邏輯并返回消息            handMsgAndResponse(socket,sb.toString());            sb = null;          }        }      } catch (Exception e) {        LOG.error("處理客戶端消息失敗.",e);      }          Thread.sleep(1000);        }  }

看完上述內(nèi)容，你們對如何利用CountDownLatch實現(xiàn)基于netty的BIO有進(jìn)一步的了解嗎？如果還想了解更多知識或者相關(guān)內(nèi)容，請關(guān)注億速云行業(yè)資訊頻道，感謝大家的支持。