SpringBoot+Netty+WebSocket，Channel和用户绑定关系，实现后台向前端推送信息

1.pom依赖

 		<!-- WebSocket -->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.36.Final</version>
        </dependency>

2.springboot启动类

@SpringBootApplication
public class SpringBootApplication {

	public static void main(String[] args) {
		SpringApplication.run(SpringBootApplication.class, args);

		new Thread(new ClientsCheck()).start();  // 客户端检查

		try {
			new NettyServer(12345).start();
		} catch (Exception e) {
			System.out.println("NettyServerError:" + e.getMessage());
		}
	}

}

3.Netty服务器配置

public class NettyServer {

	private final int port;

	public NettyServer(int port) {
		this.port = port;
	}

	public void start() throws Exception {
		EventLoopGroup bossGroup = new NioEventLoopGroup();
		EventLoopGroup group = new NioEventLoopGroup();
		try {
			ServerBootstrap sb = new ServerBootstrap();
			sb.option(ChannelOption.SO_BACKLOG, 1024);
			sb.group(group, bossGroup) // 绑定线程池
					.channel(NioServerSocketChannel.class) // 指定使用的channel
					.localAddress(this.port)// 绑定监听端口
					.childHandler(new ChannelInitializer<SocketChannel>() { // 绑定客户端连接时候触发操作
						@Override
						protected void initChannel(SocketChannel ch) throws Exception {
							System.out.println("收到新连接");
							// 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("/ws", null, true, 65536 * 10));
						}
					});
			ChannelFuture cf = sb.bind().sync(); // 服务器异步创建绑定
			System.out.println(NettyServer.class + " 启动正在监听： " + cf.channel().localAddress());
			cf.channel().closeFuture().sync(); // 关闭服务器通道
		} finally {
			group.shutdownGracefully().sync(); // 释放线程池资源
			bossGroup.shutdownGracefully().sync();
		}
	}
	
}

4.ChannelHandlerPool

通道组池，管理所有websocket连接

public class ChannelHandlerPool {

	public ChannelHandlerPool() {
	}

	public static Set<Channel> channelGroup = Collections.synchronizedSet(new HashSet<>());

}

5.WebSocketHandler

public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

	private final String USER = "user";
	private final AttributeKey<String> key = AttributeKey.valueOf(USER);

	@Override
	protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) {

	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		System.out.println("与客户端建立连接，通道开启！");
	}

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

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		// 首次连接是FullHttpRequest，处理参数
		if (null != msg && msg instanceof FullHttpRequest) {
			FullHttpRequest request = (FullHttpRequest) msg;
			String uri = request.uri();

			ConcurrentMap<String, String> paramMap = getUrlParams(uri);
			System.out.println("接收到的参数是：" + JSON.toJSONString(paramMap));

			online(paramMap.get("uid"), ctx.channel());
			// 如果url包含参数，需要处理
			if (uri.contains("?")) {
				String newUri = uri.substring(0, uri.indexOf("?"));
				request.setUri(newUri);
			}
		} else if (msg instanceof TextWebSocketFrame) {
			// 正常的TEXT消息类型
			TextWebSocketFrame frame = (TextWebSocketFrame) msg;
			System.out.println("read0: " + frame.text());
		}
		super.channelRead(ctx, msg);
	}

	private static ConcurrentMap<String, String> getUrlParams(String url) {
		ConcurrentMap<String, String> map = new ConcurrentHashMap<>();
		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;
		}
	}

	/**
	 * 上线一个用户
	 *
	 * @param channel
	 * @param userId
	 */
	private void online(String userId, Channel channel) {
		// 保存channel通道的附带信息，以用户的uid为标识
		channel.attr(key).set(userId);
		ChannelHandlerPool.channelGroup.add(channel);
	}

}

6.ClientsCheck

通道连接数

public class ClientsCheck implements Runnable {
	@Override
	public void run() {
		try {
			while (true) {
				int size = ChannelHandlerPool.channelGroup.size();
				System.out.println("client quantity -> " + size);
				Thread.sleep(5000);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

7.Controller

@RestController
public class IndexController {

	@PostMapping("/send")
	public String send2User(@RequestParam(value = "uid") String uid, @RequestParam(value = "data") String data) {
		List<Channel> channelList = getChannelByName(uid);
		if (channelList.size() <= 0) {
			return "用户" + uid + "不在线！";
		}
		channelList.forEach(channel -> channel.writeAndFlush(new TextWebSocketFrame(data)));
		return "success";
	}

	/**
	 * 根据用户id查找channel
	 * 
	 * @param name
	 * @return
	 */
	public List<Channel> getChannelByName(String name) {
		AttributeKey<String> key = AttributeKey.valueOf("user");
		return ChannelHandlerPool.channelGroup.stream().filter(channel -> channel.attr(key).get().equals(name))
				.collect(Collectors.toList());
	}

}

8.test.html

注：
1.我只会简单的前端js，自己引入jquery.js
2.以uid为55,创建了一个通道连接

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Netty-Websocket</title>
</head>
<script src="webjars/jquery/3.4.1/jquery.js"></script>
<script type="text/javascript">
    var socket;
    if (!window.WebSocket) {
        window.WebSocket = window.MozWebSocket;
    }
    if (window.WebSocket) {
    	// uid字段要和后台WebSocketHandler类中的paramMap.get("uid")字段对应
        socket = new WebSocket("ws://127.0.0.1:12345/ws?uid=" + 55); // 55代表用户id
        socket.onmessage = function (event) {
            $("#ws").html(event.data)
            console.log(event.data)
        };
        socket.onopen = function (event) {
            console.log("Netty-WebSocket服务器。。。。。。连接")
        };
        socket.onclose = function (event) {
            console.log("Netty-WebSocket服务器。。。。。。关闭")
        };
    } else {
        alert("您的浏览器不支持WebSocket协议！");
    }
</script>
<body>
    <div id="ws" class="ws">

    </div>
</body>
</html>

9.效果演示

访问：http://127.0.0.1:8700/test.html

后台控制台打印，说明用户uid：55 已经建立通道

访问接口：http://127.0.0.1:8700/send，实现后端向前端推送消息

查看test.html页面

完成后端向uid：55， 推送“test：Netty+WebSocket”的内容

本文地址：https://blog.csdn.net/qq_38370965/article/details/112572990