使用Netty搭建WebSocket服务器
1.WebSocketServer.java
public class WebSocketServer { private final ChannelGroup group = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE); private final EventLoopGroup workerGroup = new NioEventLoopGroup(); private Channel channel; public ChannelFuture start(InetSocketAddress address) { ServerBootstrap boot = new ServerBootstrap(); boot.group(workerGroup).channel(NioServerSocketChannel.class).childHandler(createInitializer(group)); ChannelFuture f = boot.bind(address).syncUninterruptibly(); channel = f.channel(); return f; } protected ChannelHandler createInitializer(ChannelGroup group2) { return new ChatServerInitializer(group2); } public void destroy() { if (channel != null) channel.close(); group.close(); workerGroup.shutdownGracefully(); } public static void main(String[] args) { final WebSocketServer server = new WebSocketServer(); ChannelFuture f = server.start(new InetSocketAddress(2048)); System.out.println("server start................"); Runtime.getRuntime().addShutdownHook(new Thread() { @Override public void run() { server.destroy(); } }); f.channel().closeFuture().syncUninterruptibly(); } private static WebSocketServer instance; private WebSocketServer() {} public static synchronized WebSocketServer getInstance() {// 懒汉,线程安全 if (instance == null) { instance = new WebSocketServer(); } return instance; } public void running(){ if(instance != null){ String port=null; port=BusinessConfigUtils.findProperty("websocket_port");//获取端口号 if(null==port||port.length()<0||!StringUtils.isNumeric(port)){ port="18080"; } instance.start(new InetSocketAddress(Integer.valueOf(port))); //ChannelFuture f = System.out.println("----------------------------------------WEBSOCKET SERVER START----------------------------------------"); /*Runtime.getRuntime().addShutdownHook(new Thread() { @Override public void run() { instance.destroy(); } }); f.channel().closeFuture().syncUninterruptibly();*/ } } }
2.ChatServerInitializer.java
public class ChatServerInitializer extends ChannelInitializer<Channel> { private final ChannelGroup group; public ChatServerInitializer(ChannelGroup group) { super(); this.group = group; } @Override protected void initChannel(Channel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast(new HttpServerCodec()); pipeline.addLast(new ChunkedWriteHandler()); pipeline.addLast(new HttpObjectAggregator(64*1024)); pipeline.addLast(new HttpRequestHandler("/ws")); pipeline.addLast(new WebSocketServerProtocolHandler("/ws")); pipeline.addLast(new TextWebSocketFrameHandler(group)); } }
3. HttpRequestHandler.java
public class HttpRequestHandler extends SimpleChannelInboundHandler<FullHttpRequest> { private LoginTimeService loginTimeService = SpringContextHolder.getBean("loginTimeServiceImpl"); private final String wsUri; public HttpRequestHandler(String wsUri) { super(); this.wsUri = wsUri; } @Override @SuppressWarnings("deprecation") protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception { if (wsUri.equalsIgnoreCase(msg.getUri().substring(0, 3))) { String userId = findUserIdByUri(msg.getUri()); if (userId != null && userId.trim() != null && userId.trim().length() > 0) { ctx.channel().attr(AttributeKey.valueOf(ctx.channel().id().asShortText())).set(userId);// 写userid值 UserIdToWebSocketChannelShare.userIdToWebSocketChannelMap.put(userId, ctx.channel()); // 用户Id与Channel绑定 loginTimeService.onLine(userId, new Date());// 统计上线记录 } else { }// 没有获取到用户Id ctx.fireChannelRead(msg.setUri(wsUri).retain()); } } private String findUserIdByUri(String uri) {// 通过Uid获取用户Id--uri中包含userId String userId = ""; try { userId = uri.substring(uri.indexOf("userId") + 7); if (userId != null && userId.trim() != null && userId.trim().length() > 0) { userId = userId.trim(); } } catch (Exception e) { } return userId; } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.close(); cause.printStackTrace(System.err); } }
4. TextWebSocketFrameHandler.java
public class TextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> { private LoginTimeService loginTimeService = SpringContextHolder.getBean("loginTimeServiceImpl"); private final ChannelGroup group; public TextWebSocketFrameHandler(ChannelGroup group) { super(); this.group = group; } @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE) { ctx.pipeline().remove(HttpRequestHandler.class); // group.writeAndFlush(""); group.add(ctx.channel()); } else { super.userEventTriggered(ctx, evt); } } @Override protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception { group.writeAndFlush(msg.retain()); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.close(); cause.printStackTrace(); } @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { // (6) Channel incoming = ctx.channel(); String userId = (String) incoming.attr(AttributeKey.valueOf(incoming.id().asShortText())).get(); UserIdToWebSocketChannelShare.userIdToWebSocketChannelMap.remove(userId);// 删除缓存的通道 loginTimeService.outLine(userId, new Date());// 下线通过 } }
时间: 2024-10-11 00:09:53