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