利用Netty+SpringBoot实现定时后端向前端推送数据
作者：xiaohong_strive
这篇文章主要介绍了BIO、NIO、AIO三种Java IO模型,并探讨了如何使用Spring Boot集成Netty实现后台向前端推送信息的功能,文中通过代码介绍的非常详细,需要的朋友可以参考下
本文介绍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模型有以下特性：
- 异步能力：AIO模型的最大特性是异步能力，对于socket和I/O操作都有效。读写操作都是异步的，完成后会自动调用回调函数。
- 回调函数：在AIO模型中，当一个异步操作完成后，会通知相关线程进行后续处理，这种处理方式称为“回调”。回调函数可以由开发者自行定义，用于处理异步操作的结果。
- 非阻塞：AIO模型实现了完全的异步非阻塞IO，不会阻塞任何线程，可以更好地利用系统资源。
- 高性能：由于AIO模型的异步能力和非阻塞特性，它可以更好地处理高并发、高伸缩性的网络通信场景，进一步提高系统的性能和效率。
- 操作系统支持：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);
}
}
}7. 最后可以在随意注入到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容器中获取需要的对象。
总结
到此这篇关于利用Netty+SpringBoot实现定时后端向前端推送数据的文章就介绍到这了,更多相关SpringBoot定时后端向前端推送数据内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！