0
0

Netty + SpringBoot 实现定时后端向前端推送数据

     本文介绍BIO，NIO，AIO 及如何使用Spring Boot集成Netty，实现后台向前端推送信息的功能。利用Spring Boot简化Netty的集成和配置。

1.BIO,NIO,AIO

BIO、NIO和AIO是Java编程语言中用于处理输入输出（IO）操作的三种不同的机制，它们分别代表 同步阻塞I/O，同步非阻塞I/O 和 异步非阻塞I/O。

1.1 BIO

BIO（Blocking IO） 是最传统的IO模型，也称为同步阻塞IO。它实现的是同步阻塞模型，即服务器实现模式为一个连接一个线程，即客户端有连接请求时服务器端就需要启动一个线程进行处理。如果这个连接不做任何事情会造成不必要的线程开销，并且线程在进行IO操作期间是被阻塞的，无法进行其他任务。在高并发环境下，BIO的性能较差，因为它需要为每个连接创建一个线程，而且线程切换开销较大，不过可以通过线程池机制改善。BIO适合一些简单的、低频的、短连接的通信场景，例如HTTP请求。

1.2 NIO

NIO是Java 1.4引入的新IO模型，也称为同步非阻塞IO，它提供了一种基于事件驱动的方式来处理I/O操作。

相比于传统的BIO模型，NIO采用了Channel、Buffer和Selector等组件，线程可以对某个IO事件进行监听，并继续执行其他任务，不需要阻塞等待。当IO事件就绪时，线程会得到通知，然后可以进行相应的操作，实现了非阻塞式的高伸缩性网络通信。在NIO模型中，数据总是从Channel读入Buffer，或者从Buffer写入Channel，这种模式提高了IO效率，并且可以充分利用系统资源。

NIO主要由三部分组成：选择器（Selector）、缓冲区（Buffer）和通道（Channel）。Channel是一个可以进行数据读写的对象，所有的数据都通过Buffer来处理，这种方式避免了直接将字节写入通道中，而是将数据写入包含一个或者多个字节的缓冲区。在多线程模式下，一个线程可以处理多个请求，这是通过将客户端的连接请求注册到多路复用器上，然后由多路复用器轮询到连接有I/O请求时进行处理。

对于NIO，如果从特性来看，它是非阻塞式IO，N是Non-Blocking的意思；如果从技术角度，NIO对于BIO来说是一个新技术，N的意思是New的意思。所以NIO也常常被称作Non-Blocking I/O或New I/O。

NIO适用于连接数目多且连接比较短（轻操作）的架构，例如聊天服务器、弹幕系统、服务器间通讯等。它通过引入非阻塞通道的概念，提高了系统的伸缩性和并发性能。同时，NIO的使用也简化了程序编写，提高了开发效率。

1.3 AIO

Java AIO（Asynchronous I/O）是Java提供的异步非阻塞IO编程模型，从Java 7版本开始支持，AIO又称NIO 2.0。

相比于NIO模型，AIO模型更进一步地实现了异步非阻塞IO，提高了系统的并发性能和伸缩性。在NIO模型中，虽然可以通过多路复用器处理多个连接请求，但仍需要在每个连接上进行读写操作，这仍然存在一定的阻塞。而在AIO模型中，所有的IO操作都是异步的，不会阻塞任何线程，可以更好地利用系统资源。

AIO模型有以下特性：

  1. 异步能力：AIO模型的最大特性是异步能力，对于socket和I/O操作都有效。读写操作都是异步的，完成后会自动调用回调函数。
  2. 回调函数：在AIO模型中，当一个异步操作完成后，会通知相关线程进行后续处理，这种处理方式称为“回调”。回调函数可以由开发者自行定义，用于处理异步操作的结果。
  3. 非阻塞：AIO模型实现了完全的异步非阻塞IO，不会阻塞任何线程，可以更好地利用系统资源。
  4. 高性能：由于AIO模型的异步能力和非阻塞特性，它可以更好地处理高并发、高伸缩性的网络通信场景，进一步提高系统的性能和效率。
  5. 操作系统支持：AIO模型需要操作系统的支持，因此在不同的操作系统上可能会有不同的表现。在Linux内核2.6版本之后增加了对真正异步IO的实现

2 Netty原理

2.1 Netty原理

    Netty基于Java NIO（非阻塞IO）实现，它采用事件驱动的编程模型，将IO操作抽象为事件，通过事件处理器来处理这些事件。Netty的主要组件包括：

Bootstrap：用于启动客户端和服务器的引导类
Channel：代表IO操作的通道，用于网络读写操作
ChannelHandler：用于处理IO事件的事件处理器
EventLoopGroup：用于处理IO操作的多线程事件循环组

3 Spring Boot集成Netty和Websocket

   在Spring Boot应用程序中，我们可以通过集成Netty，实现后台向前端推送信息的功能。首先，我们需要添加Netty依赖，然后在Spring Boot应用程序中创建一个NettyServer类，用于初始化Websocket通道。

1.引入依赖

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

2.创建 NettyConfig 配置管理所有管道

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

import java.util.concurrent.ConcurrentHashMap;

@SuppressWarnings("all")
public class NettyConfig {
    /**
     * 定义全局单例channel组 管理所有channel
     */
    private static volatile ChannelGroup channelGroup = null;

    /**
     * 存放请求ID与channel的对应关系
     */
    private static volatile ConcurrentHashMap<String, Channel> channelMap = null;

    /**
     * 定义两把锁
     */
    private static final Object lock1 = new Object();
    private static final Object lock2 = new Object();

    public static ChannelGroup getChannelGroup() {
        if (null == channelGroup) {
            synchronized (lock1) {
                if (null == channelGroup) {
                    channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
                }
            }
        }
        return channelGroup;
    }

    public static ConcurrentHashMap<String, Channel> getChannelMap() {
        if (null == channelMap) {
            synchronized (lock2) {
                if (null == channelMap) {
                    channelMap = new ConcurrentHashMap<>();
                }
            }
        }
        return channelMap;
    }

    public static Channel getChannel(String userId) {
        if (null == channelMap) {
            return getChannelMap().get(userId);
        }
        return channelMap.get(userId);
    }

}

3.创建MyChannelHandlerPool 通道组池

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

/**
 * MyChannelHandlerPool
 * 通道组池，管理所有websocket连接
 */
public class MyChannelHandlerPool {

    private MyChannelHandlerPool(){}

    public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

}

4.创建NettyServer 初始化Netty

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * NettyServer Netty服务器配置
 */
@Slf4j
@Component
@SuppressWarnings("all")
public class NettyServer {

    private String url = "/admin/socket";

    public NettyServer() {}

    public void start() throws Exception {
        // 主事件组
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        // 执行事件组
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.option(ChannelOption.SO_BACKLOG, 1024);
            int port = 12345;
            serverBootstrap.group(group, bossGroup) // 绑定线程池
                    .channel(NioServerSocketChannel.class) // 指定使用的channel
                    .localAddress(port)// 绑定监听端口
                    .childHandler(new ChannelInitializer<SocketChannel>() { // 绑定客户端连接时候触发操作
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            // 添加ObjectEncoder和ObjectDecoder来处理对象的序列化和反序列化
                            log.info("收到新连接");
                            //websocket协议本身是基于http协议的，所以这边也要使用http解编码器
                            ch.pipeline().addLast(new HttpServerCodec());
                            //以块的方式来写的处理器
                            ch.pipeline().addLast(new ChunkedWriteHandler());
                            ch.pipeline().addLast(new HttpObjectAggregator(8192));
                            ch.pipeline().addLast(new WebSocketHandler());
                            ch.pipeline().addLast(new WebSocketServerProtocolHandler(url, null, true, 65536 * 10));
                        }
                    });
            // 绑定端口并同步等待直到绑定完成
            ChannelFuture future = serverBootstrap.bind().sync();
            log.info(NettyServer.class.getName() + "启动正在监听： " + future.channel().localAddress());
            // 等待服务器通道关闭
            future.channel().closeFuture().sync();
        } finally {
            // 释放线程池资源
            group.shutdownGracefully().sync();
            bossGroup.shutdownGracefully().sync();
        }
    }

}

5. 创建 WebSocketHandler 执行任务

@Slf4j
@Component
@SuppressWarnings("all")
public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    public static final String NETTY_START = "Netty-start";

    public WebSocketHandler() {}

    private ScheduledFuture<?> sendDataTask;

    @Autowired
    private DeviceLevelFourService deviceLevelFourService;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        //添加到channelGroup通道组
        MyChannelHandlerPool.channelGroup.add(ctx.channel());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("与客户端断开连接，通道关闭！");
        //添加到channelGroup 通道组
        MyChannelHandlerPool.channelGroup.remove(ctx.channel());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 首次连接是FullHttpRequest，处理参数
        if (msg instanceof FullHttpRequest) {
            FullHttpRequest request = (FullHttpRequest) msg;
            String uri = request.uri();
            Map<String, String> paramMap = getUrlParams(uri);
            log.info("接收到的参数是：" + JSON.toJSONString(paramMap));
            // 如果url包含参数，需要处理
            if (uri.contains("?")) {
                String newUri = uri.substring(0, uri.indexOf("?"));
                log.info(newUri);
                request.setUri(newUri);
            }
            // 当连接建立时，启动定时任务
            sendDataTask = ctx.channel().eventLoop().scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    try {
                        if (deviceLevelFourService == null) {
                            deviceLevelFourService = SpringContextUtil.getBean(DeviceLevelFourServiceImpl.class);
                        }
                        // 此处可以接收路径参数 ，直接获取前端传递参数
                        // "ws://localhost:12345/admin/socket?id=1"
                        String deviceId = paramMap.get("id");
                        /**
                         * -------此处为自己的数据---------
                         */
                        // 调用service 得到前端需要的数据，用JSON工具类转换推送到前端
                        List<DeviceNettyData> deviceNettyData = deviceLevelFourService.handlerDeviceData(Long.parseLong(deviceId));
                        String json = JSON.toJSONString(deviceNettyData, SerializerFeature.WriteMapNullValue);
                        log.info(json);
                        // 将 JSON 字符串封装为 TextWebSocketFrame
                        TextWebSocketFrame frameNetty = new TextWebSocketFrame(json);
                        ctx.writeAndFlush(frameNetty); // 发送 WebSocket 帧
                    } catch (Exception e) {
                        log.error(e.getMessage(), e);
                    }
                }
            }, 0, 30, TimeUnit.SECONDS); // 立即开始，每30秒发送一次
            // 调用父类方法，处理下一个handler
            super.channelRead(ctx, request);
        } else if (msg instanceof TextWebSocketFrame frame) {
            // 正常的TEXT消息类型
            sendAllMessage(frame.text());
            // 继续传递给后续handler
            super.channelRead(ctx, frame);
        } else {
            // 如果消息类型不匹配，记录警告或处理异常情况
            log.error("未处理的消息类型：" + msg.getClass());
            super.channelRead(ctx, msg); // 仍然传递给后续处理
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {
        log.info(channelHandlerContext.name());
    }

    //读取完成刷新
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    //异常则关闭ChannelHandlerContext连接
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // Close the connection when an exception is raised.
        cause.printStackTrace();
        ctx.close();
    }

    private void sendAllMessage(String message){
        //收到信息后，群发给所有channel
        MyChannelHandlerPool.channelGroup.writeAndFlush( new TextWebSocketFrame(message));
    }

    private static Map<String ,String> getUrlParams(String url){
        Map<String,String> map = new HashMap<>();
        url = url.replace("?",";");
        if (!url.contains(";")){
            return map;
        }
        if (url.split(";").length > 0){
            String[] arr = url.split(";")[1].split("&");
            for (String s : arr){
                String key = s.split("=")[0];
                String value = s.split("=")[1];
                map.put(key,value);
            }
            return  map;

        }else{
            return map;
        }
    }

}

6.创建NettyServerRunner 用来使用新线程调用NetterServer

@Slf4j
public class NettyServerRunner implements Runnable {
    @Override
    public void run() {
        try {
            new NettyServer().start();
        } catch (Exception e) {
            // 使用Logger进行日志记录
            log.error(e.getMessage(), e);
        }
    }

}
  1. 最后可以在随意注入到spring容器类中，在项目启动时候调用，也可以在其他访问接口事件调用
// 方式一
@PostConstruct
public void init(){
    // 在新的线程中运行Netty服务器
    Thread thread = new Thread(new NettyServerRunner());
    thread.start();
}

// 方式二调用
@GetMapping("/test/{id}")
public R getDevice(@PathVariable Long id){
    // 在新的线程中运行Netty服务器
    Thread thread = new Thread(new NettyServerRunner());
    thread.start();
    return deviceService.getDeviceAlarmCount(deviceId);
}

注意：

    使用新线程时候，spring容器注入的对象为空，容易产生空指针异常，可以借鉴WebSocketHandler 类中方法，重新从spring容器中获取需要的对象。
标签： spring boot 后端
本文转载自: https://blog.csdn.net/qq_61223023/article/details/143864771
版权归原作者 xiaohong_strive 所有， 如有侵权，请联系我们删除。
登录后发布评论

“Netty + SpringBoot 实现定时后端向前端推送数据”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？

文章导航