使用netty实现websocket
1. 前言
年前有个大屏的项目是基于websocket实现的，今年又有后续的开发工作，最近就想着先把之前的代码优化一下。那么得先研究一下之前的后端实现框架 - spring-boot-starter-websocket ，这算是spring boot 官方的websocket实现框架了，但出乎意料的受到网上一致差评。有很多博主都推荐用netty来实现，正好我前几天在看nio的实现原理，就萌生了重构后端代码,用netty来实现websocket的想法。
2. nio的应用
本段是做一些技术知识点的铺垫，为了避免本末倒置，一些知识只是简单说明，详细内容你们可以找专门的书看看，每个知识点衍生出来的内容都很多。
nio模型是同步非阻塞io，它针对于传统io的优化主要体现在两点：channel中零拷贝的实现提升了资源传递效率；selector的模式提升了线程的利用率。nio模型提出的很早，现如今很多框架都完成了bio到nio的升级，例如，我们springboot开发都会用到的tomcat。
2.1. 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
2.2. netty
而基于nio模型，目前公认最有影响力的java框架就是netty，像目前大火的 dubbo、zookeeper等，都是通过netty来实现nio。建议大家还是相信大厂，在玩nio模型方面，目前还没几个玩的比netty好。
Netty中的Reactor模型主要由多路复用器(Acceptor)、事件分发器(Dispatcher)、事件处理器(Handler)组成，可以分为三种：
- 单线程模型：所有I/O操作都由一个线程完成，即多路复用、事件分发和处理都是在一个Reactor线程上完成的。适用于小容量场景。
- 多线程模型：为了解决单线程模型存在的一些问题，演化而来的Reactor线程模型。在绝大多数场景下，Reactor多线程模型都可以满足性能需求。
- 主从多线程模型：采用多个reactor，每个reactor都在自己单独的线程里执行。如果是多核，则可以同时响应多个客户端的请求，一旦链路建立成功就将链路注册到负责I/O读写的SubReactor线程池上。适用于大容量场景，比如海量连接处理。
Netty的线程模型并非固定不变，在启动辅助类中创建不同的EventLoopGroup实例并通过适当的参数配置，就可以支持上述三种Reactor线程模型。正是因为Netty对Reactor线程模型的支持提供了灵活的定制能力，所以可以满足不同业务场景的性能需求。
3. 技术选型
websocket只是一个协议，实现websocket协议的方式有很多种，在本次技术选型中，我尝试了三种方式：
- spring-boot-starter-websocket：基于tomcat容器实现websocket。
- netty-websocket-spring-boot-starter：基于netty实现websocket，封装成starter。
- netty-all：基于netty实现websocket，原生实现。
在实现websocket方式上，主要还是区别在tomcat和netty上，2和3的底层实现原理一样。
3.1. tomcat 和netty
我们来看看分别用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长连接混用一套线程模型，很多时候不好针对单独一方做性能调优。
3.2. netty-websocket-spring-boot-starter和netty-all
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来重构。
4. 代码参考
按道理，我应该分享一份“干净”的demo代码，让大家可以直接运行的。但最近实在无心写代码了，所以直接将项目代码里面websocket的部分粘贴下来，核心内容都在。
4.1.pom.xml
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.36.Final</version>
</dependency>
4.2.ChannelGroupUtil.java
@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;
}
}
4.3. ChannelHandlerBase.java
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+"用户已注册"));
}
}
4.4. DataChannelHandler.java
@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;
}
}
}
4.5. MessageSender.java
@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);
}
}
4.6. DataNettyServer.java
@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();
}
}
正文到此结束
- 本文标签： json 同步 message 长连接 HashMap 定制 js 配置 代码 实例 dubbo 线程池 MQ bug tag 协议 https map 多线程 Select final Logging HTTP协议 模型 并发 服务器 struct zookeeper Netty 开发 IOS IO springboot stream web ip apr 管理 parse 参数 ACE NIO 高并发 Reactor apache tomcat cat Bootstrap 程序员 value http executor Spring Boot BIO java pom 线程 IDE CTO key DDL Service id equals spring db tar 数据 TCP XML src 处理器 短连接 ConcurrentHashMap 端口 UI 需求
- 版权声明： 本文为互联网转载文章，出处已在文章中说明(部分除外)。如果侵权，请联系本站长删除，谢谢。
- 本文海报： 生成海报一 生成海报二
热门推荐
相关文章
Loading...
![[HBLOG]公众号](http://www.liuhaihua.cn/img/qrcode_gzh.jpg)
