<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的對應(yīng)關(guān)系
	 */
	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服務(wù)器配置
 */
@Slf4j
@Component
@SuppressWarnings("all")
public class NettyServer {

	private String url = "/admin/socket";

	public NettyServer() {}

    public void start() throws Exception {
		// 主事件組
        EventLoopGroup bossGroup = new NioEventLoopGroup();
		// 執(zhí)行事件組
        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)// 綁定監(jiān)聽端口
                    .childHandler(new ChannelInitializer<SocketChannel>() { // 綁定客戶端連接時候觸發(fā)操作
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            // 添加ObjectEncoder和ObjectDecoder來處理對象的序列化和反序列化
							log.info("收到新連接");
                            //websocket協(xié)議本身是基于http協(xié)議的，所以這邊也要使用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() + "啟動正在監(jiān)聽： " + future.channel().localAddress());
			// 等待服務(wù)器通道關(guān)閉
			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("與客戶端斷開連接，通道關(guān)閉！");
		//添加到channelGroup 通道組
		MyChannelHandlerPool.channelGroup.remove(ctx.channel());
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		// 首次連接是FullHttpRequest，處理參數(shù)
		if (msg instanceof FullHttpRequest) {
			FullHttpRequest request = (FullHttpRequest) msg;
			String uri = request.uri();
			Map<String, String> paramMap = getUrlParams(uri);
			log.info("接收到的參數(shù)是：" + JSON.toJSONString(paramMap));
			// 如果url包含參數(shù)，需要處理
			if (uri.contains("?")) {
				String newUri = uri.substring(0, uri.indexOf("?"));
				log.info(newUri);
				request.setUri(newUri);
			}
			// 當(dāng)連接建立時，啟動定時任務(wù)
			sendDataTask = ctx.channel().eventLoop().scheduleAtFixedRate(new Runnable() {
				@Override
				public void run() {
					try {
						if (deviceLevelFourService == null) {
							deviceLevelFourService = SpringContextUtil.getBean(DeviceLevelFourServiceImpl.class);
						}
                        // 此處可以接收路徑參數(shù) ，直接獲取前端傳遞參數(shù)
                        // "ws://localhost:12345/admin/socket?id=1"
						String deviceId = paramMap.get("id");
                        /**
                         * -------此處為自己的數(shù)據(jù)---------
                         */
                        // 調(diào)用service 得到前端需要的數(shù)據(jù)，用JSON工具類轉(zhuǎn)換推送到前端
						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); // 發(fā)送 WebSocket 幀
					} catch (Exception e) {
						log.error(e.getMessage(), e);
					}
				}
			}, 0, 30, TimeUnit.SECONDS); // 立即開始，每30秒發(fā)送一次
			// 調(diào)用父類方法，處理下一個handler
			super.channelRead(ctx, request);
		} else if (msg instanceof TextWebSocketFrame frame) {
			// 正常的TEXT消息類型
			sendAllMessage(frame.text());
			// 繼續(xù)傳遞給后續(xù)handler
			super.channelRead(ctx, frame);
		} else {
			// 如果消息類型不匹配，記錄警告或處理異常情況
			log.error("未處理的消息類型：" + msg.getClass());
			super.channelRead(ctx, msg); // 仍然傳遞給后續(xù)處理
		}
	}

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

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

	//異常則關(guān)閉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){
		//收到信息后，群發(fā)給所有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服務(wù)器
	Thread thread = new Thread(new NettyServerRunner());
	thread.start();
}




// 方式二調(diào)用
@GetMapping("/test/{id}")
public R getDevice(@PathVariable Long id){
	// 在新的線程中運行Netty服務(wù)器
	Thread thread = new Thread(new NettyServerRunner());
	thread.start();
	return deviceService.getDeviceAlarmCount(deviceId);
}