添加MAVEN依賴
<!-- Netty -->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.45.Final</version>
</dependency>
WebSocketProperties配置類
為了能在SpringBoot項目中靈活配置相關的值访忿，這里使用了配置類橱乱，并使用了默認值柠掂。
@Getter
@Setter
@Component
@ConfigurationProperties(prefix = "chat.websocket")
public class WebSocketProperties {
private String host = "localhost"; // 監(jiān)聽地址
private Integer port = 8081; // 監(jiān)聽端口
private String path = "/ws"; // 請求路徑
}
WebSocket連接通道池
用來管理已經(jīng)連接的客戶端通道扮念，方便數(shù)據(jù)傳輸交互。
@Slf4j
@Component
public class NioWebSocketChannelPool {
private final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
/**
* 新增一個客戶端通道
*
* @param channel
*/
public void addChannel(Channel channel) {
channels.add(channel);
}
/**
* 移除一個客戶端連接通道
*
* @param channel
*/
public void removeChannel(Channel channel) {
channels.remove(channel);
}
}
WebSocket連接數(shù)據(jù)接收處理類
@Slf4j
@Sharable
@Component
public class NioWebSocketHandler extends SimpleChannelInboundHandler<WebSocketFrame> {
@Autowired
private NioWebSocketChannelPool webSocketChannelPool;
@Autowired
private WebSocketProperties webSocketProperties;
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
log.debug("客戶端連接：{}", ctx.channel().id());
webSocketChannelPool.addChannel(ctx.channel());
super.channelActive(ctx);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
log.debug("客戶端斷開連接：{}", ctx.channel().id());
webSocketChannelPool.removeChannel(ctx.channel());
super.channelInactive(ctx);
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
ctx.channel().flush();
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) {
if (frame instanceof PingWebSocketFrame) {
pingWebSocketFrameHandler(ctx, (PingWebSocketFrame) frame);
} else if (frame instanceof TextWebSocketFrame) {
textWebSocketFrameHandler(ctx, (TextWebSocketFrame) frame);
} else if (frame instanceof CloseWebSocketFrame) {
closeWebSocketFrameHandler(ctx, (CloseWebSocketFrame) frame);
}
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("數(shù)據(jù)類型：{}", msg.getClass());
if (msg instanceof FullHttpRequest) {
fullHttpRequestHandler(ctx, (FullHttpRequest) msg);
}
super.channelRead(ctx, msg);
}
/**
* 處理連接請求麦箍，客戶端WebSocket發(fā)送握手包時會執(zhí)行這一次請求
*
* @param ctx
* @param request
*/
private void fullHttpRequestHandler(ChannelHandlerContext ctx, FullHttpRequest request) {
String uri = request.uri();
log.debug("接收到客戶端的握手包：{}", ctx.channel().id());
Map<String, String> params = RequestUriUtils.getParams(uri);
log.debug("客戶端請求參數(shù)：{}", params);
if (uri.startsWith(webSocketProperties.getPath()))
request.setUri(webSocketProperties.getPath());
else
ctx.close();
}
/**
* 客戶端發(fā)送斷開請求處理
*
* @param ctx
* @param frame
*/
private void closeWebSocketFrameHandler(ChannelHandlerContext ctx, CloseWebSocketFrame frame) {
log.debug("接收到主動斷開請求：{}", ctx.channel().id());
ctx.close();
}
/**
* 創(chuàng)建連接之后应闯，客戶端發(fā)送的消息都會在這里處理
*
* @param ctx
* @param frame
*/
private void textWebSocketFrameHandler(ChannelHandlerContext ctx, TextWebSocketFrame frame) {
String text = frame.text();
log.debug("接收到客戶端的消息：{}", text);
// 將客戶端消息回送給客戶端
ctx.channel().writeAndFlush(new TextWebSocketFrame("你發(fā)送的內(nèi)容是：" + text));
}
/**
* 處理客戶端心跳包
*
* @param ctx
* @param frame
*/
private void pingWebSocketFrameHandler(ChannelHandlerContext ctx, PingWebSocketFrame frame) {
ctx.channel().writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
}
}
WebSocket通道連接初始化
@Component
public class NioWebSocketChannelInitializer extends ChannelInitializer<SocketChannel> {
@Autowired
private WebSocketProperties webSocketProperties;
@Autowired
private NioWebSocketHandler nioWebSocketHandler;
@Override
protected void initChannel(SocketChannel socketChannel) {
socketChannel.pipeline()
.addLast(new HttpServerCodec())
.addLast(new ChunkedWriteHandler())
.addLast(new HttpObjectAggregator(8192))
.addLast(nioWebSocketHandler)
.addLast(new WebSocketServerProtocolHandler(webSocketProperties.getPath(), null, true, 65536));
}
}
Netty服務端
服務器的初始化和銷毀都交給Spring容器期丰。
@Slf4j
@Component
public class NioWebSocketServer implements InitializingBean, DisposableBean {
@Autowired
private WebSocketProperties webSocketProperties;
@Autowired
private NioWebSocketChannelInitializer webSocketChannelInitializer;
private EventLoopGroup bossGroup;
private EventLoopGroup workGroup;
private ChannelFuture channelFuture;
@Override
public void afterPropertiesSet() throws Exception {
try {
bossGroup = new NioEventLoopGroup(webSocketProperties.getBoss());
workGroup = new NioEventLoopGroup(webSocketProperties.getWork());
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.option(ChannelOption.SO_BACKLOG, 1024)
.group(bossGroup, workGroup)
.channel(NioServerSocketChannel.class)
.localAddress(webSocketProperties.getPort())
.childHandler(webSocketChannelInitializer);
channelFuture = serverBootstrap.bind().sync();
} finally {
if (channelFuture != null && channelFuture.isSuccess()) {
log.info("Netty server startup on port: {} (websocket) with context path '{}'", webSocketProperties.getPort(), webSocketProperties.getPath());
} else {
log.error("Netty server startup failed.");
if (bossGroup != null)
bossGroup.shutdownGracefully().sync();
if (workGroup != null)
workGroup.shutdownGracefully().sync();
}
}
}
@Override
public void destroy() throws Exception {
log.info("Shutting down Netty server...");
if (bossGroup != null && !bossGroup.isShuttingDown() && !bossGroup.isTerminated())
bossGroup.shutdownGracefully().sync();
if (workGroup != null && !workGroup.isShuttingDown() && !workGroup.isTerminated())
workGroup.shutdownGracefully().sync();
if (channelFuture != null && channelFuture.isSuccess())
channelFuture.channel().closeFuture().sync();
log.info("Netty server shutdown.");
}
}
到此护侮，代碼編寫完成了敌完。
