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

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

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);
	}


}

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);


}

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();
        }
    }


}

@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;
		}
	}



}

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

}

// 方式一
@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);
}