spring netty websocket 服务端开发

素颜马尾好姑娘i 2022-12-10 10:26 160阅读 0赞

## 场景 ##

**需求:** 由于公司是做在线教育的,客户的定制化需求,要求同一个账号同时只能观看一个视频。(BS)

**分析:** 刚开始想过监听浏览器的close()事件，打开视频向redis 中存一个status,关闭浏览器修改这个status。但是不能处理极端情况如: 强制杀死进程、断电等(不考虑缓存播放视频的情况)

**方案:** 想到socket,自然想到netty对socket 的支持非常好。

**为什么选择netty?**

1. API使用简单，开发门槛低  
2. 功能强大，预置了多种编解码器功能  
3. 几行代码，就能解决粘包\\拆包问题  
4. 成熟、稳定 Netty修复了所有JDK NIO BUG

代码示例: 基于spring 框架整合的Netty

<dependency>
    	<groupId>io.netty</groupId>
    	<artifactId>netty-all</artifactId>
    	<version>4.1.6.Final</version>
    </dependency>

**netty server 启动类**

@Service
    public class MyWebSocketServer { 
    
    	private static final Logger logger = LoggerFactory.getLogger(MyWebSocketServer.class);
    
    	@Autowired
    	private PropertiesUtil propertiesUtil;
    
    	// 在对象加载完依赖注入后执行
    	@PostConstruct
    	public void init() { 
    	
    		// 创建socket服务
    		new Thread(new Runnable() { 
    			@Override
    			public void run() { 
    				logger.info("正在启动websocket服务器....");
    				NioEventLoopGroup boss = new NioEventLoopGroup();
    				NioEventLoopGroup work = new NioEventLoopGroup();
    				try { 
    					ServerBootstrap bootstrap = new ServerBootstrap();
    					bootstrap.group(boss, work);
    					bootstrap.channel(NioServerSocketChannel.class);
    					bootstrap.childHandler(new ChannelInitializer() { 
    						@Override
    						protected void initChannel(Channel ch) { 
    
    							// 设置log监听器，并且日志级别为debug，方便观察运行流程
    							ch.pipeline().addLast("logging", new LoggingHandler("DEBUG"));
    							// 设置websocket解码器,将请求、响应消息解码为HTTP
    							ch.pipeline().addLast("http-codec", new HttpServerCodec());
    							// HTTP聚合器，使用websocket会用到
    							ch.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
    							// 支持浏览器和服务器进行websocket通信
    							ch.pipeline().addLast("http-chunked", new ChunkedWriteHandler());
    							// TODO 实现心跳检测需要在服务端或客户端加入下面的IdleStateHandler,指定读写的超时时间,
    							// 超时后,触发继承ChannelInboundHandlerAdapter自定义的handler中userEventTriggered方法.
    							ch.pipeline().addLast(new IdleStateHandler(20L, 0L, 0L, TimeUnit.SECONDS));
    							// 自定义的心跳服务处理
    							ch.pipeline().addLast(new HeartBeatServerHandler());
    							// 自定义的业务handler
    							ch.pipeline().addLast("handler", new MyWebSocketServiceHandler());
    						}
    					});
    					Channel channel = bootstrap.bind(9091).sync().channel();
    					logger.info("webSocket服务器启动成功：" + channel);
    					channel.closeFuture().sync();
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    					logger.info("运行出错：" + e);
    				} finally { 
    					boss.shutdownGracefully();
    					work.shutdownGracefully();
    					logger.info("websocket服务器已关闭");
    				}
    			}
    		}).start();
    	}
    
    }

1. 使用@PostConstruct这样随着spring项目启动,加载完这个类依赖的bean后,去执行init()方法创建netty服务端。  
2. 为什么要创建一个线程？因为sync().channel() 是阻塞的，试想一下如果不这样做，spring 初始化到这段代码后,将阻塞在此处，spring 初始化止于此,和跑一个main() 没区别.  
3. 重点在26~42行，initChannel(),其中最后两个 handler是自定义的。分别是处理心跳检测Handler,和业务

**心跳检测Handler**:  
 Netty中自带了一个IdleStateHandler 可以用来实现心跳检测。 这里会监听到读/写超时的客户端

public class HeartBeatServerHandler extends ChannelInboundHandlerAdapter { 
    
    	@Override
    	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { 
    
    		if (evt instanceof IdleStateEvent) { 
    			IdleStateEvent event = (IdleStateEvent) evt;
    			String eventType = null;
    			switch (event.state()) { 
    				case READER_IDLE:
    					eventType = "读空闲";
    					break;
    				case WRITER_IDLE:
    					eventType = "写空闲";
    					break;
    				case ALL_IDLE:
    					eventType = "读写空闲";
    					break;
    			}
    			System.out.println(ctx.channel().remoteAddress() + "超时事件：" + eventType);
    			ctx.channel().close();
    		} else { 
    			super.userEventTriggered(ctx, evt);
    		}
    	}
    	
    	@Override
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
    		ctx.close();
    	}
    
    }

**处理业务的Handler**:

public class MyWebSocketServiceHandler extends SimpleChannelInboundHandler<Object> { 
    
    	private static final Logger logger = LoggerFactory.getLogger(StudyRecordWebSocketHandler.class);
    	
    	private WebSocketServerHandshaker handshaker;
    
    	@Override
    	protected void channelRead0(ChannelHandlerContext ctx, Object msg) { 
    		if (msg instanceof FullHttpRequest){ 
    			// 第一次建立连接,以http请求形式接入
    			logger.info("创建连接");
    			handleHttpRequest(ctx, (FullHttpRequest) msg);
    		}else if (msg instanceof WebSocketFrame){ 
    			// 处理websocket客户端的消息
    			handlerWebSocketFrame(ctx, (WebSocketFrame) msg);
    		}
    	}
    
    
    	@Override
    	public void channelActive(ChannelHandlerContext ctx) { 
    		logger.info(" 客户端加入连接："+ctx.channel());
    	}
    
    	//断开连接
    	@Override
    	@SuppressWarnings("all")
    	public void channelInactive(ChannelHandlerContext ctx) { 
    		logger.info("客户端断开连接："+ctx.channel());
    	}
    
    	@Override
    	public void channelReadComplete(ChannelHandlerContext ctx) { 
    		ctx.flush();
    	}
    
    	private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame){ 
    
    		// 判断是否关闭链路的指令
    		if (frame instanceof CloseWebSocketFrame) { 
    			handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
    			return;
    		}
    		// 判断是否ping消息
    		if (frame instanceof PingWebSocketFrame) { 
    			ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
    			return;
    		}
    		// 仅支持文本消息，不支持二进制消息
    		if (!(frame instanceof TextWebSocketFrame)) { 
    			logger.info("本例程仅支持文本消息，不支持二进制消息");
    			throw new UnsupportedOperationException(String.format("%s frame types not supported", frame.getClass().getName()));
    		}
    
    		// 接收数据
    		String requestMsg = ((TextWebSocketFrame) frame).text();
    		// TODO 处理消息 .....
    	}
    
    
    	/** * 唯一的一次http请求，用于创建websocket */
    	private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) { 
    		// 要求Upgrade为websocket,HTTP解析失败,返回HTTP异常
    		if (!req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) { 
    			//若不是websocket方式，则创建BAD_REQUEST的req，返回给客户端
    			sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
    			return;
    		}
    
    		// 构造握手响应返回
    		WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory("ws://localhost:9091/websocket", null, false);
    		handshaker = wsFactory.newHandshaker(req);
    		if (handshaker == null) { 
    			WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
    		} else { 
    			handshaker.handshake(ctx.channel(), req);
    		}
    	}
    
    	/** * 拒绝不合法的请求，并返回错误信息 */
    	private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) { 
    		// 返回应答给客户端
    		if (res.status().code() != 200) { 
    			ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
    			res.content().writeBytes(buf);
    			buf.release();
    		}
    		ChannelFuture f = ctx.channel().writeAndFlush(res);
    		// 如果是非Keep-Alive，关闭连接
    		if (!isKeepAlive(req) || res.status().code() != 200) { 
    			f.addListener(ChannelFutureListener.CLOSE);
    		}
    	}
    
    
    	// 异常情况关闭客户端
    	@Override
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
    		cause.printStackTrace();
    		ctx.close();
    	}
    }