SpringBoot整合Netty服务端的实现示例

工作场景:使用Netty长连接实时获取第三方接口的车辆定位数据

开发环境:JDK8

Netty基本介绍

一、什么是Netty

Netty是由JBOSS提供的一个Java开源框架，现为Github上的独立项目。它是一个异步的、基于事件驱动的网络应用框架，用以快速开发高性能、高可靠性的网络IO程序。Netty主要针对在TCP协议下，面向Clients端的高并发应用，或者Peer-to-Peer场景下的大量数据持续传输的应用。

Netty提供了一套完整的API，用于处理网络IO操作，如TCP和UDP套接字。它封装了底层的网络编程细节，使得开发者可以更加专注于业务逻辑的实现。Netty使用了一种高效的线程模型，可以处理大量的并发连接，并且具有很好的伸缩性。

Netty在多个领域都有广泛的应用，如RPC框架、游戏行业、大数据领域等。它支持多种传输类型和协议，如阻塞和非阻塞、基于BIO和NIO的UDP传输、本地传输（in-VM传输）、HTTP通道等。同时，Netty还提供了丰富的编解码器，用于处理各种协议的编解码操作。

Netty的整体结构包括核心层和协议支持层。核心层提供了底层网络通信的通用抽象和实现，包括可扩展的事件模型、通用的通信API、支持零拷贝的ByteBuf等。协议支持层则覆盖了主流协议的编解码实现，如HTTP、SSL、Protobuf等。

总的来说，Netty是一个功能强大、易于使用的网络应用框架，它可以帮助开发者快速构建高性能、高可靠性的网络应用程序。

二、Netty核心组件

Netty的核心组件主要包括以下几个部分：

• Channels：Channel是Netty网络通信的抽象，用于进行I/O操作。它可以被看作是Java NIO的一个基本抽象，代表了与硬件设备、文件、网络socket等实体的开放连接，或者是一个能够完成读、写等I/O操作的程序。Channel可以被打开或关闭，连接或断开。
• Callbacks（回调）：Callback是一个方法，它是提供给另一个方法的引用，使得另一个方法可以在适当的时候回过头来调用这个Callback方法。Callback在很多编程情形中被广泛使用，是用于通知相关方某个操作已经完成最常用的方法之一。
• Futures：在Netty中，Futures用于异步I/O操作的结果。当一个异步操作开始时，会立即返回一个Future，这个Future会在操作完成时得到结果或者异常。
• Handlers：Handlers是Netty中处理I/O事件或拦截I/O操作的组件。Netty提供了许多内置的Handler，如ChannelInboundHandler、ChannelOutboundHandler等，这些Handler可以处理各种I/O事件，如连接建立、数据接收、异常处理等。
• Bootstrap与ServerBootstrap：Bootstrap和ServerBootstrap是Netty程序的引导类，主要用于配置各种参数并启动整个Netty服务。它们都继承自AbstractBootstrap抽象类，不同的是，Bootstrap用于客户端引导，而ServerBootstrap用于服务端引导。
• EventLoopGroup：EventLoopGroup可以理解为一个线程池，用于处理I/O操作。在服务端程序中，一般会绑定两个EventLoopGroup，一个用于处理Accept事件（即新的连接请求），另一个用于处理读写事件。

以上这些组件共同构成了Netty的核心框架，使得开发者可以更加专注于业务逻辑的实现，而无需过多关心底层的网络通信细节。

三、SpringBoot与Netty整合

1. 添加依赖

在SpringBoot项目的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>

        <!--   数据源     -->
        <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.创建Netty服务端

@Component
public class NettyServer {

    //负责处理接受进来的链接
    private EventLoopGroup bossGroup;
    //负责处理已经被接收的连接上的I/O操作
    private EventLoopGroup workerGroup;
    //在这个场景中，它表示服务器的绑定操作的结果
    private ChannelFuture future;

    @PostConstruct
    public void startServer() throws Exception {
        bossGroup = new NioEventLoopGroup();
        workerGroup = new NioEventLoopGroup();
        try {
            //创建ServerBootstrap,这个类封装了服务器端的网络配置，使得我们可以轻松地设置服务器参数
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new NettyServerInitializer());

            // 绑定端口并开始接受进来的连接
            future = bootstrap.bind(7000).sync();

            // 等待服务器套接字关闭
            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框架的注解，表示这个类是一个组件，Spring会扫描到这个类并将其作为Bean注册到Spring容器中。因此，这个类可以被其他Spring管理的Bean自动装配（如果需要的话）。

类成员变量

• bossGroup 和 workerGroup: 这两个是EventLoopGroup的实例，用于处理网络事件。bossGroup 主要负责接收进来的连接，而 workerGroup 负责处理已经被接收的连接上的I/O操作。
• future: 这是一个ChannelFuture的实例，代表了一个异步的I/O操作的结果。在这个场景中，它表示服务器的绑定操作的结果。

startServer 方法

• 该方法使用@PostConstruct注解，这意味着当Spring容器实例化这个Bean并完成依赖注入后，会自动调用这个方法。
• 在这个方法中，首先创建了两个NioEventLoopGroup实例，一个用于boss，一个用于worker。
• 然后，使用ServerBootstrap类来配置和启动服务器。这个类封装了服务器端的网络配置，使得我们可以轻松地设置服务器参数。
• 通过group方法设置boss和worker的EventLoopGroup。
• 通过channel方法指定使用NioServerSocketChannel作为服务器的通道实现。
• 通过childHandler方法设置一个新的连接被接受后如何处理。这里使用了NettyServerInitializer（这个类没有在提供的代码段中定义，但我们可以假设它是一个ChannelInitializer的实现，用于配置新的Channel）。
• 使用bind方法绑定服务器到指定的端口（这里是7000），并使用sync方法阻塞直到绑定完成。
• 最后，使用closeFuture().sync()方法阻塞当前线程，直到服务器套接字关闭。
• 在finally块中，无论是否发生异常，都会优雅地关闭EventLoopGroup。

stopServer 方法

• 该方法使用@PreDestroy注解，意味着当Spring容器销毁这个Bean之前，会自动调用这个方法。
• 在这个方法中，首先检查future是否已经完成（即服务器是否已经关闭）。如果没有，就调用cancel(true)方法来尝试取消这个操作。但是，需要注意的是，这里的cancel可能并不总是能立即停止服务器，它更多的是尝试停止服务器，而不是强制停止。

3.创建字符解析器,用于解析收到的消息

public class NettyServerInitializer extends ChannelInitializer<SocketChannel> {

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

        // 添加一个字符串解码器，用于将接收到的ByteBuf转换成字符串
        // 这里假设使用的是UTF-8字符集
        pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));

        // 添加一个字符串编码器，用于将发送的字符串转换成ByteBuf
        // 这样服务器发送字符串时，客户端可以直接接收到字符串
        pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));

        // 添加自定义的ChannelInboundHandlerAdapter来处理业务逻辑
        pipeline.addLast("handler", new MyChannelHandler());
    }
}

这段代码定义了一个NettyServerInitializer类，它继承自ChannelInitializer<SocketChannel>，并覆盖了initChannel方法。在Netty中，ChannelInitializer是一个特殊的处理器，它的主要目的是帮助用户配置一个新的Channel的ChannelPipeline。当一个新的连接被接受时，Netty会自动调用ChannelInitializer的initChannel方法来设置这个新连接的ChannelPipeline。

具体来说，initChannel方法会在以下情况下被调用：

• 当ServerBootstrap的bind方法被调用并成功绑定到某个端口后，开始监听传入的连接。
• 一旦有客户端连接到服务器，ServerBootstrap会接受这个连接，并创建一个新的SocketChannel来表示这个连接。
• 对于这个新的SocketChannel，Netty会调用之前设置的ChannelInitializer（在这个例子中是NettyServerInitializer）的initChannel方法。
• initChannel方法内部会配置这个新SocketChannel的ChannelPipeline，添加解码器、编码器、业务处理器等。
• 一旦initChannel方法执行完毕，这个ChannelInitializer的使命就完成了，并且会从ChannelPipeline中移除自身，因为它只负责初始化工作，不参与后续的数据处理。

所以，总结来说，NettyServerInitializer的initChannel方法会在一个新的客户端连接被服务器接受时运行，用于初始化这个新连接的ChannelPipeline。

4.创建Handler处理接受到的消息

public class MyChannelHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 在这里处理接收到的数据


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

四、开发中遇到的问题(暂未解决)

1.字符解析器定义之后,接收到的消息仍然乱码

2.项目启动后,可以访问netty的端口号,但是访问不了项目的端口号(已解决)

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

到此这篇关于SpringBoot整合Netty服务端的实现示例的文章就介绍到这了,更多相关SpringBoot整合Netty服务端内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

最新评论