java架構之路-（netty專題）netty的基本使用和netty聊天室

阿新 • • 發佈：2020-02-14

上次回顧：

　　上次部落格，我們主要說了我們的IO模型，BIO同步阻塞，NIO同步非阻塞，AIO基於NIO二次封裝的非同步非阻塞，最重要的就是我們的NIO，腦海中應該有NIO的模型圖。

Netty概念：

　　Netty是由JBOSS提供的一個java開源框架，現為 Github上的獨立專案。Netty提供非同步的、事件驅動的網路應用程式框架和工具，用以快速開發高效能、高可靠性的網路伺服器和客戶端程式。也就是說，Netty 是一個基於NIO的客戶、伺服器端的程式設計框架，使用Netty 可以確保你快速和簡單的開發出一個網路應用，例如實現了某種協議的客戶、服務端應用。Netty相當於簡化和流線化了網路應用的程式設計開發過程，例如：基於TCP和UDP的socket服務開發。 “快速”和“簡單”並不用產生維護性或效能上的問題。Netty 是一個吸收了多種協議（包括FTP、SMTP、HTTP等各種二進位制文字協議）的實現經驗，並經過相當精心設計的專案。最終，Netty 成功的找到了一種方式，在保證易於開發的同時還保證了其應用的效能，穩定性和伸縮性。---摘自百度百科。

基本使用：

　　我們來看一下程式碼吧

　　服務端：

package com.xiaocai.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class NettyServer {

    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);//新建執行緒組,用於處理請求
        EventLoopGroup workerGroup = new NioEventLoopGroup();//真正工作的執行緒組
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();//建立一個server,相當於NIO的server
            bootstrap.group(bossGroup, workerGroup) //採用鏈式程式設計,將兩個執行緒組加入到server中
                    .channel(NioServerSocketChannel.class) //設定一個通道,相當於NIO中的ServerSocketChannel
                    .option(ChannelOption.SO_BACKLOG, 1024) //對服務端給予一些設定
                    .childHandler(new ChannelInitializer<SocketChannel>() {//給予初始化,並加入回撥函式
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new NettyServerHandler());
                        }
                    });
            System.out.println("netty server start。。");
            ChannelFuture cf = bootstrap.bind(9000).sync();//繫結埠,sync方法是等待非同步操作執行完畢
            cf.channel().closeFuture().sync();//對通道關閉進行監聽，closeFuture是非同步操作，監聽通道關閉
                                        // 通過sync方法同步等待通道關閉處理完畢，這裡會阻塞等待通道關閉完成
        } finally {
            bossGroup.shutdownGracefully(); //關閉處理請求的執行緒組
            workerGroup.shutdownGracefully();//關閉真正工作的執行緒組
        }
    }
}

　　服務端的程式碼看起來還是比較亂的，但不要擔心，基本都是固定寫法的，大致就開執行緒，設定server，設定通道，初始化，給予回撥函式，我們來看一下回調函式。

package com.xiaocai.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;

public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    //當客戶端連線伺服器完成就會觸發該方法
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ByteBuf buf = Unpooled.copiedBuffer("HelloServer".getBytes(CharsetUtil.UTF_8));
        ctx.writeAndFlush(buf);
    }

    //當通道有讀取事件時會觸發，即服務端傳送資料給客戶端
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("收到客戶端的訊息:" + buf.toString(CharsetUtil.UTF_8));
        System.out.println("客戶端的地址： " + ctx.channel().remoteAddress());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("我收到了異常");
        cause.printStackTrace();
        ctx.close();
    }
}

　　這個回撥函式也基本是固定的，繼承ChannelInboundHandlerAdapter 類，然後我們寫在裡面就可以了，幾乎需要我們做的主要是這個回撥函式。

　　客戶端：

package com.xiaocai.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class NettyClient {
    public static void main(String[] args) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();// 開啟工作執行緒組
        try {
            Bootstrap bootstrap = new Bootstrap(); //建立一個和服務端相對應的server
            bootstrap.group(group) //設定執行緒組
                    .channel(NioSocketChannel.class) //使用NioSocketChannel作為客戶端的通道實現
                    .handler(new ChannelInitializer<SocketChannel>() {//設定回撥函式
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ch.pipeline().addLast(new NettyClientHandler());
                        }
                    });
            System.out.println("netty client start。。");
            ChannelFuture cf = bootstrap.connect("127.0.0.1", 9000).sync();//啟動客戶端去連線伺服器端
            //對通道關閉進行監聽
            cf.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();//關閉執行緒組
        }
    }
}

　　客戶端和服務端差不多，開啟執行緒設定一個對應的server，設定通道，設定回撥函式，我們還是看一下客戶端的回撥函式做了什麼事情。

package com.xiaocai.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;

public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    /**
     * 當客戶端連線伺服器完成就會觸發該方法
     *
     * @param ctx 設定好的通道,相當於上下文
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        ByteBuf buf = Unpooled.copiedBuffer("HelloServer".getBytes(CharsetUtil.UTF_8));
        ctx.writeAndFlush(buf);
    }

    //當通道有讀取事件時會觸發，即服務端傳送資料給客戶端
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("收到服務端的訊息:" + buf.toString(CharsetUtil.UTF_8));
        System.out.println("服務端的地址： " + ctx.channel().remoteAddress());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

　　兩側的程式碼都是差不多的，回撥函式，建立連線以後我們做什麼，接收到資料以後我們做什麼，差不多就是這個意思。這裡為什麼是給予兩個執行緒呢？簡單來說就是為了能承受更高的併發，一個執行緒專門來負責接收連線，不至於讓有的請求卡死或延遲很高的狀態，後面說底層原始碼的時候，這個還會再說的。

Netty模組元件

　　Bootstrap、ServerBootstrap：

　　Bootstrap意思是引導，一個 Netty 應用通常由一個Bootstrap 開始，主要作用是配置整個 Netty 程式，串聯各個元件，Netty 中 Bootstrap 類是客戶端程式的啟動引導類，ServerBootstrap 是服務端啟動引導類。

　　Future、ChannelFuture：

　　正如前面介紹，在 Netty 中所有的 IO 操作都是非同步的，不能立刻得知訊息是否被正確處理。但是可以過一會等它執行完成或者直接註冊一個監聽，具體的實現就是通過 Future 和 ChannelFutures，他們可以註冊一個監聽，當操作執行成功或失敗時監聽會自動觸發註冊的監聽事件。

　　Channel：

　　Netty 網路通訊的元件，能夠用於執行網路 I/O 操作。Channel 為使用者提供：

　　1）當前網路連線的通道的狀態（例如是否開啟？是否已連線？）

　　2）網路連線的配置引數（例如接收緩衝區大小）

　　3）提供非同步的網路 I/O 操作(如建立連線，讀寫，繫結埠)，非同步呼叫意味著任何 I/O 呼叫都將立即返回，並且不保證在呼叫結束時所請求的 I/O 操作已完成。

　　4）呼叫立即返回一個 ChannelFuture 例項，通過註冊監聽器到 ChannelFuture 上，可以 I/O 操作成功、失敗或取消時回撥通知呼叫方。

　　5）支援關聯 I/O 操作與對應的處理程式。不同協議、不同的阻塞型別的連線都有不同的 Channel 型別與之對應。

NioSocketChannel;//非同步的客戶端 TCPSocket 連線。
NioServerSocketChannel;//非同步的伺服器端 TCP Socket 連線。
NioDatagramChannel;//非同步的 UDP 連線。
NioSctpChannel;//非同步的客戶端 Sctp 連線。
NioSctpServerChannel;//非同步的 Sctp 伺服器端連線，這些通道涵蓋了UDP和TCP網路IO以及檔案IO。

　　Selector：

　　Netty 基於 Selector 物件實現 I/O 多路複用，通過 Selector 一個執行緒可以監聽多個連線的 Channel 事件。當向一個 Selector 中註冊 Channel 後，Selector 內部的機制就可以自動不斷地查詢(Select) 這些註冊的 Channel 是否有已就緒的 I/O 事件（例如可讀，可寫，網路連線完成等），這樣程式就可以很簡單地使用一個執行緒高效地管理多個Channel 。

　　NioEventLoop：

　　NioEventLoop 中維護了一個執行緒和任務佇列，支援非同步提交執行任務，執行緒啟動時會呼叫 NioEventLoop 的 run 方法，執行 I/O 任務和非 I/O 任務： I/O 任務，即selectionKey中ready 的事件，如 accept、connect、read、write 等，由 processSelectedKeys方法觸發。非 IO 任務，新增到 taskQueue 中的任務，如 register0、bind0 等任務，由 runAllTasks 方法觸發。

　　NioEventLoopGroup：

　　NioEventLoopGroup，主要管理 eventLoop 的生命週期，可以理解為一個執行緒池，內部維護了一組執行緒，每個執行緒(NioEventLoop)負責處理多個 Channel 上的事件，而一個 Channel 只對應於一個執行緒。

　　ChannelHandler：

　　ChannelHandler 是一個介面，處理 I/O 事件或攔截 I/O 操作，並將其轉發到其 ChannelPipeline(業務處理鏈)中的下一個處理程式。 ChannelInboundHandler 用於處理入站 I/O 事件，ChannelOutboundHandler用於處理出站 I/O事件。

　　ChannelHandlerContext：

　　儲存 Channel 相關的所有上下文資訊，同時關聯一個 ChannelHandler 物件。

　　ChannelPipline：

　　儲存 ChannelHandler 的List，用於處理或攔截 Channel 的入站事件和出站操作。內部是一個雙向連結串列，記住tail和head，後面說編解碼會詳細去說。

netty聊天室：

　　網際網路行業：在分散式系統中，各個節點之間需要遠端服務呼叫，高效能的 RPC 框架必不可少， Netty 作為非同步高效能的通訊框架，往往作為基礎通訊元件被這些 RPC 框架使用。典型的應用有：阿里分散式服務框架 Dubbo 的 RPC 框架使用 Dubbo 協議進行節點間通訊，Dubbo 協議預設使用 Netty 作為基礎通訊元件，用於實現。各程序節點之間的內部通訊。Rocketmq底層也是用的Netty作為基礎通訊元件。遊戲行業：無論是手遊服務端還是大型的網路遊戲，Java 語言得到了越來越廣泛的應用。Netty 作為高效能的基礎通訊元件，它本身提供了 TCP/UDP 和 HTTP 協議棧。我們來實現一個簡單的聊天室。一步步的我們拿著剛才的程式碼改造一下。

　　服務端：

package com.xiaocai.chat;

import com.xiaocai.netty.NettyServerHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

public class ChatRoomServer {

    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new StringEncoder());
                            ch.pipeline().addLast(new StringDecoder());
                            ch.pipeline().addLast(new ChatRoomServerHandler());
                        }
                    });
            System.out.println("聊天室開啟了！！！");
            ChannelFuture cf = bootstrap.bind(9000).sync();
            cf.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

　　程式還是和原來的程式差不多的，這裡都是一些固定的寫法，只有兩行程式碼我們剛才沒說，加入了兩個攔截器，一個是編碼一個是解碼，下次部落格會重點說這個，我們來看一下我們的回撥函式。

package com.xiaocai.chat;

import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

import java.text.SimpleDateFormat;
import java.util.Date;

public class ChatRoomServerHandler extends SimpleChannelInboundHandler<String> {

    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    //當客戶端連線伺服器完成就會觸發該方法
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        String nowTime = sdf.format(new Date());
        String userName = channel.id() + "";
        String msg = nowTime + " " + userName + " 上線了!";
        channelGroup.writeAndFlush(msg);
        channelGroup.add(channel);
        System.out.println("[服務端]: " + msg);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        //獲取到當前 channel
        Channel channel = ctx.channel();
        //這時我們遍歷 channelGroup, 根據不同的情況， 回送不同的訊息
        channelGroup.forEach(ch -> {
            if (channel != ch) { //不是當前的 channel,轉發訊息
                ch.writeAndFlush(channel.id() + " 傳送了訊息：" + msg);
            } else {//回顯自己傳送的訊息給自己
                ch.writeAndFlush("[ 自己 ]傳送了訊息：" + msg);
            }
        });
    }

    //表示 channel 處於不活動狀態, 提示離線了
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        Channel channel = ctx.channel();
        //將客戶離開資訊推送給當前線上的客戶
        channelGroup.writeAndFlush("[ 客戶端 ]" + channel.id() + " 下線了");
        System.out.println(channel.id() + " 下線了");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        //關閉通道
        ctx.close();
    }
}

　　這次我們繼承的是SimpleChannelInboundHandler這個類，他是對於ChannelInboundHandlerAdapter的再次封裝，然後我們的是通過通道組來發送訊息的，先來看channelActive方法，就是當有客戶端連線的時候，我們應該自己有所感知，並且將某某上線的訊息廣播出去。channelRead0方法就是當有資料傳輸過來時的一些處理，我們用channelGroup來判斷是否是自己傳送的訊息，是自己傳送的訊息我們給予自己說了什麼什麼，別的訊息就是某某某說了什麼什麼。

　　客戶端：

package com.xiaocai.chat;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.util.Scanner;

public class ChatRoomClient {
    public static void main(String[] args) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group) //設定執行緒組
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {//設定回撥函式
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ch.pipeline().addLast(new StringEncoder());
                            ch.pipeline().addLast(new StringDecoder());
                            ch.pipeline().addLast(new ChatRoomClientHandler());
                        }
                    });
            ChannelFuture cf = bootstrap.connect("127.0.0.1", 9000).sync();
            //得到 channel
            Channel channel = cf.channel();

            System.out.println("我" + channel.id() + "進入了聊天室");
            //客戶端需要輸入資訊， 建立一個掃描器
            Scanner scanner = new Scanner(System.in);
            while (scanner.hasNextLine()) {
                String msg = scanner.nextLine();
                //通過 channel 傳送到伺服器端
                channel.writeAndFlush(msg);
            }
            cf.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
}

　　客戶端的程式碼和原來的也是差不多了，還是多兩個攔截器，進行編碼解碼，然後下面就是我們的掃描器，來掃描控制檯輸入的，一會我們要用控制檯來完成這個聊天室，所以我們要掃描控制檯的輸入，我們來看一下我們的攔截器。

package com.xiaocai.chat;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class ChatRoomClientHandler extends SimpleChannelInboundHandler<String> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        System.out.println(msg);
    }
}

　　就是當有訊息發過來時，我們打印出來就可以了，我們來看看我們的效果。

　　大概就是這樣的，就是我們的客戶端傳送訊息，由服務端接收，然後由服務端再次轉發出去。

總結：

　　今天我們主要說了Netty的基本使用，需要記住那幾行固定程式碼都是什麼意思，為什麼要開啟兩個執行緒來共同處理，然後就是我們的一個小例子，聊天室程式，下次我們來說一下我們的編碼解碼，還有粘包拆包。

最進弄了一個公眾號，小菜技術，歡迎大家的加入

java架構之路-（netty專題）netty的基本使用和netty聊天室

java架構之路-（MQ專題）RabbitMQ安裝和基本使用

java架構之路-（Redis專題）Redis的高效能和持久化

java架構之路-（Redis專題）Redis的主從、哨兵和叢集

java架構之路-（Redis專題）SpringBoot連線Redis超簡單

java架構之路-（Redis專題）簡單聊聊redis分散式鎖

java架構之路-（Redis專題）聊聊大廠那些redis

java架構之路-（Redis專題）redis面試助力滿分+

java架構之路-（MQ專題）RocketMQ從入坑到叢集詳解

java架構之路（MQ專題）kafka叢集配置和簡單使用

java架構之路-（dubbo專題）dubbo的基本使用

java架構之路-（設計模式）五種建立型模式之單例模式

java架構之路-（分散式zookeeper）zookeeper叢集配置和選舉機制詳解

java架構之路-（SpringMVC篇）SpringMVC主要流程原始碼解析（上）原始碼執行流程

java架構之路-（面試篇）JVM虛擬機器面試大全

java架構之路-（面試篇）Mysql面試大全

java架構之路-（netty專題）初步認識BIO、NIO、AIO

java架構之路-（netty專題）netty的基本使用和netty聊天室

java架構之路-（netty專題）netty的編解碼（出入戰）與粘包拆包

java架構之路-（微服務專題）初步認識微服務與nacos初步搭建

java架構之路-（微服務專題）nacos叢集精講實戰

java架構之路-（netty專題）netty的基本使用和netty聊天室

相關推薦