文章目录
- 一. VectorNettyApplication启动类配置
- 二.WebSocketServerBoot初始化服务端Netty
- 三. WebsocketServerChannelInitializer初始化服务端Netty读写处理器
- 四.initParamHandler处理器-去参websocket识别
- 五.MessageHandler核心业务处理类-采用工厂策略模式
- 5.1 策略上下文
- 六.统一响应
- 七.统一输出处理器
一. VectorNettyApplication启动类配置
初始化SpringBoot线程同时初始化Netty线程
/**
* @description: 通知启动类
* @Title: VectorNotification
* @Package com.vector.notification
* @Author YuanJie
* @Date 2023/3/2 12:57
*/
@EnableDiscoveryClient // 开启服务注册与发现
@SpringBootApplication(scanBasePackages = {"com.vector"},exclude = {DataSourceAutoConfiguration.class}) // 开启组件扫描和自动配置
public class VectorNettyApplication implements CommandLineRunner {
@Value("${netty.host}")
private String host;
@Value("${netty.port}")
private Integer port;
@Resource
private WebSocketServerBoot webSocketServerBoot;
public static void main(String[] args) {
SpringApplication.run(VectorNettyApplication.class, args);
}
// springboot启动后执行netty服务端启动
@Override
public void run(String... args) throws Exception {
ChannelFuture channelFuture = webSocketServerBoot.bind(host, port);
// 优雅关闭, jvm关闭时将netty服务端关闭
Runtime.getRuntime().addShutdownHook(new Thread(() -> webSocketServerBoot.destroy()));
// 阻塞 直到channel关闭
channelFuture.channel().closeFuture().syncUninterruptibly();
}
}
二.WebSocketServerBoot初始化服务端Netty
主要进行netty的基本配置
/**
* @author YuanJie
* @projectName vector-server
* @package com.vector.netty.accept
* @className com.vector.netty.accept.ServerBootstrap
* @copyright Copyright 2020 vector, Inc All rights reserved.
* @date 2023/6/9 18:34
*/
@Component
@Slf4j
public class WebSocketServerBoot {
private final EventLoopGroup parentGroup = new NioEventLoopGroup();
private final EventLoopGroup childGroup = new NioEventLoopGroup(2);
private Channel channel;
@Resource
private WebsocketServerChannelInitializer websocketServerChannelInitializer;
/**
* 初始化服务端
* sync()：等待Future直到其完成，如果这个Future失败，则抛出失败原因；
* syncUninterruptibly()：不会被中断的sync()；
*/
public ChannelFuture bind(String host, Integer port) {
ChannelFuture channelFuture = null;
try {
channelFuture = new ServerBootstrap()
.group(parentGroup, childGroup) // 指定线程模型 一个用于接收客户端连接，一个用于处理客户端读写操作
.channel(NioServerSocketChannel.class) // 指定服务端的IO模型
.option(ChannelOption.SO_BACKLOG, 1024) // 设置TCP缓冲区
.childOption(ChannelOption.SO_KEEPALIVE, true) // 保持连接 tcp底层心跳机制
.childHandler(websocketServerChannelInitializer) // 指定处理新连接数据的读写处理逻辑
.bind(host, port)
.addListener(new GenericFutureListener<Future<? super Void>>() {
@Override
public void operationComplete(Future<? super Void> future) throws Exception {
if (future.isSuccess()) {
log.info("服务端启动成功，监听端口：{}", port);
} else {
log.error("服务端启动失败，监听端口：{}", port);
bind(host, port + 1);
}
}
})
.syncUninterruptibly();// 绑定端口
channel = channelFuture.channel(); // 获取channel
} finally {
if (null == channelFuture) {
channel.close();
parentGroup.shutdownGracefully();
childGroup.shutdownGracefully();
}
}
return channelFuture;
}
/**
* 销毁
*/
public void destroy() {
if (null == channel) return;
channel.close();
parentGroup.shutdownGracefully();
childGroup.shutdownGracefully();
}
/**
* 获取通道
*
* @return
*/
public Channel getChannel() {
return channel;
}
}
三. WebsocketServerChannelInitializer初始化服务端Netty读写处理器
主要规划netty的读写处理器
/**
* @author YuanJie
* @projectName vector-server
* @package com.vector.netty.config
* @className com.vector.netty.server.ServerChannelInitializer
* @copyright Copyright 2020 vector, Inc All rights reserved.
* @date 2023/6/9 19:13
*/
@Component
public class WebsocketServerChannelInitializer extends ChannelInitializer<SocketChannel> {
// @Sharable
private final LoggingHandler loggingHandler = new LoggingHandler(LogLevel.INFO);
public final static String WEBSOCKET_PATH = "/ws";
@Resource
private InitParamHandler initParamHandler;
@Resource
private MessageHandler messageHandler;
@Resource
private OutBoundHandler outBoundHandler;
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
// 日志打印
pipeline.addLast(loggingHandler);
// http报文解析器 线程不安全不能被共享
pipeline.addLast(new HttpServerCodec());
// // 添加对大数据流的支持
pipeline.addLast(new ChunkedWriteHandler());
// // 消息聚合器 8192 8M
pipeline.addLast(new HttpObjectAggregator(1 << 13));
// 进行设置心跳检测
pipeline.addLast(new IdleStateHandler(60, 30, 60 * 30, TimeUnit.SECONDS));
// ================= 上述是用于支持http协议的 ==============
//websocket 服务器处理的协议，用于给指定的客户端进行连接访问的路由地址
// 处理uri参数 WebSocketServerProtocolHandler不允许带参数 顺序不可调换
pipeline.addLast(initParamHandler);
pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH,null, true, 1<<16,true,true,5000));
pipeline.addLast(messageHandler);
// 自定义出栈处理器
pipeline.addLast(outBoundHandler);
}
}
四.initParamHandler处理器-去参websocket识别
主要为了去参，WebSocketServerProtocolHandler不允许带参数，同时初始化一些信道用户数据
/**
* URL参数处理程序，这时候连接还是个http请求，没有升级成webSocket协议，此处SimpleChannelInboundHandler泛型使用FullHttpRequest
*
* @author YuanJie
* @date 2023/5/7 15:07
*/
@Slf4j
@ChannelHandler.Sharable
@Component
public class InitParamHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
/**
* 存储已经登录用户的channel对象
*/
public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
/**
* 存储用户id和用户的channelId绑定
*/
public static final Map<Long, ChannelId> userMap = new ConcurrentHashMap<>();
/**
* 用于存储群聊房间号和群聊成员的channel信息
*/
public static final Map<Long, ChannelGroup> groupMap = new ConcurrentHashMap<>();
@DubboReference
private MemberRemote memberRemote;
/**
* 此处进行url参数提取，重定向URL，访问webSocket的url不支持带参数的，带参数会抛异常，这里先提取参数，将参数放入通道中传递下去，重新设置一个不带参数的url
*
* @param ctx the {@link ChannelHandlerContext} which this {@link SimpleChannelInboundHandler}
* belongs to
* @param request the message to handle
* @throws Exception
*/
@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
if (!this.acceptInboundMessage(request)) {
ctx.fireChannelRead(request.retain());
}
String uri = request.uri();
log.info("NettyWebSocketParamHandler.channelRead0 --> : 格式化URL... {}", uri);
Map<CharSequence, CharSequence> queryMap = UrlBuilder.ofHttp(uri).getQuery().getQueryMap();
//将参数放入通道中传递下去
String senderId = "senderId";
if (StringUtils.isBlank(queryMap.get(senderId))) {
log.info("NettyWebSocketParamHandler.channelRead0 --> : 参数缺失 senderId");
ctx.close();
}
// 验证token
// verifyToken(ctx,senderId);
// 初始化数据
// initData(ctx, Long.valueOf(queryMap.get(senderId).toString()));
// 获取？之前的路径
request.setUri(WebsocketServerChannelInitializer.WEBSOCKET_PATH);
ctx.fireChannelRead(request.retain());
}
@Override
public void channelActive(ChannelHandlerContext ctx) {
//添加到channelGroup通道组
channelGroup.add(ctx.channel());
ctx.channel().id();
}
@Override
public void channelInactive(ChannelHandlerContext ctx) {
// 移除channelGroup 通道组
channelGroup.remove(ctx.channel());
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
log.error("NettyWebSocketParamHandler.exceptionCaught --> cause: ", cause);
ctx.close();
}
private void verifyToken(ChannelHandlerContext ctx, Long senderId) {
String userKey = CacheConstants.LOGIN_TOKEN_KEY + senderId;
RedissonCache redissonCache = SpringContextUtil.getBean(RedissonCache.class);
Boolean hasKey = redissonCache.hasKey(userKey);
if (!hasKey) {
log.info("NettyWebSocketParamHandler.channelRead0 --> : 用户未登录... {}", senderId);
ctx.close();
}
// token续期
redissonCache.expire(userKey, SystemConstants.TOKEN_EXPIRE_TIME, TimeUnit.MILLISECONDS);
}
/**
* 加入聊天室
*
* @param ctx
* @param senderId
* @throws ExecutionException
* @throws InterruptedException
*/
private void joinGroup(ChannelHandlerContext ctx, Long senderId) {
R r = null;
try {
CompletableFuture<R> result = memberRemote.getGroupListById(senderId);
r = result.get(3, TimeUnit.SECONDS);
} catch (Exception e) {
log.error("messageHandler.joinGroup查询群聊列表失败 ===> {}", e.getMessage());
ctx.channel().write(WSMessageDTO.error("查询群聊列表失败"));
return;
}
if (r == null || r.getCode() != 200) {
log.error("查询群聊列表失败 ====> {}", r.getMsg());
ctx.channel().write(WSMessageDTO.error("查询群聊列表失败"));
return;
}
//查询成功
//获取群聊列表
String json = JacksonInstance.toJson(r.getData());
List<Long> groupIds = JacksonInstance.toObjectList(json, new TypeReference<List<Long>>() {
});
ChannelGroup group;
for (Long groupId : groupIds) {
group = groupMap.get(groupId);
if (group == null) {
//如果群聊信道不存在,则创建一个群聊信道
group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
groupMap.put(groupId, group);
}
//将当前用户加入到群聊信道中
group.add(ctx.channel());
}
}
/**
* 加入聊天信道
*/
private void joinChat(ChannelHandlerContext ctx, Long senderId) {
//将当前用户的channelId放入map中
userMap.put(senderId, ctx.channel().id());
}
private void initData(ChannelHandlerContext ctx, Long senderId) {
joinChat(ctx, senderId);
joinGroup(ctx, senderId);
}
}
五.MessageHandler核心业务处理类-采用工厂策略模式
使得业务和通信协议无关，无感知。具体业务可以增加策略
/**
* @author YuanJie
* @projectName vector-server
* @package com.vector.netty.handler
* @className com.vector.netty.handler.MessageTypeHandler
* @copyright Copyright 2020 vector, Inc All rights reserved.
* @date 2023/6/15 16:23
*/
@Slf4j
@Component
@ChannelHandler.Sharable
public class MessageHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
@Resource
private MessageStrategyContext messageStrategyContext;
@Override
public void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) {
// 获取客户端发送的数据
WSMessageDTO wsMessageDTO = JacksonInstance.toObject(msg.text(), new TypeReference<WSMessageDTO>() {
});
wsMessageDTO.setMessageId(SnowFlakeUtil.getNextId());
log.info("客户端收到服务器数据：{}", wsMessageDTO.getMessage());
verifyParams(ctx, wsMessageDTO);
// 根据消息类型获取对应的处理器 核心处理方法
messageStrategyContext.messageType(ctx, wsMessageDTO);
}
private void verifyParams(ChannelHandlerContext ctx, WSMessageDTO wsMessageDTO) {
StringBuilder sb = new StringBuilder();
if (wsMessageDTO.getSenderId() == null) {
sb.append("senderId不能为空");
}
if (!EnumBusiness.containsBusiness(wsMessageDTO.getBusinessType())) {
sb.append("businessType不能为空");
}
if (!EnumMessage.containsMessage(wsMessageDTO.getMessageType())) {
sb.append("messageType不能为空");
}
if (wsMessageDTO.getMessage() == null) {
sb.append("message不能为空");
}
if (sb.length() > 0) {
log.error("参数校验失败:{}", sb.toString());
ctx.channel().write(WSMessageDTO.error("参数校验失败:" + sb.toString()));
ctx.close();
}
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
}
5.1 策略上下文
具体工厂策略可以详看我的策略模式文章
枚举维护前端参数和bean对象名
/**
* @author YuanJie
* @projectName vector-server
* @package com.vector.netty.enums
* @className com.vector.netty.enums.BusinessEnums
* @copyright Copyright 2020 vector, Inc All rights reserved.
* @date 2023/6/14 16:13
*/
public enum EnumBusiness {
/**
* 单聊
*/
chatMessage("chat", ChatMessageStrategy.class.getSimpleName()),
/**
* 群聊
*/
groupMessage("group", GroupMessageStrategy.class.getSimpleName()),
/**
* 在线人数
*/
onlineCount("onlineCount", OnlineCountStrategy.class.getSimpleName()),
TEST("test",TestStrategy.class.getSimpleName());
private final String businessType;
private final String beanName;
EnumBusiness(String businessType, String beanName) {
this.businessType = businessType;
this.beanName = StringUtils.isNotEmpty(beanName)?beanName.toLowerCase():null;
}
/**
* 根据code获取对应的枚举对象
*/
public static EnumBusiness getEnum(String businessType) {
EnumBusiness[] values = EnumBusiness.values(); // 获取枚举列表
if (null != businessType && values.length > 0) {
for (EnumBusiness value : values) {
if (value.businessType.equals(businessType)) {
return value; // 返回枚举对象
}
}
}
return null;
}
/**
* 该code在枚举列表code属性是否存在
*/
public static boolean containsBusiness(String businessType) {
EnumBusiness anEnum = getEnum(businessType); // 获取枚举对象
return anEnum != null;
}
/**
* 判断code与枚举中的code是否相同
*/
public static boolean equals(String businessType, EnumBusiness calendarSourceEnum) {
return calendarSourceEnum.businessType.equals(businessType);
}
public String getBusinessType() {
return businessType;
}
public String getBeanName() {
return beanName;
}
}
策略根据bean名获取实例对象
/**
* @author YuanJie
* @projectName vector-server
* @package com.vector.netty.service
* @className com.vector.netty.service.MessageContext
* @copyright Copyright 2020 vector, Inc All rights reserved.
* @date 2023/6/14 17:02
*/
@Component
@Slf4j
public class MessageStrategyContext {
/** 策略实例集合 */
private final ConcurrentHashMap<String, MessageStrategy> strategyConcurrentHashMap =
new ConcurrentHashMap<>(20);
/**
* 注入策略实例
* 如果使用的是构造器注入，可能会有多个参数注入进来。
*
* 如果使用的是field反射注入
*
* 如果使用的是setter方法注入，那么你将不能将属性设置为final。
*
* @param strategyMap
* 注意注入类型要是Map基础类型
* 注入接口,spring会自动注入他的所有被spring托管的实现类
*/
@Autowired
public MessageStrategyContext(Map<String, MessageStrategy> strategyMap) {
//清空集合数据
this.strategyConcurrentHashMap.clear();
if (!CollectionUtils.isEmpty(strategyMap)) {
strategyMap.forEach((beanName, messageStrategy) -> {
if (StringUtils.isEmpty(beanName) || messageStrategy == null) {
return;
}
this.strategyConcurrentHashMap.put(beanName.toLowerCase(), messageStrategy);
});
}
}
/**
* 选择业务方式
* 单聊,群聊,统计在线人数...
*
* @param msg 信息
*/
public void messageType(ChannelHandlerContext ctx, WSMessageDTO msg){
EnumBusiness enumerateInstances = EnumBusiness.getEnum(msg.getBusinessType());
if (CollectionUtils.isEmpty(strategyConcurrentHashMap)) {
log.info("策略实例集合初始化失败，请检查是否正确注入！");
}
MessageStrategy messageStrategy = strategyConcurrentHashMap.get(enumerateInstances.getBeanName());
messageStrategy.messageType(ctx, msg);
}
}
六.统一响应
注意使用该统一响应对象，所有入栈处理器必须使用即调用方必须是SimpleChannelInboundHandler，详细原因在下文 七.统一输出处理器中
/**
* @author YuanJie
* @projectName vector-server
* @package com.vector.netty.entity
* @className com.vector.netty.entity.SocketMessage
* @copyright Copyright 2020 vector, Inc All rights reserved.
* @date 2023/6/14 19:35
*/
@Data
public class WSMessageDTO {
/**
* 消息发送者
*/
private Long senderId;
/**
* 消息接收者/群聊id
*/
private Long chatId;
/**
* 消息类型 0文本 1图片 2文件 3视频 4语音 5位置 6名片 7链接 8系统消息
* @see com.vector.netty.enums.EnumMessage
*/
private byte messageType;
/**
* 业务类型 chat单聊 group群聊 onlineCount在线人数
* @see com.vector.netty.enums.EnumBusiness
*/
private String businessType;
/**
* 记录每条消息的id
*/
private Long messageId;
/**
* 消息内容
*/
private String message;
/**
* 消息发送时间
*/
private LocalDateTime sendTime;
/**
* 消息接收时间
*/
private LocalDateTime receiveTime;
/**
* 最后一条消息内容
*/
private String lastMessage;
/**
* 消息状态 0失败 1成功
*/
private byte code;
/**
* 封装统一返回格式
* @return
*/
public static TextWebSocketFrame ok(){
WSMessageDTO data = new WSMessageDTO();
data.setCode((byte) 1);
return new TextWebSocketFrame(JacksonInstance.toJson(data)).retain();
}
public static TextWebSocketFrame ok(WSMessageDTO data){
data.setCode((byte) 1);
return new TextWebSocketFrame(JacksonInstance.toJson(data)).retain();
}
public static TextWebSocketFrame error(String message){
WSMessageDTO data = new WSMessageDTO();
data.setCode((byte) 0);
data.setMessage(message);
return new TextWebSocketFrame(JacksonInstance.toJson(data)).retain();
}
}
七.统一输出处理器
- 若调用WSMessageDTO方法，必须注意内存泄露
- 即调用方必须是SimpleChannelInboundHandler<>
- 严禁使用ChannelInboundHandlerAdapter, 否则将造成严重内存泄露
- 相应地，必须使用此处的写出@param msg ，释放@param msg 引用
/**
* @author YuanJie
* @projectName vector-server
* @package com.vector.netty.handler
* @className com.vector.netty.handler.OutBoundHandler
* @copyright Copyright 2020 vector, Inc All rights reserved.
* @date 2023/7/24 22:38
*/
@Slf4j
@Component
@ChannelHandler.Sharable
public class OutBoundHandler extends ChannelOutboundHandlerAdapter {
/**
* 若调用WSMessageDTO方法，必须注意内存泄露
* 即调用方必须是SimpleChannelInboundHandler<>
* 严禁使用ChannelInboundHandlerAdapter, 否则将造成严重内存泄露
* 相应地，必须使用此处的写出@param msg ，释放@param msg 引用
* @param ctx the {@link ChannelHandlerContext} for which the write operation is made
* @param msg the message to write
* @param promise the {@link ChannelPromise} to notify once the operation completes
* @throws Exception
*/
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
if (msg instanceof FullHttpMessage){
log.info("webSocket协议升级成功");
// 出栈必须得这样写，不能自定义通信消息，可能把websocket反馈的消息覆盖了。 也不能在最后处理器调ctx.fireChannelRead()
ctx.writeAndFlush(msg,promise);
return;
} else if (msg instanceof TextWebSocketFrame) {
log.info("我要给客户端发送消息了。。。。");
ctx.writeAndFlush(msg, promise);
return;
}
log.error("OutBoundHandler.write: 消息类型错误");
ctx.writeAndFlush(WSMessageDTO.error("服务器内部错误: OutBoundHandler.write()"),promise);
ctx.close();
}
}
![[RK3128_LINUX5.1] 关于 RetroArch 使用](https://img-blog.csdnimg.cn/direct/76d61d8652b34e7483bb9e64b15fb323.png)
