// ai生成的demo
public class WebSocketServer {
private final int port;
public WebSocketServer(int port) {
this.port = port;
}
public void run() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
// HTTP 编解码器
pipeline.addLast(new HttpServerCodec());
// HTTP 消息聚合器
pipeline.addLast(new HttpObjectAggregator(65536));
// WebSocket 协议处理器
pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
// 自定义 WebSocket 处理器
pipeline.addLast(new WebSocketFrameHandler());
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true);
// 绑定端口并启动服务器
ChannelFuture f = b.bind(port).sync();
System.out.println("WebSocket 服务器启动，监听端口 " + port);
// 等待服务器 socket 关闭
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception {
int port = 8080;
new WebSocketServer(port).run();
}
}
public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception {
final HttpObject httpObject = (HttpObject) msg;
if (httpObject instanceof HttpRequest) {
final HttpRequest req = (HttpRequest) httpObject;
isWebSocketPath = isWebSocketPath(req);
if (!isWebSocketPath) {
ctx.fireChannelRead(msg);
return;
}
try {
final WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
getWebSocketLocation(ctx.pipeline(), req, serverConfig.websocketPath()),
serverConfig.subprotocols(), serverConfig.decoderConfig());
final WebSocketServerHandshaker handshaker = wsFactory.newHandshaker(req);
final ChannelPromise localHandshakePromise = handshakePromise;
if (handshaker == null) {
WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
} else {
WebSocketServerProtocolHandler.setHandshaker(ctx.channel(), handshaker);
ctx.pipeline().remove(this);
final ChannelFuture handshakeFuture = handshaker.handshake(ctx.channel(), req);
handshakeFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) {
if (!future.isSuccess()) {
localHandshakePromise.tryFailure(future.cause());
ctx.fireExceptionCaught(future.cause());
} else {
localHandshakePromise.trySuccess();
ctx.fireUserEventTriggered(
WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE);
// 发送握手成功事件
ctx.fireUserEventTriggered(
new WebSocketServerProtocolHandler.HandshakeComplete(
req.uri(), req.headers(), handshaker.selectedSubprotocol()));
}
}
});
applyHandshakeTimeout();
}
} finally {
ReferenceCountUtil.release(req);
}
} else if (!isWebSocketPath) {
ctx.fireChannelRead(msg);
} else {
ReferenceCountUtil.release(msg);
}
}
public WebSocketServerProtocolHandler(String websocketPath, String subprotocols,
boolean allowExtensions, int maxFrameSize, boolean allowMaskMismatch,
boolean checkStartsWith, boolean dropPongFrames) {
this(websocketPath, subprotocols, allowExtensions, maxFrameSize, allowMaskMismatch, checkStartsWith,
dropPongFrames, DEFAULT_HANDSHAKE_TIMEOUT_MILLIS);
}
@Component
public class CustomWebSocketExtensionFilter implements WebSocketExtensionFilter {
@Resource
private WebSocketConfig webSocketConfig;
@Override
public boolean mustSkip(WebSocketFrame frame) {
if(frame instanceof TextWebSocketFrame || frame instanceof BinaryWebSocketFrame) {
// 小于一定长度的消息则跳过压缩，返回值代表是否跳过压缩
if(frame.content().readableBytes() < webSocketConfig.getMinCompressionLength()) {
return true;
}
}
return false;
}
}
@Component
public class WebSocketServerExtensionHandlerFactory {
@Resource
private CustomWebSocketExtensionFilter customWebSocketExtensionFilter;
private CustomWebSocketExtensionFilterProvider customWebSocketExtensionFilterProvider = new CustomWebSocketExtensionFilterProvider();
public WebSocketServerExtensionHandler newInstance() {
return new WebSocketServerExtensionHandler(new PerMessageDeflateServerExtensionHandshaker(6,
ZlibCodecFactory.isSupportingWindowSizeAndMemLevel(),
PerMessageDeflateServerExtensionHandshaker.MAX_WINDOW_SIZE,
false, false,
customWebSocketExtensionFilterProvider
));
}
private class CustomWebSocketExtensionFilterProvider implements WebSocketExtensionFilterProvider {
@Override
public WebSocketExtensionFilter encoderFilter() {
return customWebSocketExtensionFilter;
}
@Override
public WebSocketExtensionFilter decoderFilter() {
return WebSocketExtensionFilter.NEVER_SKIP;
}
}
}
public class WritableHandler extends ChannelOutboundHandlerAdapter {
@Resource
private WebSocketConfig webSocketConfig;
@Resource
private ChannelStore channelStore;
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
if (!ctx.channel().isWritable()) {
Channel channel = ctx.channel();
// 释放掉消息，不进行推送
ReferenceCountUtil.safeRelease(msg);
if (webSocketConfig.isCloseNotWritableChannel() && ctx.channel().isOpen()) {
// 根据配置决定是否关闭这个channel
Monitor.counter("closeNotWritableChannel").end();
channelStore.closeChannel(channel, CloseReason.NOT_WRITABLE);
}
return;
}
super.write(ctx, msg, promise);
}
}
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。