websocket（三）进阶！netty框架实现websocket达到高并发

引言：

在前面两篇文章中，我们对原生websocket进行了了解，且用demo来简单的讲解了其用法。但是在实际项目中，那样的用法是不可取的，理由是tomcat对高并发的支持不怎么好，特别是tomcat9之前，可以测试发现websocket连接达到的数量很低，且容易断开。
所以有现在的第三篇，对websocket的一种进阶方法。

什么是Netty

Netty是业界最流行的NIO框架之一，它的健壮性、功能、性能、可定制性和可扩展性在同类框架中都是首屈一指的，它已经得到成百上千的商用项目验证，例如Hadoop的RPC框架Avro就使用了Netty作为底层通信框架，其他还有业界主流的RPC框架，也使用Netty来构建高性能的异步通信能力。
通过对Netty的分析，我们将它的优点总结如下：

API使用简单，开发门槛低；
功能强大，预置了多种编解码功能，支持多种主流协议；
定制能力强，可以通过ChannelHandler对通信框架进行灵活地扩展；
性能高，通过与其他业界主流的NIO框架对比，Netty的综合性能最优；
成熟、稳定，Netty修复了已经发现的所有JDK NIO BUG，业务开发人员不需要再为NIO的BUG而烦恼；
社区活跃，版本迭代周期短，发现的BUG可以被及时修复，同时，更多的新功能会加入；
经历了大规模的商业应用考验，质量得到验证。Netty在互联网、大数据、网络游戏、企业应用、电信软件等众多行业已经得到了成功商用，证明它已经完全能够满足不同行业的商业应用了。

基于Netty的websocket压力测试

点此进入

Demo详解

1.导入netty包

<!-- netty -->

<dependency>

<groupId>io.netty</groupId>

<artifactId>netty-all</artifactId>

<version>5.0.0.Alpha1</version>

</dependency>

2.server启动类
以下@Service，@PostConstruct注解是标注spring启动时启动的注解，新开一个线程去开启netty服务器端口。

package com.nettywebsocket;

import javax.annotation.PostConstruct;

import org.springframework.stereotype.Service;

import io.netty.bootstrap.ServerBootstrap;

import io.netty.channel.Channel;

import io.netty.channel.EventLoopGroup;

import io.netty.channel.nio.NioEventLoopGroup;

import io.netty.channel.socket.nio.NioServerSocketChannel;

/**

* ClassName:NettyServer 注解式随spring启动

* Function: TODO ADD FUNCTION.

* @author hxy

*/

@Service

public class NettyServer {

public static void main(String[] args) {

new NettyServer().run();

}

@PostConstruct

public void initNetty(){

new Thread(){

public void run() {

new NettyServer().run();

}

}.start();

}

public void run(){

System.out.println("===========================Netty端口启动========");

// Boss线程：由这个线程池提供的线程是boss种类的，用于创建、连接、绑定socket， （有点像门卫）然后把这些socket传给worker线程池。

// 在服务器端每个监听的socket都有一个boss线程来处理。在客户端，只有一个boss线程来处理所有的socket。

EventLoopGroup bossGroup = new NioEventLoopGroup();

// Worker线程：Worker线程执行所有的异步I/O，即处理操作

EventLoopGroup workGroup = new NioEventLoopGroup();

try {

// ServerBootstrap 启动NIO服务的辅助启动类,负责初始话netty服务器，并且开始监听端口的socket请求

ServerBootstrap b = new ServerBootstrap();

b.group(bossGroup, workGroup);

// 设置非阻塞,用它来建立新accept的连接,用于构造serversocketchannel的工厂类

b.channel(NioServerSocketChannel.class);

// ChildChannelHandler 对出入的数据进行的业务操作,其继承ChannelInitializer

b.childHandler(new ChildChannelHandler());

System.out.println("服务端开启等待客户端连接 ... ...");

Channel ch = b.bind(7397).sync().channel();

ch.closeFuture().sync();

} catch (Exception e) {

e.printStackTrace();

}finally{

bossGroup.shutdownGracefully();

workGroup.shutdownGracefully();

}

}

}

3.channle注册类

package com.nettywebsocket;

import io.netty.channel.ChannelInitializer;

import io.netty.channel.socket.SocketChannel;

import io.netty.handler.codec.http.HttpObjectAggregator;

import io.netty.handler.codec.http.HttpServerCodec;

import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;

import io.netty.handler.stream.ChunkedWriteHandler;

/**

* ClassName:ChildChannelHandler

* Function: TODO ADD FUNCTION.

* @author hxy

*/

public class ChildChannelHandler extends ChannelInitializer<SocketChannel>{

@Override

protected void initChannel(SocketChannel e) throws Exception {

// 设置30秒没有读到数据，则触发一个READER_IDLE事件。

// pipeline.addLast(new IdleStateHandler(30, 0, 0));

// HttpServerCodec：将请求和应答消息解码为HTTP消息

e.pipeline().addLast("http-codec",new HttpServerCodec());

// HttpObjectAggregator：将HTTP消息的多个部分合成一条完整的HTTP消息

e.pipeline().addLast("aggregator",new HttpObjectAggregator(65536));

// ChunkedWriteHandler：向客户端发送HTML5文件

e.pipeline().addLast("http-chunked",new ChunkedWriteHandler());

// 在管道中添加我们自己的接收数据实现方法

e.pipeline().addLast("handler",new MyWebSocketServerHandler());

}

}

4.存储类

以下类是用来存储访问的channle，channelGroup的原型是set集合，保证channle的唯一，如需根据参数标注存储，可以使用currentHashMap来存储。

package com.nettywebsocket;

import io.netty.channel.group.ChannelGroup;

import io.netty.channel.group.DefaultChannelGroup;

import io.netty.util.concurrent.GlobalEventExecutor;

/**

* ClassName:Global

* Function: TODO ADD FUNCTION.

* @author hxy

*/

public class Global {

public static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

}

5.实际处理类
以下处理类虽然做了注释，但是在这里还是详细讲解下。

这个类是单例的，每个线程处理会新实例化一个类。
每个成功的线程访问顺序：channelActive（开启连接）-handleHttpRequest（http握手处理）-messageReceived（消息接收处理）-handlerWebSocketFrame（实际处理，可以放到其他类里面分业务进行）
注意：这个demo中我做了路由功能，在handleHttpRequest中对每个channel连接的时候对每个连接的url进行绑定参数，然后在messageReceived中获取绑定的参数进行分发处理（handlerWebSocketFrame或handlerWebSocketFrame2），同时也获取了uri后置参数，有注释。
针对第三点路由分发，还有一种方法就是handshaker的uri（）方法，看源码即可，简单好用。
群发的时候遍历集合或者map的时候，必须每个channle都实例化一个TextWebSocketFrame对象，否则会报错或者发不出。

package com.nettywebsocket;

import java.util.Date;

import java.util.List;

import java.util.Map;

import java.util.logging.Level;

import java.util.logging.Logger;

import io.netty.buffer.ByteBuf;

import io.netty.buffer.Unpooled;

import io.netty.channel.ChannelFuture;

import io.netty.channel.ChannelFutureListener;

import io.netty.channel.ChannelHandlerContext;

import io.netty.channel.SimpleChannelInboundHandler;

import io.netty.handler.codec.http.DefaultFullHttpResponse;

import io.netty.handler.codec.http.FullHttpRequest;

import io.netty.handler.codec.http.HttpHeaders;

import io.netty.handler.codec.http.HttpMethod;

import io.netty.handler.codec.http.HttpResponseStatus;

import io.netty.handler.codec.http.HttpVersion;

import io.netty.handler.codec.http.QueryStringDecoder;

import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;

import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;

import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;

import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;

import io.netty.handler.codec.http.websocketx.WebSocketFrame;

import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;

import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;

import io.netty.util.AttributeKey;

import io.netty.util.CharsetUtil;

/**

* ClassName:MyWebSocketServerHandler Function: TODO ADD FUNCTION.

*

* @author hxy

*/

public class MyWebSocketServerHandler extends SimpleChannelInboundHandler<Object> {

private static final Logger logger = Logger.getLogger(WebSocketServerHandshaker.class.getName());

private WebSocketServerHandshaker handshaker;

/**

* channel 通道 action 活跃的 当客户端主动链接服务端的链接后，这个通道就是活跃的了。也就是客户端与服务端建立了通信通道并且可以传输数据

*/

@Override

public void channelActive(ChannelHandlerContext ctx) throws Exception {

// 添加

Global.group.add(ctx.channel());

System.out.println("客户端与服务端连接开启：" + ctx.channel().remoteAddress().toString());

}

/**

* channel 通道 Inactive 不活跃的 当客户端主动断开服务端的链接后，这个通道就是不活跃的。也就是说客户端与服务端关闭了通信通道并且不可以传输数据

*/

@Override

public void channelInactive(ChannelHandlerContext ctx) throws Exception {

// 移除

Global.group.remove(ctx.channel());

System.out.println("客户端与服务端连接关闭：" + ctx.channel().remoteAddress().toString());

}

/**

* 接收客户端发送的消息 channel 通道 Read 读 简而言之就是从通道中读取数据，也就是服务端接收客户端发来的数据。但是这个数据在不进行解码时它是ByteBuf类型的

*/

@Override

protected void messageReceived(ChannelHandlerContext ctx, Object msg) throws Exception {

// 传统的HTTP接入

if (msg instanceof FullHttpRequest) {

handleHttpRequest(ctx, ((FullHttpRequest) msg));

// WebSocket接入

} else if (msg instanceof WebSocketFrame) {

System.out.println(handshaker.uri());

if("anzhuo".equals(ctx.attr(AttributeKey.valueOf("type")).get())){

handlerWebSocketFrame(ctx, (WebSocketFrame) msg);

}else{

handlerWebSocketFrame2(ctx, (WebSocketFrame) msg);

}

}

}

/**

* channel 通道 Read 读取 Complete 完成 在通道读取完成后会在这个方法里通知，对应可以做刷新操作 ctx.flush()

*/

@Override

public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {

ctx.flush();

}

private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {

// 判断是否关闭链路的指令

if (frame instanceof CloseWebSocketFrame) {

System.out.println(1);

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)) {

System.out.println("本例程仅支持文本消息，不支持二进制消息");

throw new UnsupportedOperationException(

String.format("%s frame types not supported", frame.getClass().getName()));

}

// 返回应答消息

String request = ((TextWebSocketFrame) frame).text();

System.out.println("服务端收到：" + request);

if (logger.isLoggable(Level.FINE)) {

logger.fine(String.format("%s received %s", ctx.channel(), request));

}

TextWebSocketFrame tws = new TextWebSocketFrame(new Date().toString() + ctx.channel().id() + "：" + request);

// 群发

Global.group.writeAndFlush(tws);

// 返回【谁发的发给谁】

// ctx.channel().writeAndFlush(tws);

}

private void handlerWebSocketFrame2(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)) {

System.out.println("本例程仅支持文本消息，不支持二进制消息");

throw new UnsupportedOperationException(

String.format("%s frame types not supported", frame.getClass().getName()));

}

// 返回应答消息

String request = ((TextWebSocketFrame) frame).text();

System.out.println("服务端2收到：" + request);

if (logger.isLoggable(Level.FINE)) {

logger.fine(String.format("%s received %s", ctx.channel(), request));

}

TextWebSocketFrame tws = new TextWebSocketFrame(new Date().toString() + ctx.channel().id() + "：" + request);

// 群发

Global.group.writeAndFlush(tws);

// 返回【谁发的发给谁】

// ctx.channel().writeAndFlush(tws);

}

private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {

// 如果HTTP解码失败，返回HHTP异常

if (!req.getDecoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) {

sendHttpResponse(ctx, req,

new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));

return;

}

//获取url后置参数

HttpMethod method=req.getMethod();

String uri=req.getUri();

QueryStringDecoder queryStringDecoder = new QueryStringDecoder(uri);

Map<String, List<String>> parameters = queryStringDecoder.parameters();

System.out.println(parameters.get("request").get(0));

if(method==HttpMethod.GET&&"/webssss".equals(uri)){

//....处理

ctx.attr(AttributeKey.valueOf("type")).set("anzhuo");

}else if(method==HttpMethod.GET&&"/websocket".equals(uri)){

//...处理

ctx.attr(AttributeKey.valueOf("type")).set("live");

}

// 构造握手响应返回，本机测试

WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(

"ws://"+req.headers().get(HttpHeaders.Names.HOST)+uri, null, false);

handshaker = wsFactory.newHandshaker(req);

if (handshaker == null) {

WebSocketServerHandshakerFactory.sendUnsupportedWebSocketVersionResponse(ctx.channel());

} else {

handshaker.handshake(ctx.channel(), req);

}

}

private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) {

// 返回应答给客户端

if (res.getStatus().code() != 200) {

ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8);

res.content().writeBytes(buf);

buf.release();

}

// 如果是非Keep-Alive，关闭连接

ChannelFuture f = ctx.channel().writeAndFlush(res);

if (!HttpHeaders.isKeepAlive(req) || res.getStatus().code() != 200) {

f.addListener(ChannelFutureListener.CLOSE);

}

}

/**

* exception 异常 Caught 抓住 抓住异常，当发生异常的时候，可以做一些相应的处理，比如打印日志、关闭链接

*/

@Override

public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

cause.printStackTrace();

ctx.close();

}

}

以上就是netty-websocket的Demo了，应该已经解释的很详细了，同时应对的并发量也满足一般企业用于websocket的连接，如果需要不够，可以用nginx负载均衡增加。
最后给大家一条建议，在实际项目中，别让这种长连接一直保持，在nginx中可以设置连接无交流超时断开，大概设置10分钟左右，然后每8分钟定时从服务端发送一条心跳，具体想法就看你们喽~

巴特西

websocket（三）进阶！netty框架实现websocket达到高并发

引言：

什么是Netty

基于Netty的websocket压力测试

Demo详解

最新文章

热门文章

巴特西

websocket（三） 进阶！netty框架实现websocket达到高并发

引言：

什么是Netty

基于Netty的websocket压力测试

Demo详解

最新文章

热门文章

websocket（三）进阶！netty框架实现websocket达到高并发