工作場景:使用Netty長連接實(shí)時(shí)獲取第三方接口的車輛定位數(shù)據(jù)

開發(fā)環(huán)境:JDK8

Netty基本介紹

一、什么是Netty

Netty是由JBOSS提供的一個(gè)Java開源框架，現(xiàn)為Github上的獨(dú)立項(xiàng)目。它是一個(gè)異步的、基于事件驅(qū)動(dòng)的網(wǎng)絡(luò)應(yīng)用框架，用以快速開發(fā)高性能、高可靠性的網(wǎng)絡(luò)IO程序。Netty主要針對(duì)在TCP協(xié)議下，面向Clients端的高并發(fā)應(yīng)用，或者Peer-to-Peer場景下的大量數(shù)據(jù)持續(xù)傳輸?shù)膽?yīng)用。

Netty提供了一套完整的API，用于處理網(wǎng)絡(luò)IO操作，如TCP和UDP套接字。它封裝了底層的網(wǎng)絡(luò)編程細(xì)節(jié)，使得開發(fā)者可以更加專注于業(yè)務(wù)邏輯的實(shí)現(xiàn)。Netty使用了一種高效的線程模型，可以處理大量的并發(fā)連接，并且具有很好的伸縮性。

Netty在多個(gè)領(lǐng)域都有廣泛的應(yīng)用，如RPC框架、游戲行業(yè)、大數(shù)據(jù)領(lǐng)域等。它支持多種傳輸類型和協(xié)議，如阻塞和非阻塞、基于BIO和NIO的UDP傳輸、本地傳輸（in-VM傳輸）、HTTP通道等。同時(shí)，Netty還提供了豐富的編解碼器，用于處理各種協(xié)議的編解碼操作。

Netty的整體結(jié)構(gòu)包括核心層和協(xié)議支持層。核心層提供了底層網(wǎng)絡(luò)通信的通用抽象和實(shí)現(xiàn)，包括可擴(kuò)展的事件模型、通用的通信API、支持零拷貝的ByteBuf等。協(xié)議支持層則覆蓋了主流協(xié)議的編解碼實(shí)現(xiàn)，如HTTP、SSL、Protobuf等。

總的來說，Netty是一個(gè)功能強(qiáng)大、易于使用的網(wǎng)絡(luò)應(yīng)用框架，它可以幫助開發(fā)者快速構(gòu)建高性能、高可靠性的網(wǎng)絡(luò)應(yīng)用程序。

二、Netty核心組件

Netty的核心組件主要包括以下幾個(gè)部分：

• Channels：Channel是Netty網(wǎng)絡(luò)通信的抽象，用于進(jìn)行I/O操作。它可以被看作是Java NIO的一個(gè)基本抽象，代表了與硬件設(shè)備、文件、網(wǎng)絡(luò)socket等實(shí)體的開放連接，或者是一個(gè)能夠完成讀、寫等I/O操作的程序。Channel可以被打開或關(guān)閉，連接或斷開。
• Callbacks（回調(diào)）：Callback是一個(gè)方法，它是提供給另一個(gè)方法的引用，使得另一個(gè)方法可以在適當(dāng)?shù)臅r(shí)候回過頭來調(diào)用這個(gè)Callback方法。Callback在很多編程情形中被廣泛使用，是用于通知相關(guān)方某個(gè)操作已經(jīng)完成最常用的方法之一。
• Futures：在Netty中，F(xiàn)utures用于異步I/O操作的結(jié)果。當(dāng)一個(gè)異步操作開始時(shí)，會(huì)立即返回一個(gè)Future，這個(gè)Future會(huì)在操作完成時(shí)得到結(jié)果或者異常。
• Handlers：Handlers是Netty中處理I/O事件或攔截I/O操作的組件。Netty提供了許多內(nèi)置的Handler，如ChannelInboundHandler、ChannelOutboundHandler等，這些Handler可以處理各種I/O事件，如連接建立、數(shù)據(jù)接收、異常處理等。
• Bootstrap與ServerBootstrap：Bootstrap和ServerBootstrap是Netty程序的引導(dǎo)類，主要用于配置各種參數(shù)并啟動(dòng)整個(gè)Netty服務(wù)。它們都繼承自AbstractBootstrap抽象類，不同的是，Bootstrap用于客戶端引導(dǎo)，而ServerBootstrap用于服務(wù)端引導(dǎo)。
• EventLoopGroup：EventLoopGroup可以理解為一個(gè)線程池，用于處理I/O操作。在服務(wù)端程序中，一般會(huì)綁定兩個(gè)EventLoopGroup，一個(gè)用于處理Accept事件（即新的連接請(qǐng)求），另一個(gè)用于處理讀寫事件。

以上這些組件共同構(gòu)成了Netty的核心框架，使得開發(fā)者可以更加專注于業(yè)務(wù)邏輯的實(shí)現(xiàn)，而無需過多關(guān)心底層的網(wǎng)絡(luò)通信細(xì)節(jié)。

三、SpringBoot與Netty整合

1. 添加依賴

在SpringBoot項(xiàng)目的pom.xml文件中，我們需要添加Netty的依賴。Netty的官方Maven倉庫地址為：https://mvnrepository.com/artifact/io.netty/netty-all

    <dependencies>
        <!-- mybatis-plus -->
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-boot-starter</artifactId>
        </dependency>

        <!-- Mysql -->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
        </dependency>

        <!--   數(shù)據(jù)源     -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid-spring-boot-starter</artifactId>
        </dependency>

        <!--   netty     -->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
        </dependency>

        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-pool2</artifactId>
        </dependency>

        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-boot-starter</artifactId>
        </dependency>

        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>dynamic-datasource-spring-boot-starter</artifactId>
            <version>${mybatis-plus.version}</version>
        </dependency>

    </dependencies>

2.創(chuàng)建Netty服務(wù)端

@Component
public class NettyServer {

    //負(fù)責(zé)處理接受進(jìn)來的鏈接
    private EventLoopGroup bossGroup;
    //負(fù)責(zé)處理已經(jīng)被接收的連接上的I/O操作
    private EventLoopGroup workerGroup;
    //在這個(gè)場景中，它表示服務(wù)器的綁定操作的結(jié)果
    private ChannelFuture future;

    @PostConstruct
    public void startServer() throws Exception {
        bossGroup = new NioEventLoopGroup();
        workerGroup = new NioEventLoopGroup();
        try {
            //創(chuàng)建ServerBootstrap,這個(gè)類封裝了服務(wù)器端的網(wǎng)絡(luò)配置，使得我們可以輕松地設(shè)置服務(wù)器參數(shù)
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new NettyServerInitializer());

            // 綁定端口并開始接受進(jìn)來的連接
            future = bootstrap.bind(7000).sync();

            // 等待服務(wù)器套接字關(guān)閉
            future.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }

    @PreDestroy
    public void stopServer() {
        if (future != null && !future.isDone()) {
            future.cancel(true);
        }
        workerGroup.shutdownGracefully();
        bossGroup.shutdownGracefully();
    }
}

代碼解析

類注解

• @Component: 這是Spring框架的注解，表示這個(gè)類是一個(gè)組件，Spring會(huì)掃描到這個(gè)類并將其作為Bean注冊(cè)到Spring容器中。因此，這個(gè)類可以被其他Spring管理的Bean自動(dòng)裝配（如果需要的話）。

類成員變量

• bossGroup 和 workerGroup: 這兩個(gè)是EventLoopGroup的實(shí)例，用于處理網(wǎng)絡(luò)事件。bossGroup 主要負(fù)責(zé)接收進(jìn)來的連接，而 workerGroup 負(fù)責(zé)處理已經(jīng)被接收的連接上的I/O操作。
• future: 這是一個(gè)ChannelFuture的實(shí)例，代表了一個(gè)異步的I/O操作的結(jié)果。在這個(gè)場景中，它表示服務(wù)器的綁定操作的結(jié)果。

startServer 方法

• 該方法使用@PostConstruct注解，這意味著當(dāng)Spring容器實(shí)例化這個(gè)Bean并完成依賴注入后，會(huì)自動(dòng)調(diào)用這個(gè)方法。
• 在這個(gè)方法中，首先創(chuàng)建了兩個(gè)NioEventLoopGroup實(shí)例，一個(gè)用于boss，一個(gè)用于worker。
• 然后，使用ServerBootstrap類來配置和啟動(dòng)服務(wù)器。這個(gè)類封裝了服務(wù)器端的網(wǎng)絡(luò)配置，使得我們可以輕松地設(shè)置服務(wù)器參數(shù)。
• 通過group方法設(shè)置boss和worker的EventLoopGroup。
• 通過channel方法指定使用NioServerSocketChannel作為服務(wù)器的通道實(shí)現(xiàn)。
• 通過childHandler方法設(shè)置一個(gè)新的連接被接受后如何處理。這里使用了NettyServerInitializer（這個(gè)類沒有在提供的代碼段中定義，但我們可以假設(shè)它是一個(gè)ChannelInitializer的實(shí)現(xiàn)，用于配置新的Channel）。
• 使用bind方法綁定服務(wù)器到指定的端口（這里是7000），并使用sync方法阻塞直到綁定完成。
• 最后，使用closeFuture().sync()方法阻塞當(dāng)前線程，直到服務(wù)器套接字關(guān)閉。
• 在finally塊中，無論是否發(fā)生異常，都會(huì)優(yōu)雅地關(guān)閉EventLoopGroup。

stopServer 方法

• 該方法使用@PreDestroy注解，意味著當(dāng)Spring容器銷毀這個(gè)Bean之前，會(huì)自動(dòng)調(diào)用這個(gè)方法。
• 在這個(gè)方法中，首先檢查future是否已經(jīng)完成（即服務(wù)器是否已經(jīng)關(guān)閉）。如果沒有，就調(diào)用cancel(true)方法來嘗試取消這個(gè)操作。但是，需要注意的是，這里的cancel可能并不總是能立即停止服務(wù)器，它更多的是嘗試停止服務(wù)器，而不是強(qiáng)制停止。

3.創(chuàng)建字符解析器,用于解析收到的消息

public class NettyServerInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch){
        ChannelPipeline pipeline = ch.pipeline();

        // 添加一個(gè)字符串解碼器，用于將接收到的ByteBuf轉(zhuǎn)換成字符串
        // 這里假設(shè)使用的是UTF-8字符集
        pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));

        // 添加一個(gè)字符串編碼器，用于將發(fā)送的字符串轉(zhuǎn)換成ByteBuf
        // 這樣服務(wù)器發(fā)送字符串時(shí)，客戶端可以直接接收到字符串
        pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));

        // 添加自定義的ChannelInboundHandlerAdapter來處理業(yè)務(wù)邏輯
        pipeline.addLast("handler", new MyChannelHandler());
    }
}

這段代碼定義了一個(gè)NettyServerInitializer類，它繼承自ChannelInitializer<SocketChannel>，并覆蓋了initChannel方法。在Netty中，ChannelInitializer是一個(gè)特殊的處理器，它的主要目的是幫助用戶配置一個(gè)新的Channel的ChannelPipeline。當(dāng)一個(gè)新的連接被接受時(shí)，Netty會(huì)自動(dòng)調(diào)用ChannelInitializer的initChannel方法來設(shè)置這個(gè)新連接的ChannelPipeline。

具體來說，initChannel方法會(huì)在以下情況下被調(diào)用：

• 當(dāng)ServerBootstrap的bind方法被調(diào)用并成功綁定到某個(gè)端口后，開始監(jiān)聽傳入的連接。
• 一旦有客戶端連接到服務(wù)器，ServerBootstrap會(huì)接受這個(gè)連接，并創(chuàng)建一個(gè)新的SocketChannel來表示這個(gè)連接。
• 對(duì)于這個(gè)新的SocketChannel，Netty會(huì)調(diào)用之前設(shè)置的ChannelInitializer（在這個(gè)例子中是NettyServerInitializer）的initChannel方法。
• initChannel方法內(nèi)部會(huì)配置這個(gè)新SocketChannel的ChannelPipeline，添加解碼器、編碼器、業(yè)務(wù)處理器等。
• 一旦initChannel方法執(zhí)行完畢，這個(gè)ChannelInitializer的使命就完成了，并且會(huì)從ChannelPipeline中移除自身，因?yàn)樗回?fù)責(zé)初始化工作，不參與后續(xù)的數(shù)據(jù)處理。

所以，總結(jié)來說，NettyServerInitializer的initChannel方法會(huì)在一個(gè)新的客戶端連接被服務(wù)器接受時(shí)運(yùn)行，用于初始化這個(gè)新連接的ChannelPipeline。

4.創(chuàng)建Handler處理接受到的消息

public class MyChannelHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 在這里處理接收到的數(shù)據(jù)


        System.out.println("msg = " + msg);
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 在這里處理異常
    }
}

四、開發(fā)中遇到的問題(暫未解決)

1.字符解析器定義之后,接收到的消息仍然亂碼

2.項(xiàng)目啟動(dòng)后,可以訪問netty的端口號(hào),但是訪問不了項(xiàng)目的端口號(hào)(已解決)

// future.channel().closeFuture().sync();把這段代碼屏蔽就可以

到此這篇關(guān)于SpringBoot整合Netty服務(wù)端的實(shí)現(xiàn)示例的文章就介紹到這了,更多相關(guān)SpringBoot整合Netty服務(wù)端內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論