怎么做有数据库的网站,大连做网站科技有限公司,大芬网站建设,阳江做网站seo本文基于Netty实现WebSocket服务端#xff0c;实现和客户端的交互通信#xff0c;客户端基于JavaScript实现。
在【WebSocket简介-CSDN博客】中#xff0c;我们知道WebSocket是基于Http协议的升级#xff0c;而Netty提供了Http和WebSocket Frame的编解码器和Handler#…
本文基于Netty实现WebSocket服务端实现和客户端的交互通信客户端基于JavaScript实现。
在【WebSocket简介-CSDN博客】中我们知道WebSocket是基于Http协议的升级而Netty提供了Http和WebSocket Frame的编解码器和Handler我们可以基于Netty快速实现WebSocket服务端。
一、基于Netty快速实现WebSocket服务端
我们可以直接利用Netty提供了Http和WebSocket Frame的编解码器和Handler快速启动一个WebSocket服务端。服务端收到Client消息后转发给其他客户端。
服务端代码
ChannelSupervise用户存储客户端信息
import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;/*** 用户存储客户端信息*/
public class ChannelSupervise {private static ChannelGroup GlobalGroup new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);private static ConcurrentMapString, ChannelId ChannelMap new ConcurrentHashMap();public static void addChannel(Channel channel) {GlobalGroup.add(channel);ChannelMap.put(channel.id().asShortText(), channel.id());}public static void removeChannel(Channel channel) {GlobalGroup.remove(channel);ChannelMap.remove(channel.id().asShortText());}public static Channel findChannel(String id) {return GlobalGroup.find(ChannelMap.get(id));}public static void send2All(TextWebSocketFrame tws) {GlobalGroup.writeAndFlush(tws);}
}
服务端Netty配置
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.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.stream.ChunkedWriteHandler;/*** * 实现长链接 客户端与服务端*/
public class SimpleWsChatServer {public static void main(String[] args) throws InterruptedException {EventLoopGroup bossGroup new NioEventLoopGroup(1);EventLoopGroup workGroup new NioEventLoopGroup();try {ServerBootstrap serverBootstrap new ServerBootstrap();serverBootstrap.group(bossGroup, workGroup).channel(NioServerSocketChannel.class).// 在 bossGroup 增加一个日志处理器handler(new LoggingHandler(LogLevel.INFO)).childHandler(new ChannelInitializerSocketChannel() {Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {ChannelPipeline pipeline socketChannel.pipeline();// 基于http协议的长连接 需要使用http协议的解码 编码器pipeline.addLast(new HttpServerCodec());// 以块的方式处理pipeline.addLast(new ChunkedWriteHandler());/*** http数据传输过程中是分段 HttpObjectAggregator 将多个段聚合起来* 当浏览器发起大量数据的时候会发起多次http请求*/pipeline.addLast(new HttpObjectAggregator(8192));/*** 对于websocket是以frame的形式传递* WebSocketFrame* 浏览器 ws://localhost:7070/ 不在是http协议* WebSocketServerProtocolHandler 将http协议升级为ws协议 即保持长链接*/pipeline.addLast(new WebSocketServerProtocolHandler(/helloWs));// 自定义handler专门处理浏览器请求pipeline.addLast(new MyTextWebSocketFrameHandler());}});ChannelFuture channelFuture serverBootstrap.bind(7070).sync();channelFuture.channel().closeFuture().sync();} finally {bossGroup.shutdownGracefully();workGroup.shutdownGracefully();}}
}
消息转发逻辑
import com.huawei.websocket.nio2.global.ChannelSupervise;import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;import java.text.SimpleDateFormat;
import java.util.Date;/*** TextWebSocketFrame 表示一个文本贞* 浏览器和服务端以TextWebSocketFrame 格式交互*/
public class MyTextWebSocketFrameHandler extends SimpleChannelInboundHandlerTextWebSocketFrame {private SimpleDateFormat sdf new SimpleDateFormat(yyyy-mm-dd hh:MM:ss);Overrideprotected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame)throws Exception {System.out.println(服务端收到消息 textWebSocketFrame.text());// 回复浏览器String resp sdf.format(new Date()) : textWebSocketFrame.text();// 转发给所有的客户端ChannelSupervise.send2All(new TextWebSocketFrame(resp));// 发送给当前客户单// channelHandlerContext.channel().writeAndFlush(new TextWebSocketFrame(resp));}Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {// 添加连接System.out.println(客户端加入连接 ctx.channel());ChannelSupervise.addChannel(ctx.channel());}Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {// 断开连接System.out.println(客户端断开连接 ctx.channel());ChannelSupervise.removeChannel(ctx.channel());}Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {System.out.println(异常发生 cause.getMessage());ctx.channel().close();}
} 参考NIO框架NettyWebSocket实现网页聊天_nio实现websocket-CSDN博客 一、基于Netty手动处理WebSocket握手信息
参考GitCode - 开发者的代码家园https://gitcode.com/Siwash/websocketWithNetty/blob/master/README.md 基于netty搭建websocket实现消息的主动推送_websocket_rpf_siwash-GitCode 开源社区 服务端代码
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
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.logging.LoggingHandler;
import io.netty.handler.stream.ChunkedWriteHandler;import org.apache.log4j.Logger;/*** https://gitcode.com/Siwash/websocketWithNetty/blob/master/README.md*/
public class NioWebSocketServer {private final Logger logger Logger.getLogger(getClass());private void init() {logger.info(正在启动websocket服务器);NioEventLoopGroup boss new NioEventLoopGroup();NioEventLoopGroup work new NioEventLoopGroup();try {ServerBootstrap bootstrap new ServerBootstrap();bootstrap.group(boss, work);bootstrap.channel(NioServerSocketChannel.class);bootstrap.childHandler(new ChannelInitializerSocketChannel() {Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(logging, new LoggingHandler(DEBUG));// 设置log监听器并且日志级别为debug方便观察运行流程ch.pipeline().addLast(http-codec, new HttpServerCodec());// 设置解码器ch.pipeline().addLast(aggregator, new HttpObjectAggregator(65536));// 聚合器使用websocket会用到ch.pipeline().addLast(http-chunked, new ChunkedWriteHandler());// 用于大数据的分区传输ch.pipeline().addLast(handler, new NioWebSocketHandler());// 自定义的业务handler,这里处理WebSocket建链请求和消息发送请求}});Channel channel bootstrap.bind(7070).sync().channel();logger.info(webSocket服务器启动成功 channel);channel.closeFuture().sync();} catch (InterruptedException e) {e.printStackTrace();logger.info(运行出错 e);} finally {boss.shutdownGracefully();work.shutdownGracefully();logger.info(websocket服务器已关闭);}}public static void main(String[] args) {new NioWebSocketServer().init();}
}
import static io.netty.handler.codec.http.HttpUtil.isKeepAlive;import com.huawei.websocket.nio2.global.ChannelSupervise;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.util.CharsetUtil;import org.apache.log4j.Logger;import java.util.Date;/*** 这里处理WebSocket建链请求和消息发送请求*/
public class NioWebSocketHandler extends SimpleChannelInboundHandlerObject {private final Logger logger Logger.getLogger(getClass());private WebSocketServerHandshaker handshaker;Overrideprotected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {logger.debug(收到消息 msg);if (msg instanceof FullHttpRequest) {// 以http请求形式接入但是走的是websockethandleHttpRequest(ctx, (FullHttpRequest) msg);} else if (msg instanceof WebSocketFrame) {// 处理websocket客户端的消息handlerWebSocketFrame(ctx, (WebSocketFrame) msg);}}Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {// 添加连接logger.debug(客户端加入连接 ctx.channel());ChannelSupervise.addChannel(ctx.channel());}Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {// 断开连接logger.debug(客户端断开连接 ctx.channel());ChannelSupervise.removeChannel(ctx.channel());}Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {ctx.flush();}private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {// 判断是否关闭链路的指令if (frame instanceof CloseWebSocketFrame) {handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());return;}// 判断是否ping消息if (frame instanceof PingWebSocketFrame) {logger.debug(服务端收到ping消息);ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));return;}// 本例程仅支持文本消息不支持二进制消息if (!(frame instanceof TextWebSocketFrame)) {logger.debug(本例程仅支持文本消息不支持二进制消息);throw new UnsupportedOperationException(String.format(%s frame types not supported, frame.getClass().getName()));}// 返回应答消息String request ((TextWebSocketFrame) frame).text();logger.debug(服务端收到 request);TextWebSocketFrame tws new TextWebSocketFrame(new Date().toString() ctx.channel().id() request);// 群发ChannelSupervise.send2All(tws);// 返回【谁发的发给谁】// ctx.channel().writeAndFlush(tws);}/*** 唯一的一次http请求用于创建websocket* */private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {// 要求Upgrade为websocket过滤掉get/Postif (!req.decoderResult().isSuccess() || (!websocket.equals(req.headers().get(Upgrade)))) {// 若不是websocket方式则创建BAD_REQUEST的req返回给客户端sendHttpResponse(ctx, req,new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));return;}logger.debug(服务端收到WebSocket建链消息);WebSocketServerHandshakerFactory wsFactory new WebSocketServerHandshakerFactory(ws://localhost:7070/helloWs, null, false);handshaker wsFactory.newHandshaker(req);if (handshaker null) {WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());} else {handshaker.handshake(ctx.channel(), req);}}/*** 拒绝不合法的请求并返回错误信息* */private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) {// 返回应答给客户端if (res.status().code() ! 200) {ByteBuf buf Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);res.content().writeBytes(buf);buf.release();}ChannelFuture f ctx.channel().writeAndFlush(res);// 如果是非Keep-Alive关闭连接if (!isKeepAlive(req) || res.status().code() ! 200) {f.addListener(ChannelFutureListener.CLOSE);}}
} 这里我们手动处理了握手信息
可以看到服务端handler日志
2024-05-24 09:43:51 DEBUG [NioWebSocketHandler] 收到消息HttpObjectAggregator$AggregatedFullHttpRequest(decodeResult: success, version: HTTP/1.1, content: CompositeByteBuf(ridx: 0, widx: 0, cap: 0, components0))
GET /helloWs HTTP/1.1
Host: localhost:7070
Connection: Upgrade
Pragma: no-cache
Cache-Control: no-cache
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 Edg/124.0.0.0
Upgrade: websocket
Origin: null
Sec-WebSocket-Version: 13
Accept-Encoding: gzip, deflate, br, zstd
Accept-Language: zh-CN,zh;q0.9,en;q0.8,en-GB;q0.7,en-US;q0.6
Sec-WebSocket-Key: KYh4//SBKLinSu6v1kYqw
Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits
content-length: 0
2024-05-24 09:43:51 DEBUG [NioWebSocketHandler] 服务端收到WebSocket建链消息 Client1的发送和接收消息在F12小可以看到 测试用的Client代码如下
!DOCTYPE html
html langenheadmeta charsetUTF-8titleWebSocket Client/title/headbodyscriptvar socket;//check current explorer wether support WebSockekif (window.WebSocket) {socket new WebSocket(ws://localhost:7070/helloWs);//event : msg from serversocket.onmessage function(event) {console.log(receive msg: event.data);var rt document.getElementById(responseText);rt.value rt.value \n event.data;}//eq open WebSocketsocket.onopen function(event) {var rt document.getElementById(responseText);rt.value open WebSocket;}socket.onclose function(event) {var rt document.getElementById(responseText);rt.value rt.value \n close WebSocket;}} else {alert(current exploer not support websocket)}//send msg to WebSocket Serverfunction send(message) {if (socket.readyState WebSocket.OPEN) {//send msg to WebSocket by socketsocket.send(message);} else {alert(WebSocket is not open)}}/scriptform onsubmitreturn falsetextarea namemessage styleheight: 300px; width: 300px;/textareainput typebutton value发送消息 onclicksend(this.form.message.value)textarea idresponseText styleheight: 300px; width: 300px;/textareainput typebutton value清空内容 onclickdocument.getElementById(responseText).value/form/body
/html
直接保存为html文件使用浏览器打开即可运行测试
