Netty網(wǎng)絡編程實戰(zhàn)之搭建Netty服務器

更新時間：2022年10月10日 10:17:06 作者：哪?吒

Netty是JBOSS開源的一款NIO網(wǎng)絡編程框架，可用于快速開發(fā)網(wǎng)絡的應用。Netty是一個異步的、基于事件驅動的網(wǎng)絡應用框架，用于快速開發(fā)高性能的服務端和客戶端。本文將詳細說說如何搭建Netty服務器，需要的可以參考一下

一、Netty是什么

Netty是JBOSS開源的一款NIO網(wǎng)絡編程框架，可用于快速開發(fā)網(wǎng)絡的應用。Netty是一個異步的、基于事件驅動的網(wǎng)絡應用框架，用于快速開發(fā)高性能的服務端和客戶端。可以極大的簡化基于TCP、UDP等協(xié)議的網(wǎng)絡服務。并且Netty對于各種傳輸類型（阻塞或非阻塞式的socket）、通信方式（HTTP或websocket）都提供了統(tǒng)一的API接口，提供了靈活的可擴展性，高度可自定義的線程模型（多線程、線程池等），支持使用無連接的數(shù)據(jù)報UDP進行通信，具有高吞吐量、低延遲、資源消耗低、最低限度的內(nèi)存復制等特性。除了優(yōu)越的性能外，Netty還支持SSL/TLS和StartTLS等加密傳輸協(xié)議，保證了數(shù)據(jù)傳輸?shù)陌踩浴?/p>

在實際使用時，Netty可以作為Socket編程的中間件；也可以和Protobuf技術結合使用，實現(xiàn)一個RPC框架，實現(xiàn)遠程過程調用；也可以作為一個websocket的長鏈接服務器，實現(xiàn)客戶端和服務端的長連接通信。

二、Hello Netty

使用Netty作為一個Web服務器，用于接收用戶請求并給出響應。

Netty程序一般都是按套路來寫，依次編寫主程序類、自定義初始化器、自定義處理器。

1、主程序類MyNettyServerTest

通過ServerBootstrap注冊serverGroup和clientGroup兩個事件循環(huán)組，其中serverGroup用于獲取客戶端連接，clientGroup用于處理客戶端連接，類似于常見的Master-Slave結構。

2、初始化器MyNettyServerInitializer

繼承Netty提供的初始化器ChannelInitializer。

Netty封裝了各種各樣的內(nèi)置處理器，用于實現(xiàn)各種功能。并且ChannelInitializer的initChannel()方法會在某一個連接注冊到Channel后立即被觸發(fā)調用。因此，可以根據(jù)業(yè)務需求，在initChannel()中添加若干個Netty內(nèi)置處理器，利用Netty強大的類庫直接處理大部分業(yè)務。最后再在initChannel()中添加一個自定義處理器，用于實現(xiàn)特定業(yè)務的具體功能。

3、自定義處理器MyNettyServerHandler

繼承SimpleChannelInboundHandler，該父類的channelRead0()方法可以接收客戶端的所有請求，并作出響應，輸出“Hello Netty”。

簡單講，Netty程序就是通過主程序類關聯(lián)自定義初始化器，然后在初始化器中加入Netty內(nèi)置處理器和自定義處理器，最后在自定義處理器中編寫處理特定需求的業(yè)務代碼。

三、代碼實例

1、maven中加入netty-all

<!-- https://mvnrepository.com/artifact/io.netty/netty-all -->
<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.68.Final</version>
</dependency>

2、主程序類MyNettyServerTest

package com.guor.demo.netty;

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

/**
 * 主程序類
 */
public class MyNettyServerTest {
    public static void main(String[] args) throws InterruptedException {
        /**
         * EventLoopGroup：事件循環(huán)組，是一個線程池，也是一個死循環(huán)，用于不斷地接收用戶請求；
         * serverGroup：用戶監(jiān)聽及建立連接，并把每一個連接抽象為一個channel，最后再將連接交給clientGroup處理；
         * clientGroup：真正的處理連接
         */
        EventLoopGroup serverGroup = new NioEventLoopGroup();
        EventLoopGroup clientGroup = new NioEventLoopGroup();
        try {
            // 服務端啟動時的初始化操作
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            // 1、將serverGroup和clientGroup注冊到服務端的Channel上；
            // 2、注冊一個服務端的初始化器MyNettyServerInitializer；
            // 3、該初始化器中的initChannel()方法會在連接被注冊到Channel后立刻執(zhí)行；
            // 5、最后將端口號綁定到8080；
            ChannelFuture channelFuture = serverBootstrap.group(serverGroup, clientGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new MyNettyServerInitializer()).bind(8080).sync();
            channelFuture.channel().closeFuture().sync();
        }catch (Exception e){
            System.out.println(e);
        }finally {
            serverGroup.shutdownGracefully();
            clientGroup.shutdownGracefully();
        }
    }
}

3、初始化器MyNettyServerInitializer

package com.guor.demo.netty;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpServerCodec;

/**
 * 初始化器
 */
public class MyNettyServerInitializer extends ChannelInitializer<SocketChannel> {

    // 連接被注冊到Channel后，立刻執(zhí)行此方法
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        // 加入netty提供的處理器
        pipeline.addLast("HttpServerCodec",new HttpServerCodec());
        // 增加自定義處理器MyNettyServerHandler，用于實際處理請求，并給出響應
        pipeline.addLast("MyNettyServerHandler",new MyNettyServerHandler());
    }
}

4、自定義處理器MyNettyServerHandler

package com.guor.demo.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.*;
import io.netty.util.CharsetUtil;

/**
 * 自定義處理器
 * Inbound代表"進入"的請求
 */
public class MyNettyServerHandler extends SimpleChannelInboundHandler<HttpObject> {

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, HttpObject httpObject) throws Exception {
        // 定義響應的內(nèi)容
        ByteBuf byteBuf = Unpooled.copiedBuffer("Hello Netty", CharsetUtil.UTF_8);
        DefaultFullHttpResponse defaultFullHttpResponse = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, byteBuf);
        defaultFullHttpResponse.headers().set(HttpHeaderNames.CONTENT_TYPE,"text/plain");
        byteBuf.readableBytes();
        // 等響應返回給客戶端
        channelHandlerContext.writeAndFlush(defaultFullHttpResponse);
    }
}

5、通過curl http://localhost:8080訪問Netty服務

此時，可能會提示curl不是內(nèi)部或外部命令，如何解決？

四、curl不是內(nèi)部或外部命令

1、下載curl for 64-bit

https://curl.se/windows/

2、將zip解壓到文件夾

3、在環(huán)境變量中配置

五、重寫SimpleChannelInboundHandler中一些重要的回調方法

1、重寫回調方法

package com.guor.demo.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.*;
import io.netty.util.CharsetUtil;

/**
 * 自定義處理器
 * Inbound代表"進入"的請求
 */
public class MyNettyServerHandler extends SimpleChannelInboundHandler<HttpObject> {

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, HttpObject httpObject) throws Exception {
        // 定義響應的內(nèi)容
        ByteBuf byteBuf = Unpooled.copiedBuffer("Hello Netty", CharsetUtil.UTF_8);
        DefaultFullHttpResponse defaultFullHttpResponse = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, byteBuf);
        defaultFullHttpResponse.headers().set(HttpHeaderNames.CONTENT_TYPE,"text/plain");
        byteBuf.readableBytes();
        // 等響應返回給客戶端
        channelHandlerContext.writeAndFlush(defaultFullHttpResponse);
    }

    /**
     * 增加新處理器時，觸發(fā)此方法
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception{
        System.out.println("1、handlerAdded、增加了新的處理器");
        super.handlerAdded(ctx);
    }

    /**
     * 當通道被注冊到一個事件循環(huán)組EventLoop上時，觸發(fā)此方法
     */
    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception{
        System.out.println("2、channelRegistered、通道被注冊");
        super.channelRegistered(ctx);
    }

    /**
     * 當遠端處于活躍狀態(tài)時（連接到了某個遠端，可以收發(fā)數(shù)據(jù)），執(zhí)行此方法
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception{
        System.out.println("3、channelActive、通道連接到了遠端，處于活躍狀態(tài)");
        super.channelActive(ctx);
    }

    /**
     * 當通道處于非活躍狀態(tài)時（與遠端斷開了連接），執(zhí)行此方法
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception{
        System.out.println("4、channelInactive、通道遠端斷開了連接，處于非活躍狀態(tài)");
        super.channelInactive(ctx);
    }

    /**
     * 當通道被取消了注冊時，執(zhí)行此方法
     */
    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception{
        System.out.println("5、channelUnregistered、通道被取消了注冊");
        super.channelUnregistered(ctx);
    }

    /**
     * 當程序發(fā)生異常，執(zhí)行此方法
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx,Throwable e) throws Exception{
        System.out.println("6、exceptionCaught、程序發(fā)生了異常");
        e.printStackTrace();
        ctx.close();
    }
}