网站建设与网页设计专业的,wordpress主题 评论,网站备案起名要求,怎样拓展客户除了构建TCP和UDP服务器和客户端#xff0c;Netty还可以用于构建WebSocket服务器。WebSocket是一种基于TCP协议的双向通信协议#xff0c;可以在Web浏览器和Web服务器之间建立实时通信通道。下面是一个简单的示例#xff0c;演示如何使用Netty构建一个WebSocket服务器。 项目…除了构建TCP和UDP服务器和客户端Netty还可以用于构建WebSocket服务器。WebSocket是一种基于TCP协议的双向通信协议可以在Web浏览器和Web服务器之间建立实时通信通道。下面是一个简单的示例演示如何使用Netty构建一个WebSocket服务器。 项目目录 引入pom依赖 dependencygroupIdio.netty/groupIdartifactIdnetty-all/artifactIdversion4.1.69.Final/version/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdoptionaltrue/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() {// 创建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 ChannelInitializerSocketChannel() {Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {ChannelPipeline pipeline socketChannel.pipeline();pipeline.addLast(new HttpServerCodec());// 最大数据长度pipeline.addLast(new HttpObjectAggregator(65536));// 添加接收websocket请求的url匹配路径pipeline.addLast(new WebSocketServerProtocolHandler(/websocket));// 10秒内收不到消息强制断开连接// 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 的关闭sync 方法作用是同步等待 channel 关闭// 阻塞future.channel().closeFuture().sync();} catch (Exception e) {log.error(websocket server exception, e);throw new RuntimeException(e);} finally {log.info(websocket server close);// 关闭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 SimpleChannelInboundHandlerObject {private WebSocketServerHandshaker webSocketServerHandshaker;private static final String WEB_SOCKET_URL ws://127.0.0.1:8888/websocket;Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {// 创建连接时执行NettyConfig.group.add(ctx.channel());log.info(client channel active, id{}, ctx.channel().id().toString());}Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {// 关闭连接时执行NettyConfig.group.remove(ctx.channel());log.info(client channel disconnected, id{}, ctx.channel().id().toString());}Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {// 服务端接收客户端发送过来的数据结束之后调用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连接业务handlerWebSocketFrame(ctx, (WebSocketFrame) msg);}}/*** 处理websocket连接业务** param ctx* param frame*/private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) throws Exception {log.info(handlerWebSocketFrameclass{}, frame.getClass().getName());// 判断是否是关闭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);/*** 可通过客户传输的text设计处理策略* 如text{type: messageHandler, userId: 111}* 服务端根据type采用策略模式自行派发处理** 注意这里不需要使用线程池因为Netty 采用 Reactor线程模型目前使用的是主从Reactor模型* Handler已经是线程处理每个用户的请求是线程隔离的*/// 返回WebSocket响应ctx.writeAndFlush(new TextWebSocketFrame(server return: text));/*// 群发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(handlerHttpRequestclass{}, request.getClass().getName());// 判断是否采用WebSocket协议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();}// 服务端向客户端发送数据ChannelFuture f ctx.channel().writeAndFlush(response);if (response.getStatus().code() ! 200) {f.addListener(ChannelFutureListener.CLOSE);}}Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {// 非正常断开时调用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测试
