feng一样的双子
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”的内容
