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();
}
}

 ---------------------