Spring Boot整合Netty

概述

  Netty是由JBOSS提供的一个java开源框架，现为 Github上的独立项目。Netty提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。
  它是一个NIO客户端服务器框架，可以快速轻松地开发网络应用程序，极大地简化了网络编程，例如TCP和UDP套接字服务器开发。

整合

在pom.xml中添加依赖：

        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.42.Final</version>
        </dependency>

️以下使用时间服务器作为例子。

服务端实现

创建NettyServer类添加@Component注解交由容器管理

@Component
public class NettyServer {

    private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);

    // 服务端NIO线程组
    private final EventLoopGroup bossGroup = new NioEventLoopGroup();
    private final EventLoopGroup workGroup = new NioEventLoopGroup();

    public ChannelFuture start(String host, int port) {
        ChannelFuture channelFuture = null;
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            // 自定义服务处理
                            socketChannel.pipeline().addLast(new ServerHandler());
                        }
                    });
            // 绑定端口并同步等待
            channelFuture = bootstrap.bind(host, port).sync();
            logger.info("======Start Up Success!=========");
        } catch (Exception e) {
            e.printStackTrace();
        }
        return channelFuture;
    }

    public void close() {
        workGroup.shutdownGracefully();
        bossGroup.shutdownGracefully();
        logger.info("======Shutdown Netty Server Success!=========");
    }
}

创建服务处理类ServerHandler继承ChannelInboundHandlerAdapter类并重写channelRead、channelReadComplete、exceptionCaught方法。

public class ServerHandler extends ChannelInboundHandlerAdapter {

    /**
     * 客户端数据到来时触发
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("client request: " + buf.toString(CharsetUtil.UTF_8));
        SimpleDateFormat sf = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
        String callback = sf.format(new Date());
        ctx.write(Unpooled.copiedBuffer(callback.getBytes()));
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        // 将发送缓冲区的消息全部写到SocketChannel中
        ctx.flush();
    }

    /**
     * 发生异常时触发
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        // 释放与ChannelHandlerContext相关联的资源
        ctx.close();
    }
}

在SpringBoot启动类上实现CommandLineRunner接口。
注入容器中的NettyServer对象，在run方法中添加需要监听的地址以及端口开启服务。
获取当前线程钩子，使jvm关闭前服务同时关闭释放相关资源。

@SpringBootApplication
public class NioApplication implements CommandLineRunner {

    public static void main(String[] args) {
        SpringApplication.run(NioApplication.class, args);
    }

    @Resource
    private NettyServer nettyServer;

    @Override
    public void run(String... args) throws Exception {
        // 开启服务
        ChannelFuture future = nettyServer.start("localhost", 7070);
        // 在JVM销毁前关闭服务
        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                nettyServer.close();
            }
        });
        future.channel().closeFuture().sync();
    }
}

客户端实现

新建一个普通的项目并添加依赖netty-all-4.1.43.Final.jar（下载并解压后只需添加all-in-one文件夹内的netty-all-4.1.43.Final.jar）

创建NettyClient类作为客户端

public class NettyClient {

    private static final EventLoopGroup group = new NioEventLoopGroup();

    public static void main(String[] args) throws Exception {
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    // 自定义处理程序
                    socketChannel.pipeline().addLast(new ClientHandler());
                }
            });
            // 绑定端口并同步等待
            ChannelFuture channelFuture = bootstrap.connect("localhost", 7070).sync();
            channelFuture.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
    
}

创建处理程序类ClientHandler继承ChannelInboundHandlerAdapter类重写channelActive、channelRead、exceptionCaught方法。

public class ClientHandler extends ChannelInboundHandlerAdapter {

    /**
     * 连接到服务器时触发
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(Unpooled.copiedBuffer("current time", CharsetUtil.UTF_8));
    }

    /**
     * 消息到来时触发
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("Current Time: " + buf.toString(CharsetUtil.UTF_8));
    }

    /**
     * 发生异常时触发
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

}

测试

以上一个简单的服务端以及客户端就配置完成了，分别运行服务端程序以及客户端程序进行测试。

Server-1 Client-1

粘包处理

经过测试一个简单的时间服务器配置成功，现在修改客户端，使客户端能够同时发送大量数据，并进行测试。

修改客户端处理程序ClientHandler类中的channelActive方法，连续发送100条信息。

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        for (int i = 0; i < 100; i++) {
            ctx.write(Unpooled.copiedBuffer("current time", CharsetUtil.UTF_8));
        }
        ctx.flush();
    }

服务端接收如下所示：

Server-2

客户端接收如下所示：

Client-2

由图可知服务端只接收了2次，并且数据被连接在了一起，可知发生了粘包的情况。
下面我们使用netty提供的解码器来解决粘包问题。

修改发送的数据格式，在数据之间添加换行符 \n 进行间隔：

        //--------ClientHandler---------
        ctx.write(Unpooled.copiedBuffer("current time\n", CharsetUtil.UTF_8));
        //--------ServerHandler---------
        String callback = sf.format(new Date()) + "\n";

在之前配置的NettyServer类的服务处理程序之前添加解码器LineBasedFrameDecoder：

 bootstrap.group(bossGroup, workGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            // LineBasedFrameDecoder能够将接收到的数据在行尾进行拆分。
                            // 设置解码帧的最大长度，如果帧的长度超过此值抛出异常
                            socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
                            // 服务处理
                            socketChannel.pipeline().addLast(new ServerHandler());
                        }
                    });

同理客户端类NettyClient也同时添加解码器LineBasedFrameDecoder：

        bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
                    socketChannel.pipeline().addLast(new ClientHandler());
                }
            });

再次进行测试

Server-3 Client-3