Spring-Boot快速集成netty-socketio(socket服务实现，支持认证)_Java

Spring-Boot快速集成netty-socketio(socket服务实现，支持认证)

netty-socketio是一个开源的Socket.io服务器端的一个java的实现，它基于Netty框架，可用于服务端推送消息给客户端。

说到服务端推送技术，一般会涉及WebSocket，WebSocket是HTML5最新提出的规范，虽然主流浏览器都已经支持，但仍然可能有不兼容的情况，为了兼容所有浏览器，给程序员提供一致的编程体验，SocketIO将WebSocket、AJAX和其它的通信方式全部封装成了统一的通信接口，也就是说，使用SocketIO时不用担心兼容问题，底层会自动选用最佳的通信

基于Tomcat的webSocket的并发量很低，所以使用netty框架的socket性能更好，反应更快，兼容更好，而且自带认证

netty-socketio 框架事件流程

在这里插入图片描述

socket服务配置

yml配置

#socket服务配置 club.dlblog.socketio: host: localhost #IP地址 port: 9000 #端口号

socket服务配置类

/**
 * socket服务配置
 * @author machenike
 */ @Configuration public class NettySocketConfig { @Value("${club.dlblog.socketio.host}") private String host; @Value("${club.dlblog.socketio.port}") private Integer port; /**
     * netty-socketio服务器
     * @return
     **/ @Bean public SocketIOServer socketIOServer() { com.corundumstudio.socketio.Configuration config = new com.corundumstudio.socketio.Configuration(); //设置host config.setHostname(host); //设置端口 config.setPort(port); SocketIOServer server = new SocketIOServer(config); //启动socket服务 server.start(); return server; } /**
     *用于扫描netty-socketio的注解，比如 @OnConnect、@OnEvent
     *
     **/ @Bean public SpringAnnotationScanner springAnnotationScanner() { return new SpringAnnotationScanner(socketIOServer()); } /**
     * 注入socket处理拦截器
     * @return
     */ @Bean public NettySocketHandler nettySocketHandler(){ return new NettySocketHandler(); } }

认证监听

认证用服务定义

/**
 * 默认认证服务实现
 * @author machenike
 */ public class DefaultSocketAuthServiceImpl implements SocketAuthService { /**
     * 日志
     */ private final static Logger logger = LoggerFactory.getLogger(DefaultSocketAuthServiceImpl.class); private final static String QUERY_CLIENT_ID = "clientId"; static Map<String, SocketIOClient> socketMap; static{ socketMap = NettySocketHandler.clientMap; } @Override public boolean auth(HandshakeData handshakeData) { String clientId = handshakeData.getSingleUrlParam(QUERY_CLIENT_ID); if(clientId!=null){ //若客户端存在 if(socketMap.get(clientId)!=null){ logger.debug("current socket clientId - "+clientId+" is repeated"); //认证失败 return false; } logger.debug("socket client auth success [clientId="+clientId+"]"); return true; } logger.debug("socket client auth failed [clientId="+clientId+"]"); return false; } }

认证监听器定义，当isAuthorized方法返回false时，认证失败，连接失败返回401

/**
 * 认证监听器
 * @author machenike
 */ public class SocketAuthListener implements AuthorizationListener { private SocketAuthService authService; @Override public boolean isAuthorized(HandshakeData handshakeData) { return authService.auth(handshakeData); } public SocketAuthListener(SocketAuthService authService) { this.authService = authService; } public SocketAuthListener() { this.authService = new DefaultSocketAuthServiceImpl(); } }

注入认证监听
在上述的config类中增加监听器配置

 /**
     * netty-socketio服务器
     * @return
     **/ @Bean public SocketIOServer socketIOServer() { com.corundumstudio.socketio.Configuration config = new com.corundumstudio.socketio.Configuration(); //设置host config.setHostname(host); //设置端口 config.setPort(port); //初始化认证监听器 AuthorizationListener SocketAuthListener = new SocketAuthListener(); //设置认证监听器 config.setAuthorizationListener(SocketAuthListener); SocketIOServer server = new SocketIOServer(config); //启动socket服务 server.start(); return server; }

事件监听

当前连接成功时触发onConnect事件，同时将客户端实例保存到线程安全的ConcurrentHashMap中。

/**
 * socket处理拦截器
 * @author machenike
 */ public class NettySocketHandler implements CommandLineRunner { /**
     * 日志
     */ private final static Logger logger = LoggerFactory.getLogger(NettySocketHandler.class); /**
     * 客户端保存用Map
     */ public static Map<String, SocketIOClient> clientMap = new ConcurrentHashMap<>(); /**
     * 连接数
     */ public static AtomicInteger onlineCount = new AtomicInteger(0); private final static String QUERY_CLIENT_ID = "clientId"; /**
     * 客户端连上socket服务器时执行此事件
     * @param client
     */ @OnConnect public void onConnect(SocketIOClient client) { String clientId = client.getHandshakeData().getSingleUrlParam(QUERY_CLIENT_ID); logger.debug("onConnect: [clientId="+clientId+"]"); if(clientId!=null) { clientMap.put(clientId, client); onlineCount.addAndGet(1); logger.debug("connect success: [clientId="+clientId+",onlineCount="+onlineCount.get()+"]"); } } /**
     * 客户端断开socket服务器时执行此事件
     * @param client
     */ @OnDisconnect public void onDisconnect(SocketIOClient client) { String clientId = client.getHandshakeData().getSingleUrlParam(QUERY_CLIENT_ID); if (clientId != null) { clientMap.remove(clientId); client.disconnect(); onlineCount.addAndGet(-1); logger.debug("disconnect success: [clientId="+clientId+",onlineCount="+onlineCount.get()+"]"); } } /**
     *
     * @param client
     */ @OnEvent( value = "message") public void onMessage(SocketIOClient client, AckRequest request, Object data) { String clientId = client.getHandshakeData().getSingleUrlParam(QUERY_CLIENT_ID); logger.debug("onMessage: [clientId="+clientId+",data="+data+"]"); //request.sendAckData("message is revived"); NettySocketUtil.sendNotice("test message"); client.sendEvent("ack",1); } @Override public void run(String... args) throws Exception { logger.debug("socketHandler start-------------------------------"); } }

到此SocketIO服务端代码就结束了

测试

在线演示地址

https://www.dlblog.club/file/20200808/IGMaDsur.html

在这里插入图片描述

消息推送

因为我将存储客户端的Map设置成public static，全局都可以取得，线程安全多线程可用

/**
 * socket发送消息用工具类
 * @author machenike
 */ public class NettySocketUtil { /**
     * 保存client资源Map
     */ static Map<String, SocketIOClient> socketMap; static{ //client资源Map赋值 socketMap = NettySocketHandler.clientMap; } /**
     * 发送消息 指定客户端 指定event
     * @param clientId
     * @param event
     * @param message
     */ public static void sendMessage(String clientId,String event,Object message){ socketMap = NettySocketHandler.clientMap; socketMap.get(clientId).sendEvent(event,message); } /**
     * 发送消息 指定客户端
     * @param clientId
     * @param message
     */ public static void sendMessage(String clientId,Object message){ socketMap = NettySocketHandler.clientMap; socketMap.get(clientId).sendEvent("message",message); } /**
     * 发送消息 全部客户端
     * @param message
     */ public static void sendNotice(Object message){ Set<String> clientIdSet = socketMap.keySet(); for(String clientId:clientIdSet){ socketMap.get(clientId).sendEvent("message",message); } } /**
     * 发送消息 指定event 全部客户端
     * @param message
     */ public static void sendNotice(Object message,String event){ Set<String> clientIdSet = socketMap.keySet(); for(String clientId:clientIdSet){ socketMap.get(clientId).sendEvent(event,message); } } }