引言：
在前面兩篇文章中，我們對原生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分鍾定時從服務端發送一條心跳，具體想法就看你們嘍~