Netty簡(jiǎn)單案例

• 前言
• • 環(huán)境準(zhǔn)備
• 前置知識(shí)
• • 網(wǎng)絡(luò)傳輸?shù)膸追N實(shí)現(xiàn)方式
  • • BIO——同步阻塞IO
    • NIO——同步非阻塞IO
    • AIO——異步非阻塞IO
    • 適用范圍
  • Netty
  • • 簡(jiǎn)介
    • 特點(diǎn)
    • 核心組件
    • 使用場(chǎng)景
    • 運(yùn)行簡(jiǎn)圖
• 案例
• • 簡(jiǎn)介
  • 關(guān)鍵代碼
  • • 客戶端
    • 服務(wù)器端
• 運(yùn)行狀況
• 總結(jié)

前言

最近學(xué)完了Netty，在這里關(guān)于Netty中實(shí)現(xiàn)NIO做一些小總結(jié)，并附上一個(gè)小案例，最好讀者有一點(diǎn)Netty的基礎(chǔ)。這里附上git的地址，看一下netty的各種案例運(yùn)行一下。github

環(huán)境準(zhǔn)備

Maven 3.X、JDK15

前置知識(shí)

網(wǎng)絡(luò)傳輸?shù)膸追N實(shí)現(xiàn)方式

建議看一下之前的一篇文章《Java網(wǎng)絡(luò)編程之阻塞式IO與非阻塞IO》中關(guān)于阻塞和非阻塞IO在Java中的使用。

BIO——同步阻塞IO

阻塞式的IO，服務(wù)器以輪詢的方式，不斷查看是否有新的連接。當(dāng)然其性能可以使用線程池得到略微改善。

NIO——同步非阻塞IO

通過Selector以及Channel的組合使用，實(shí)現(xiàn)了多路復(fù)用，雖然實(shí)現(xiàn)了服務(wù)器的異步處理，但是客戶端必須要在服務(wù)器響應(yīng)到達(dá)才能發(fā)起下一個(gè)請(qǐng)求，即客戶端需要某線程持續(xù)監(jiān)聽是否有響應(yīng)發(fā)送回來。大體流程如下：

AIO——異步非阻塞IO

該模式不僅服務(wù)器實(shí)現(xiàn)了異步、客戶端也實(shí)現(xiàn)了異步，能夠在請(qǐng)求沒有到達(dá)之前，繼續(xù)向服務(wù)器發(fā)送數(shù)據(jù)。這里之后補(bǔ)充。

適用范圍

• NIO方式適用于連接數(shù)目多且連接比較短（輕操作）的架構(gòu)，比如聊天服務(wù)器，并發(fā)局限于應(yīng)用中，JDK1.4開始支持。

• AIO方式使用于連接數(shù)目多且連接比較長(zhǎng)（重操作）的架構(gòu)，比如HTTP服務(wù)器等，充分調(diào)用OS參與并發(fā)操作，JDK7開始支持

Netty

簡(jiǎn)介

Netty是高性能的水平擴(kuò)展的分布式架構(gòu)

特點(diǎn)

健壯、安全、高可用、高性能、更新快、易用

核心組件

• channel：傳入、傳出的數(shù)據(jù)載體
• 回調(diào)：給請(qǐng)求的響應(yīng)
• Feature：異步編程的的一個(gè)任務(wù)的開啟
• 事件和ChannelHandler：很多框架，前端后端都包含這類"發(fā)布者訂閱者"設(shè)計(jì)思想

使用場(chǎng)景

• 內(nèi)部的RPC框架
  低延遲高吞吐量
• 負(fù)載和性能測(cè)試
  用于負(fù)載和性能測(cè)試框架，可以通過Netty和Redis結(jié)合，來以最小的負(fù)載測(cè)試端到端的消息吞吐量。
• 同步協(xié)議的異步客戶端
  Netty為同步的協(xié)議創(chuàng)建異步的客戶端，如Kafka、Memcached。使得在同步和異步之間來回切換，不需要更改任何上有代碼。

消息推送、實(shí)時(shí)流量監(jiān)控都可以使用Netty，有待大家自己探索。

運(yùn)行簡(jiǎn)圖

簡(jiǎn)而言之，服務(wù)器監(jiān)聽端口，就是新建一個(gè)ServerChannel，客戶端建立連接connect也是建立一個(gè)連接，之后ServerChannel收到之后通過EventLoopGroup把該channel分配到EventLoop，如果該channel對(duì)應(yīng)的任務(wù)已經(jīng)存在于EventLoop中就直接執(zhí)行，否則就要放入EventLoop中，等待執(zhí)行。

• EventLoopGroup
  EventLoopGroup好比線程池，EventLoop好比線程，Channel的pipeline了所有的Handler
  

• EventLoop的Task注冊(cè)
  

• 連接建立示意圖

  • 服務(wù)器
    

  • 客戶端
    

案例

簡(jiǎn)介

客戶通過一個(gè)終端輸入查詢的編號(hào)，如果問題在庫(kù)中存在，人工客服返回相應(yīng)的回答；若不存在，就會(huì)返回默認(rèn)回復(fù)——”致電人工客服“；

如果是非法的輸入，就直接模擬服務(wù)器宕機(jī)，所有的連接都斷開。

注意 ：這里后續(xù)作為入門案例展示，如果這里看不太懂，建議看一下git的quickstart部分。這里只展示關(guān)鍵代碼

關(guān)鍵代碼

客戶端

public void start() {Bootstrap bootstrap = new Bootstrap ( );// 引導(dǎo)類的配置，包括事件組，channel類型以及處理器bootstrap.group (group).channel (NioSocketChannel.class).handler (createInitializer ( ));ChannelFuture future = bootstrap.connect (new InetSocketAddress (8080));future.syncUninterruptibly ( );channel = future.channel ( );}// 注冊(cè)所有的處理器private ChannelInitializer<Channel> createInitializer() {return new TerminalChatClientInitializer ( );}// 通過控制臺(tái)不斷地詢問智能客服static void chat() throws IOException {while (true) {String input = getInputMsg ( );if (!input.equals (OVER)) {channel.writeAndFlush (Unpooled.copiedBuffer (input, StandardCharsets.UTF_8));} else {break;}}} // 處理器代碼 public class TerminalClientInHandler extends SimpleChannelInboundHandler<ByteBuf> {// 把響應(yīng)打印出來@Overrideprotected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {out.println (LocalDateTime.now ( ).format (DateTimeFormatter.ofPattern ("yyyy/MM/dd HH:mm:ss")));String resp = msg.toString (StandardCharsets.UTF_8);out.println (resp);}// 提示連接成功@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {out.println ("connect to server successfully : " + ctx.channel ( ).remoteAddress ( ));}// 提示連接關(guān)閉@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {out.println ("已經(jīng)關(guān)閉連接");} }

服務(wù)器端

public void start() {// 初始化服務(wù)器引導(dǎo)ServerBootstrap server = new ServerBootstrap ( );server.group (mainGroup).channel (NioServerSocketChannel.class).childHandler (createInitializer (channelGroup));// 監(jiān)聽8080端口ChannelFuture future = server.bind (8080);future.syncUninterruptibly ( );channel = future.channel ( );}private ChannelInitializer<Channel> createInitializer(ChannelGroup channelGroup) {return new TerminalChatServerInitializer (channelGroup);}public void destroy() {if (channel != null) {channel.close ( );}mainGroup.shutdownGracefully ( );subGroup.shutdownGracefully ( );} // 處理器關(guān)鍵代碼 protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception { // 打印收到的時(shí)間和消息String input = msg.toString (StandardCharsets.UTF_8);out.println (ctx.channel ().remoteAddress () + "\n" + input);try {// 解析查詢的Id,并且響應(yīng)相應(yīng)的答案int id = Integer.parseInt (input);String res = resMap.getOrDefault (id, "請(qǐng)致電人工：10086");ctx.writeAndFlush (Unpooled.copiedBuffer (res, StandardCharsets.UTF_8));} catch (Exception e) {// 非法輸入，模擬服務(wù)器宕機(jī)，向所有的客戶端發(fā)送 服務(wù)器關(guān)閉消息group.writeAndFlush (Unpooled.copiedBuffer ("不明原因，服務(wù)器暫時(shí)關(guān)閉",StandardCharsets.UTF_8));group.close ();}}// 有新連接，打印客戶機(jī)地址，并且加入到聊天群組@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {out.println ("client connected successfully : " + ctx.channel ( ).remoteAddress ( ));group.add (ctx.channel ( ));}// 某客戶端下線，打印客戶機(jī)地址，這里不需要手動(dòng)從群組移除，Netty已經(jīng)幫我們實(shí)現(xiàn)了該功能@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {out.println ("客戶端: " + ctx.channel ( ).remoteAddress ( ).toString ( ) + " 結(jié)束");}

運(yùn)行狀況

總結(jié)

相對(duì)于Java原生Nio的消息讀取以及消息處理來說，Netty的實(shí)現(xiàn)方式更加簡(jiǎn)單。完整代碼放于git上了。

若文章有錯(cuò)誤，歡迎大家指正，同時(shí)希望大佬能夠給予一些指導(dǎo)！

總結(jié)

以上是生活随笔為你收集整理的Netty入门——基于NIO实现机器客服案例的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。