Netty异步NIO框架（二）前端后端聊天 私聊及群聊
基于上篇文章扩展
1. 引入Netty依赖
<!--后端采用springboot项目，netty只需引入这一个依赖 -->
<!--netty依赖 -->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
</dependency>
2. 创建netty服务器
package com.cnpc.modules.netty;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;
/**
* @author wuzhenyong
* ClassName:NettyWebSocketServer.java
* date:2022-05-05 8:48
* Description: Netty服务器
*/
@Component
public class NettyWebSocketServer implements ApplicationListener<ContextRefreshedEvent> {
@Autowired
private WebSocketChannelInit webSocketChannelInit;
/**
* 容器初始化完成后调用
*
* @param contextRefreshedEvent
*/
@Override
public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
// 启动netty服务器
this.init();
}
private EventLoopGroup bossGroup = new NioEventLoopGroup(1);
private EventLoopGroup workerGroup = new NioEventLoopGroup();
public void init() {
try {
//1.创建服务端启动助手
ServerBootstrap serverBootstrap = new ServerBootstrap();
//2.设置线程组
serverBootstrap.group(bossGroup, workerGroup);
//3.设置参数
serverBootstrap.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.DEBUG))
.childHandler(webSocketChannelInit);
//4.启动 绑定端口不能和服务端口一致
ChannelFuture channelFuture = serverBootstrap.bind(9090).sync();
System.out.println("--Netty服务端启动成功---");
channelFuture.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
3. 创建通道初始化对象
package com.cnpc.modules.netty;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
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 org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* @author wuzhenyong
* ClassName:WebSocketChannelInit.java
* date:2022-05-05 8:53
* Description: 通道初始化对象
*/
@Component
public class WebSocketChannelInit extends ChannelInitializer {
@Autowired
private WebSocketHandler webSocketHandler;
@Override
protected void initChannel(Channel channel) throws Exception {
ChannelPipeline pipeline = channel.pipeline();
//对http协议的支持.
pipeline.addLast(new HttpServerCodec());
// 对大数据流的支持
pipeline.addLast(new ChunkedWriteHandler());
//post请求分三部分. request line / request header / message body
// HttpObjectAggregator将多个信息转化成单一的request或者response对象
pipeline.addLast(new HttpObjectAggregator(8000));
// 将http协议升级为ws协议. websocket的支持
pipeline.addLast(new WebSocketServerProtocolHandler("/ws", null, true, 65536, false, true));
// 自定义处理handler
pipeline.addLast(webSocketHandler);
}
}
4. 创建自定义处理类
package com.cnpc.modules.netty;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.springframework.stereotype.Component;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* @author wuzhenyong
* ClassName:WebSocketHandler.java
* date:2022-05-05 8:54
* Description: 自定义处理类
*/
@Component
@ChannelHandler.Sharable
public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
/**
* 管理channel的组，可以理解为channel的池 —— 客服使用
*/
public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
/**
* 用户通道管理
*/
public static ConcurrentHashMap<String, Channel> channelMap = new ConcurrentHashMap<>(16);
/**
* 通道绑定管理
*/
public static ConcurrentHashMap<String, String> bindMap = new ConcurrentHashMap<>(16);
public static List<Channel> channelList = new ArrayList<>();
/**
* 通道就绪事件
*
* @param ctx
* @throws Exception
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
// channel.read();
//当有新的客户端连接的时候, 将通道放入集合
SocketAddress socketAddress = ctx.channel().remoteAddress();
// 放入通道组
channels.add(channel);
System.out.println("有新的连接." + socketAddress);
}
/**
* 用户事件触发 token校验
*
* @param ctx ctx
* @param evt evt
* @throws Exception 异常
*/
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
System.out.println("触发事件");
if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete) {
WebSocketServerProtocolHandler.HandshakeComplete complete = (WebSocketServerProtocolHandler.HandshakeComplete) evt;
HttpHeaders httpHeaders = complete.requestHeaders();
// 自行处理鉴权问题
System.out.println("uri: " + uri);
System.out.println("握手成功");
channelMap.put(paramValue, ctx.channel());
}
super.userEventTriggered(ctx, evt);
}
/**
* 收到消息事件
*
* @param ctx 通道处理程序上下文
* @param textWebSocketFrame 文本框架网络套接字
* @throws Exception 异常
*/
@Override
protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame textWebSocketFrame) throws Exception {
// 按照自己公司的逻辑进行处理消息
System.out.println("通道： " + ctx.channel().remoteAddress() + "发送了消息");
String msg = textWebSocketFrame.text();
System.out.println("msg:" + msg);
String[] params = msg.split(":");
if (params[1].contains("yangmingquan")) {
// 私发
channelMap.get("yangmingquan").writeAndFlush(new TextWebSocketFrame(msg));
}
if (params[1].contains("wuzhenyong")) {
// 私发
channelMap.get("wuzhenyong").writeAndFlush(new TextWebSocketFrame(msg));
}
if (params[1].contains("all")) {
// 群发
channels.writeAndFlush(new TextWebSocketFrame(msg));
}
}
/**
* 通道未就绪--channel下线
*
* @param ctx ctx
* @throws Exception 异常
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
SocketAddress socketAddress = ctx.channel().remoteAddress();
System.out.println("通道：" + socketAddress + "已下线");
//当有客户端断开连接的时候,就移除对应的通道
channelList.remove(channel);
ctx.close();
}
/**
* 异常处理事件
*
* @param ctx ctx
* @param cause 导致
* @throws Exception 异常
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
Channel channel = ctx.channel();
//移除集合
channelList.remove(channel);
ctx.close();
}
}
5. 创建常量类
package com.cnpc.modules.netty;
/**
* @author wuzhenyong
* date:2022-05-05 8:35
* Description:
*/
public final class WebSocketConstant {
/**
* websocket head参数key
*/
public static final String SEC_WEBSOCKET_PROTOCOL = "Sec-WebSocket-Protocol";
}
6. 前端js
//1.创建websocket客户端
var wsServer = 'ws://ip/';
var limitConnect = 3; // 断线重连次数
var timeConnect = 0;
function websocket() {
//这里需要注意的是，prompt有两个参数，前面是提示的话，后面是当对话框出来后，在对话框里的默认值
var username = prompt("请输入您的名字", ""); //将输入的内容赋给变量 name ，
//建立WebSocket通讯
//注意：如果你要兼容ie8+，建议你采用 socket.io 的版本。下面是以原生WS为例
wsServer = 'ws://127.0.0.1:9090/ws?username=' + username;
var socket = new WebSocket(wsServer);
//连接成功时触发
socket.onopen = function() {
};
//收到的消息事件 按自己需求处理
socket.onmessage = function(res) {
//res为接受到的值，如 {"emit": "messageName", "data": {}}
//emit即为发出的事件名，用于区分不同的消息
console.log('接收到消息：', res)
} ;
socket.onclose = function() {
reconnect();
};
// 另外还有onclose、onerror，分别是在链接关闭和出错时触发
}
// 重连
function reconnect() {
// lockReconnect加锁，防止onclose、onerror两次重连
if (limitConnect > 0) {
limitConnect--;
timeConnect++;
console.log("第" + timeConnect + "次重连");
// 进行重连
setTimeout(function() {
websocket();
},2000);
} else {
console.log("TCP连接已超时");
}
}
7. 以上就可以使用websocket的方式进行聊天了
8. 遇到的问题
-
websocket路径传参问题 通道初始化对象 WebSocketChannelInit类
// 将http协议升级为ws协议. websocket的支持 pipeline.addLast(new WebSocketServerProtocolHandler("/ws", null, true, 65536, false, true)); // 最后一个参数设置为true则连接websocket可以进行路径传参，否则传参连接不成功 -
websocket路径参数获取问题 自定义处理器 WebSocketHandler类
// 重写userEventTriggered方法 用户连接触发事件 /** * 用户事件触发 token校验 * * @param ctx ctx * @param evt evt * @throws Exception 异常 */ @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { System.out.println("触发事件"); if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete) { WebSocketServerProtocolHandler.HandshakeComplete complete = (WebSocketServerProtocolHandler.HandshakeComplete) evt; // 获取header信息 HttpHeaders httpHeaders = complete.requestHeaders(); // 自行处理鉴权问题 System.out.println("uri: " + uri); System.out.println("握手成功"); channelMap.put(paramValue, ctx.channel()); } super.userEventTriggered(ctx, evt); } -
启动失败问题 Netty服务器 NettyWebSocketServer类
//init方法更换为一下代码，删除异常捕获关闭事件 public void init() throws InterruptedException { EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); //1.创建服务端启动助手 ServerBootstrap serverBootstrap = new ServerBootstrap(); //2.设置线程组 serverBootstrap.group(bossGroup, workerGroup); //3.设置参数 serverBootstrap.channel(NioServerSocketChannel.class) .handler(new LoggingHandler(LogLevel.DEBUG)) .childHandler(webSocketChannelInit); //4.启动 绑定端口不能和服务端口一致 ChannelFuture channelFuture = serverBootstrap.bind(9090).sync(); System.out.println("--Netty服务端启动成功---"); } -
js websocket传入header参数
var ws = new WebSocket("地址", ['header参数信息']);
