年前有个大屏的项目是基于websocket实现的,今年又有后续的开发工作,最近就想着先把之前的代码优化一下。那么得先研究一下之前的后端实现框架 - spring-boot-starter-websocket ,这算是spring boot 官方的websocket实现框架了,但出乎意料的受到网上一致差评。有很多博主都推荐用netty来实现,正好我前几天在看nio的实现原理,就萌生了重构后端代码,用netty来实现websocket的想法。
本段是做一些技术知识点的铺垫,为了避免本末倒置,一些知识只是简单说明,详细内容你们可以找专门的书看看,每个知识点衍生出来的内容都很多。
nio模型是同步非阻塞io,它针对于传统io的优化主要体现在两点:channel中零拷贝的实现提升了资源传递效率;selector的模式提升了线程的利用率。nio模型提出的很早,现如今很多框架都完成了bio到nio的升级,例如,我们springboot开发都会用到的tomcat。
tomcat 6 之前几乎都是bio,你可以在启动日志中看到每个执行线程名字都是以 “ http-bio-端口号-exec-
” 为前缀的。而后都是nio,对应线程名前缀是 “ http-nio-端口号-exec-
” 。后来又出现了apr,对于tomcat中 bio、nio、apr 三者之间的区别,有兴趣可以了解一下,这里就不细讲了。
因为spring boot是内嵌tomcat,而很多有意思的日志信息需要你手动来开启,可以加上如下的配置信息。
logging.level.org.apache.tomcat=debug logging.level.org.apache.catalina=debug
而基于nio模型,目前公认最有影响力的java框架就是netty,像目前大火的 dubbo、zookeeper等,都是通过netty来实现nio。建议大家还是相信大厂,在玩nio模型方面,目前还没几个玩的比netty好。
Netty中的Reactor模型主要由多路复用器(Acceptor)、事件分发器(Dispatcher)、事件处理器(Handler)组成,可以分为三种:
Netty的线程模型并非固定不变,在启动辅助类中创建不同的EventLoopGroup实例并通过适当的参数配置,就可以支持上述三种Reactor线程模型。正是因为Netty对Reactor线程模型的支持提供了灵活的定制能力,所以可以满足不同业务场景的性能需求。
websocket只是一个协议,实现websocket协议的方式有很多种,在本次技术选型中,我尝试了三种方式:
在实现websocket方式上,主要还是区别在tomcat和netty上,2和3的底层实现原理一样。
我们来看看分别用tomcat和netty实现websocket之后,程序的线程运行情况。
1、(spring-boot-starter-websocket)线程监控
2、(netty)线程监控
在图一中,对于tomcat,在nio多路复用上是典型的单线程模型。由一个一直轮询的poller线程来连接请求(另一个是处理阻塞的),然后将封装后的任务交给worker线程池,即图中 http-nio-9001-exec- 。而且这时,tomcat处理的请求,既包括传统的http请求,还包括websocket。
在图二中,因为是在springboot项目中使用netty,因此即能看到tomcat的nio实现,也能看到netty的Reactor模型(nioEventLoopGroup-x-x)。我的设计是,专门用netty的nio来处理websocket,其他的交给tomcat。针对高并发的websocket场景,我还可以选用netty的主从多线程模型来应对。
二者对比上,主要体现在netty对于nio的实现更好,不仅体现在图示中的线程模式上,还包括在底层的数据拷贝。而且如果程序中websocket的应用规模比较大的话,无论是tomcat还是netty,都建议独立出一个连接,单独映射一个端口。如果http短连接和websocket长连接混用一套线程模型,很多时候不好针对单独一方做性能调优。
netty-websocket-spring-boot-starter的内部实现其实就是netty-all,只是一位程序员好心的将netty-all封装了一下。封装后,netty-websocket-spring-boot-starter的代码实现,基本上和spring-boot-starter-websocket就没什么区别了,让你基本看不到netty的影子。
这样很好吗?开发上可能更简单了,但是这样的封装实在太强行了。强行往官方的spring-boot-starter-websocket上靠,失去了自己的灵活性,有些时候还容易带来很多误解。我在使用netty-websocket-spring-boot-starter之后,就是被它的名词概念绕来绕去。最后实在忍不住,删掉代码,直接基于netty-all来重构。
按道理,我应该分享一份“干净”的demo代码,让大家可以直接运行的。但最近实在无心写代码了,所以直接将项目代码里面websocket的部分粘贴下来,核心内容都在。
<dependency> <groupId>io.netty</groupId> <artifactId>netty-all</artifactId> <version>4.1.36.Final</version> </dependency>
@Slf4j @Service public class ChannelGroupUtil { public static final String KEY_USER_ID = "userId"; public static final AttributeKey<String> ATTRIBUTE_KEY_USER_ID = AttributeKey.valueOf(KEY_USER_ID); private static ChannelGroup gChannelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); private static ConcurrentHashMap<String, Channel> gUserChannelMap = new ConcurrentHashMap<>(); /** * 新增 Channel * * @param channel */ public void addChannel(Channel channel) { gChannelGroup.add(channel); } /** * 关闭 Channel * * @param channel */ public void removeChannel(Channel channel) { gChannelGroup.remove(channel); channel.close(); } /** * 关闭 userId的所有 channel * @param userId */ public void removeChannel(String userId) { if (userId != null) { gChannelGroup.stream() .forEach((Channel channel) -> { String channelUserId = channel.attr(ChannelGroupUtil.ATTRIBUTE_KEY_USER_ID).get(); if (userId.equals(channelUserId)) { gChannelGroup.remove(channel); channel.close(); } }); } } /** * 用户 注册 * 1、设置 Channel 中 userId属性 * 2、存放在 gUserChannelMap 中 * * @param channel * @param userId */ public void userRegister(Channel channel, String userId) { this.setChannelUserAttribute(channel, userId); gUserChannelMap.put(userId, channel); log.info(userId + " 已连接"); } /** * 用户 注销 * 1、已注册channel 关闭 * 2、未注册channel 关闭 * * @param channel */ public void userRemove(Channel channel) { String userId = this.getUserIdByChannel(channel); if(userId!=null){ gUserChannelMap.remove(userId); this.removeChannel(userId); }else { this.removeChannel(channel); } log.info(userId + " 已退出"); } /** * 获取用户 对应的 Channel * * @param userId * @return */ public Channel getChannelByUserId(String userId) { return gUserChannelMap.get(userId); } /** * 给 Channel 设置用户名 userId 的属性 * * @param channel * @param userId */ private void setChannelUserAttribute(Channel channel, String userId) { channel.attr(ATTRIBUTE_KEY_USER_ID).setIfAbsent(userId); } /** * 获取 channel 的userId * * @param channel * @return */ public String getUserIdByChannel(Channel channel) { return channel.attr(ATTRIBUTE_KEY_USER_ID).get(); } public ChannelGroup getChannelGroup() { return gChannelGroup; } public ConcurrentHashMap<String, Channel> getUserChannelMap() { return gUserChannelMap; } }
public class ChannelHandlerBase extends SimpleChannelInboundHandler<TextWebSocketFrame> { @Autowired private ChannelGroupUtil channelGroupUtil; @Autowired private MessageSender messageSender; /** * 连接开启 * @param channelHandlerContext */ @Override public void handlerAdded(ChannelHandlerContext channelHandlerContext) { channelGroupUtil.addChannel(channelHandlerContext.channel()); } /** * 接受消息 * 由子类实现 */ @Override protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame msg) { } /** * 连接关闭 * @param channelHandlerContext */ @Override public void handlerRemoved(ChannelHandlerContext channelHandlerContext) { channelGroupUtil.userRemove(channelHandlerContext.channel()); } /** * 异常处理 * @param channelHandlerContext * @param cause */ @Override public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable cause) { channelGroupUtil.userRemove(channelHandlerContext.channel()); } /** * 心跳检查处理 HEART_CHECK * @param channel */ protected void heartCheckHandler(Channel channel){ messageSender.sendMessage(channel, WebSocketDTO.data(ModuleEnum.HEART_CHECK, "回复心跳检查")); } /** * 注册用户 USER_REGISTER * @param channel * @param request * @param webSocketEnum */ protected void registerHandler(Channel channel, WebSocketDTO request, WebSocketEnum webSocketEnum){ UserRegisterDataDTO userRegisterDataDTO =JSON.parseObject(JSON.toJSONString(request.getData()),UserRegisterDataDTO.class); String userId= userRegisterDataDTO.getUsername()+"_"+webSocketEnum.name()+"_"+ userRegisterDataDTO.getDevice().name(); //关闭 userId 的其他channel channelGroupUtil.removeChannel(userId); //将该 channel绑定userId,存入UserChannelMap channelGroupUtil.userRegister(channel,userId); messageSender.sendMessage(channel, WebSocketDTO.data(ModuleEnum.USER_REGISTER, userId+"用户已注册")); } }
@Component @ChannelHandler.Sharable public class DataChannelHandler extends ChannelHandlerBase { @Override public void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame msg) { Channel channel = channelHandlerContext.channel(); WebSocketDTO request = JSON.parseObject(msg.text(), WebSocketDTO.class); switch (request.getModule()) { case USER_REGISTER: super.registerHandler(channel, request, WebSocketEnum.DATA); break; case HEART_CHECK: super.heartCheckHandler(channel); break; default: break; } } }
@Service public class MessageSender { @Autowired private ChannelGroupUtil channelGroupUtil; /** * 发送给 具体用户 * * @param userId * @param webSocketDTO */ public void sendMessage(String userId, WebSocketDTO webSocketDTO) { Channel channel = channelGroupUtil.getChannelByUserId(userId); this.sendMessage(channel,webSocketDTO); } public void sendMessage(String username, WebSocketEnum webSocketEnum, DeviceEnum deviceEnum, WebSocketDTO webSocketDTO) { String userId = username + "_" + webSocketEnum.name() + "_" + deviceEnum.name(); this.sendMessage(userId, webSocketDTO); } public void sendMessage(Channel channel, WebSocketDTO webSocketDTO) { if(channel!=null) { channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(webSocketDTO))); } } /** * 发送给 该用户下的所有设备 * * @param username * @param webSocketEnum * @param webSocketDTO */ public void sendMessageToAllDevices(String username, WebSocketEnum webSocketEnum, WebSocketDTO webSocketDTO) { for (DeviceEnum deviceEnum : DeviceEnum.values()) { String userId = username + "_" + webSocketEnum.name() + "_" + deviceEnum.name(); this.sendMessage(userId, webSocketDTO); } } /** * 转换 userId * * @param userId * @param webSocketEnum * @param deviceEnum * @return */ public String convertUserId(String userId, WebSocketEnum webSocketEnum, DeviceEnum deviceEnum) { if (userId != null) { if (webSocketEnum != null) { for (WebSocketEnum we : WebSocketEnum.values()) { if (!we.equals(webSocketEnum)) { userId= userId.replace("_" + we.name(), "_" + webSocketEnum.name()); } } } if (deviceEnum != null) { for (DeviceEnum de : DeviceEnum.values()) { if (!de.equals(deviceEnum)) { userId=userId.replace("_" + de.name(), "_" + deviceEnum.name()); } } } } return userId; } public String convertUserId(Channel channel, WebSocketEnum webSocketEnum, DeviceEnum deviceEnum) { String userId = channelGroupUtil.getUserIdByChannel(channel); return convertUserId(userId, webSocketEnum, deviceEnum); } }
@Slf4j @Component public class DataNettyServer { @Autowired private DataChannelHandler dataChannelHandler; private static final String WEB_SOCKET_PROTOCOL = "WebSocket"; private static final String THREAD_NAME_PREFIX="nettyStart-"; @Value("${ws.data.port}") private int port; @Value("${ws.data.webSocketPath}") private String webSocketPath; private EventLoopGroup bossGroup; private EventLoopGroup workGroup; /** * 启动 * * @throws InterruptedException */ private void start() throws InterruptedException { this.bossGroup= new NioEventLoopGroup(); this.workGroup= new NioEventLoopGroup(16); ServerBootstrap bootstrap = new ServerBootstrap(); // bossGroup辅助客户端的tcp连接请求, workGroup负责与客户端之前的读写操作 bootstrap.group(this.bossGroup, this.workGroup); // 设置NIO类型的channel bootstrap.channel(NioServerSocketChannel.class); // 设置监听端口 bootstrap.localAddress(new InetSocketAddress(this.port)); // 连接到达时会创建一个通道 bootstrap.childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) { // 流水线管理通道中的处理程序(Handler),用来处理业务 // webSocket协议本身是基于http协议的,所以这边也要使用http编解码器 ch.pipeline().addLast(new HttpServerCodec()); ch.pipeline().addLast(new ObjectEncoder()); // 以块的方式来写的处理器 ch.pipeline().addLast(new ChunkedWriteHandler()); //http数据在传输过程中是分段的,HttpObjectAggregator可以将多个段聚合 ch.pipeline().addLast(new HttpObjectAggregator(8192)); ch.pipeline().addLast(new WebSocketServerProtocolHandler(webSocketPath, WEB_SOCKET_PROTOCOL, true, 65536 * 10)); // 自定义的handler,处理业务逻辑 ch.pipeline().addLast(dataChannelHandler); } }); // 配置完成,开始绑定server,通过调用sync同步方法阻塞直到绑定成功 ChannelFuture channelFuture = bootstrap.bind().sync(); // 对关闭通道进行监听 channelFuture.channel().closeFuture().sync(); } @PreDestroy public void destroy() throws InterruptedException { if (this.bossGroup != null) { this.bossGroup.shutdownGracefully().sync(); } if (this.workGroup != null) { this.workGroup.shutdownGracefully().sync(); } } @PostConstruct public void init() { //需要开启一个新的线程来执行netty server 服务器 Thread initThread = new Thread(() -> { try { start(); } catch (InterruptedException e) { e.printStackTrace(); } }); initThread.setName(THREAD_NAME_PREFIX+ WebSocketEnum.DATA.name()); initThread.start(); } }