除了構(gòu)建TCP和UDP服務(wù)器和客戶端，Netty還可以用于構(gòu)建WebSocket服務(wù)器。WebSocket是一種基于TCP協(xié)議的雙向通信協(xié)議，可以在Web瀏覽器和Web服務(wù)器之間建立實時通信通道。下面是一個簡單的示例，演示如何使用Netty構(gòu)建一個WebSocket服務(wù)器。
項目目錄：

引入pom依賴：

 <dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.69.Final</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency>

編寫SocketServer：

package com.lzq.websocket.config;import com.lzq.websocket.handlers.WebSocketHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
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.timeout.ReadTimeoutHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Configuration;import java.util.concurrent.TimeUnit;@Slf4j
@Configuration
public class WebSocketConfig implements CommandLineRunner {private static final Integer PORT = 8888;@Overridepublic void run(String... args) throws Exception {new WebSocketConfig().start();}public void start() {// 創(chuàng)建EventLoopGroupEventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workerGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2);try {ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {ChannelPipeline pipeline = socketChannel.pipeline();pipeline.addLast(new HttpServerCodec());// 最大數(shù)據(jù)長度pipeline.addLast(new HttpObjectAggregator(65536));// 添加接收websocket請求的url匹配路徑pipeline.addLast(new WebSocketServerProtocolHandler("/websocket"));// 10秒內(nèi)收不到消息強制斷開連接// pipeline.addLast(new ReadTimeoutHandler(10, TimeUnit.SECONDS));pipeline.addLast(new WebSocketHandler());}});ChannelFuture future = serverBootstrap.bind(PORT).sync();log.info("websocket server started, port={}", PORT);// 處理 channel 的關(guān)閉，sync 方法作用是同步等待 channel 關(guān)閉// 阻塞future.channel().closeFuture().sync();} catch (Exception e) {log.error("websocket server exception", e);throw new RuntimeException(e);} finally {log.info("websocket server close");// 關(guān)閉EventLoopGroupbossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}

編寫WebSocketHandler：

package com.lzq.websocket.handlers;import com.lzq.websocket.config.NettyConfig;
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.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
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.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import lombok.extern.slf4j.Slf4j;import java.nio.charset.StandardCharsets;@Slf4j
public class WebSocketHandler extends SimpleChannelInboundHandler<Object> {private WebSocketServerHandshaker webSocketServerHandshaker;private static final String WEB_SOCKET_URL = "ws://127.0.0.1:8888/websocket";@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {// 創(chuàng)建連接時執(zhí)行NettyConfig.group.add(ctx.channel());log.info("client channel active, id={}", ctx.channel().id().toString());}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {// 關(guān)閉連接時執(zhí)行NettyConfig.group.remove(ctx.channel());log.info("client channel disconnected, id={}", ctx.channel().id().toString());}@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {// 服務(wù)端接收客戶端發(fā)送過來的數(shù)據(jù)結(jié)束之后調(diào)用ctx.flush();}@Overridepublic void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete) {WebSocketServerProtocolHandler.HandshakeComplete handshake = (WebSocketServerProtocolHandler.HandshakeComplete) evt;log.info("client channel connected, id={}, url={}", ctx.channel().id().toString(), handshake.requestUri());}}@Overrideprotected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {if (msg instanceof FullHttpRequest) {// 處理客戶端http握手請求handlerHttpRequest(ctx, (FullHttpRequest) msg);} else if (msg instanceof WebSocketFrame) {// 處理websocket連接業(yè)務(wù)handlerWebSocketFrame(ctx, (WebSocketFrame) msg);}}/*** 處理websocket連接業(yè)務(wù)** @param ctx* @param frame*/private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) throws Exception {log.info("handlerWebSocketFrame>>>>class={}", frame.getClass().getName());// 判斷是否是關(guān)閉websocket的指令if (frame instanceof CloseWebSocketFrame) {webSocketServerHandshaker.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)) {throw new RuntimeException("不支持消息類型：" + frame.getClass().getName());}String text = ((TextWebSocketFrame) frame).text();if ("ping".equals(text)) {ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));return;}log.info("WebSocket message received: {}", text);/*** 可通過客戶傳輸?shù)膖ext，設(shè)計處理策略：* 如：text={"type": "messageHandler", "userId": "111"}* 服務(wù)端根據(jù)type，采用策略模式，自行派發(fā)處理** 注意：這里不需要使用線程池，因為Netty 采用 Reactor線程模型（目前使用的是主從Reactor模型），* Handler已經(jīng)是線程處理，每個用戶的請求是線程隔離的*/// 返回WebSocket響應(yīng)ctx.writeAndFlush(new TextWebSocketFrame("server return:" + text));/*// 群發(fā)TextWebSocketFrame twsf = new TextWebSocketFrame(new Date().toString()+ ctx.channel().id()+ " : "+ text);NettyConfig.group.writeAndFlush(twsf);*/}/*** 處理客戶端http握手請求** @param ctx* @param request*/private void handlerHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request) {log.info("handlerHttpRequest>>>>class={}", request.getClass().getName());// 判斷是否采用WebSocket協(xié)議if (!request.getDecoderResult().isSuccess() || !("websocket".equals(request.headers().get("Upgrade")))) {sendHttpResponse(ctx, request, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));return;}WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(WEB_SOCKET_URL, null, false);webSocketServerHandshaker = wsFactory.newHandshaker(request);if (webSocketServerHandshaker == null) {WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());} else {webSocketServerHandshaker.handshake(ctx.channel(), request);}}private void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest request, DefaultFullHttpResponse response) {if (response.getStatus().code() != 200) {ByteBuf buf = Unpooled.copiedBuffer(response.getStatus().toString(), StandardCharsets.UTF_8);response.content().writeBytes(buf);buf.release();}// 服務(wù)端向客戶端發(fā)送數(shù)據(jù)ChannelFuture f = ctx.channel().writeAndFlush(response);if (response.getStatus().code() != 200) {f.addListener(ChannelFutureListener.CLOSE);}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {// 非正常斷開時調(diào)用log.error("client channel execute exception, id={}", ctx.channel().id().toString(), cause);ctx.close();}
}

NettyConfig：

package com.lzq.websocket.config;import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;public class NettyConfig {/*** 存儲接入的客戶端的channel對象*/public static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
}

使用Apifox測試：