创可贴网站怎么做图片,公司想制作网站吗,企业管理培训课程,网站底部友情链接怎么做的文章目录 一. VectorNettyApplication启动类配置二.WebSocketServerBoot初始化服务端Netty三. WebsocketServerChannelInitializer初始化服务端Netty读写处理器四.initParamHandler处理器-去参websocket识别五.MessageHandler核心业务处理类-采用工厂策略模式5.1 策略上下文 六… 文章目录 一. 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;Resourceprivate WebSocketServerBoot webSocketServerBoot;public static void main(String[] args) {SpringApplication.run(VectorNettyApplication.class, args);}// springboot启动后执行netty服务端启动Overridepublic 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;Resourceprivate 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 GenericFutureListenerFuture? super Void() {Overridepublic 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 ChannelInitializerSocketChannel {// Sharableprivate final LoggingHandler loggingHandler new LoggingHandler(LogLevel.INFO);public final static String WEBSOCKET_PATH /ws;Resourceprivate InitParamHandler initParamHandler;Resourceprivate MessageHandler messageHandler;Resourceprivate OutBoundHandler outBoundHandler;Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {ChannelPipeline pipeline socketChannel.pipeline();// 日志打印pipeline.addLast(loggingHandler);// http报文解析器 线程不安全不能被共享pipeline.addLast(new HttpServerCodec());
// // 添加对大数据流的支持pipeline.addLast(new ChunkedWriteHandler());
// // 消息聚合器 8192 8Mpipeline.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, 116,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 SimpleChannelInboundHandlerFullHttpRequest {/*** 存储已经登录用户的channel对象*/public static ChannelGroup channelGroup new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);/*** 存储用户id和用户的channelId绑定*/public static final MapLong, ChannelId userMap new ConcurrentHashMap();/*** 用于存储群聊房间号和群聊成员的channel信息*/public static final MapLong, ChannelGroup groupMap new ConcurrentHashMap();DubboReferenceprivate 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*/Overrideprotected 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);MapCharSequence, 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());}Overridepublic void channelActive(ChannelHandlerContext ctx) {//添加到channelGroup通道组channelGroup.add(ctx.channel());ctx.channel().id();}Overridepublic void channelInactive(ChannelHandlerContext ctx) {// 移除channelGroup 通道组channelGroup.remove(ctx.channel());}Overridepublic 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 {CompletableFutureR 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());ListLong groupIds JacksonInstance.toObjectList(json, new TypeReferenceListLong() {});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 SimpleChannelInboundHandlerTextWebSocketFrame {Resourceprivate MessageStrategyContext messageStrategyContext;Overridepublic void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) {// 获取客户端发送的数据WSMessageDTO wsMessageDTO JacksonInstance.toObject(msg.text(), new TypeReferenceWSMessageDTO() {});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();}}Overridepublic 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 ConcurrentHashMapString, MessageStrategy strategyConcurrentHashMap new ConcurrentHashMap(20);/*** 注入策略实例* 如果使用的是构造器注入可能会有多个参数注入进来。** 如果使用的是field反射注入** 如果使用的是setter方法注入那么你将不能将属性设置为final。** param strategyMap* 注意注入类型要是Map基础类型* 注入接口,spring会自动注入他的所有被spring托管的实现类*/Autowiredpublic MessageStrategyContext(MapString, 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*/Overridepublic 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();}
}
