SpringBoot集成SocketIO

您所在的位置：网站首页 › 白头单怎么写 › SpringBoot集成SocketIO

SpringBoot集成SocketIO

2023-08-18 09:57| 来源: 网络整理| 查看: 265

SpringBoot集成SocketIO 简介集成SpringBoot添加SocketIO依赖修改配置文件新建SocketIO的配置文件新建SocketIoHandler处理器新建处理消息的逻辑客户端代码

简介

Socket.IO是一个完全由JavaScript实现、基于Node.js、支持WebSocket的协议用于实时通信、跨平台的开源框架，它包括了客户端的JavaScript和服务器端的Node.js。Socket.IO除了支持WebSocket通讯协议外，还支持许多种轮询（Polling）机制以及其它实时通信方式，并封装成了通用的接口，并且在服务端实现了这些实时机制的相应代码。Socket.IO实现的Polling通信机制包括Adobe Flash Socket、AJAX长轮询、AJAX multipart streaming、持久Iframe、JSONP轮询等。Socket.IO能够根据浏览器对通讯机制的支持情况自动地选择最佳的方式来实现网络实时应用。

集成SpringBoot 添加SocketIO依赖 !-- SocketIO--> com.corundumstudio.socketio netty-socketio 1.7.7 修改配置文件 server.port=8081 #============================================================================ # netty socket io setting #============================================================================ # host在本地测试可以设置为localhost或者本机IP，在Linux服务器跑可换成服务器IP socketio.host=localhost socketio.port=9099 # 设置最大每帧处理数据的长度，防止他人利用大数据来攻击服务器 socketio.maxFramePayloadLength=1048576 # 设置http交互最大内容长度 socketio.maxHttpContentLength=1048576 # socket连接数大小（如只监听一个端口boss线程组为1即可） socketio.bossCount=1 socketio.workCount=100 socketio.allowCustomRequests=true # 协议升级超时时间（毫秒），默认10秒。HTTP握手升级为ws协议超时时间 socketio.upgradeTimeout=1000000 # Ping消息超时时间（毫秒），默认60秒，这个时间间隔内没有接收到心跳消息就会发送超时事件 socketio.pingTimeout=6000000 # Ping消息间隔（毫秒），默认25秒。客户端向服务器发送一条心跳消息间隔 socketio.pingInterval=25000 新建SocketIO的配置文件 package com.zh.config; import com.corundumstudio.socketio.SocketConfig; import com.corundumstudio.socketio.SocketIOServer; import com.corundumstudio.socketio.annotation.SpringAnnotationScanner; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /** * @Description: * @ClassName SocketIOConfig * @date: 2021.09.08 16:14 * @Author: zhanghang */ @Configuration public class SocketIOConfig { @Value("${socketio.host}") private String host; @Value("${socketio.port}") private Integer port; @Value("${socketio.bossCount}") private int bossCount; @Value("${socketio.workCount}") private int workCount; @Value("${socketio.allowCustomRequests}") private boolean allowCustomRequests; @Value("${socketio.upgradeTimeout}") private int upgradeTimeout; @Value("${socketio.pingTimeout}") private int pingTimeout; @Value("${socketio.pingInterval}") private int pingInterval; /** * 以下配置在上面的application.properties中已经注明 * @return */ @Bean public SocketIOServer socketIOServer() { SocketConfig socketConfig = new SocketConfig(); socketConfig.setTcpNoDelay(true); socketConfig.setSoLinger(0); com.corundumstudio.socketio.Configuration config = new com.corundumstudio.socketio.Configuration(); config.setSocketConfig(socketConfig); config.setHostname(host); config.setPort(port); config.setBossThreads(bossCount); config.setWorkerThreads(workCount); config.setAllowCustomRequests(allowCustomRequests); config.setUpgradeTimeout(upgradeTimeout); config.setPingTimeout(pingTimeout); config.setPingInterval(pingInterval); return new SocketIOServer(config); } /** * 用于扫描netty-socketio的注解，比如 @OnConnect、@OnEvent */ @Bean public SpringAnnotationScanner springAnnotationScanner() { return new SpringAnnotationScanner(socketIOServer()); } }

注意：如果想要SocketIO 的注解生效，必须注入SpringAnnotationScanner 这个类。

新建SocketIoHandler处理器 package com.zh.socket.handler; import com.corundumstudio.socketio.SocketIOClient; import com.corundumstudio.socketio.SocketIOServer; import com.corundumstudio.socketio.annotation.OnConnect; import com.corundumstudio.socketio.annotation.OnDisconnect; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import java.util.HashMap; import java.util.Map; import java.util.Observable; /** * @Description: * @ClassName SocketIOMessageEventHandler * @date: 2021.09.09 09:06 * @Author: zhanghang */ @Component @Slf4j public class SocketIOMessageEventHandler extends Observable { @Autowired private SocketIOServer socketIoServer; /** * Spring IoC容器创建之后，在加载SocketIOServiceImpl Bean之后启动 * * @throws Exception */ @PostConstruct private void autoStartup() throws Exception { try { socketIoServer.start(); }catch (Exception ex){ ex.printStackTrace(); log.error("SocketIOServer启动失败"); } } /** * Spring IoC容器在销毁SocketIOServiceImpl Bean之前关闭,避免重启项目服务端口占用问题 * * @throws Exception */ @PreDestroy private void autoStop() throws Exception { socketIoServer.stop(); } /** * 客户端连接的时候触发 * * @param client */ @OnConnect public void onConnect(SocketIOClient client) { String username = client.getHandshakeData().getSingleUrlParam("username"); // 这里可以传入token验证 // 提醒观察者 log.info("客户端:" + client.getRemoteAddress() + " sessionId:" + client.getSessionId() +" username: "+ username+ "已连接"); } /** * 客户端关闭连接时触发 * * @param client */ @OnDisconnect public void onDisconnect(SocketIOClient client) { log.info("客户端:" + client.getSessionId() + "断开连接"); Map paramMap = new HashMap(); paramMap.put("type", "disconnect"); paramMap.put("sessionId", client.getSessionId().toString()); // 提醒观察者 this.setChanged(); this.notifyObservers(paramMap); } }

说明：@OnDisconnect，@OnConnect，@OnEvent都属于SocketIO的注解，想要注解生效，则必须在配置配配置SpringAnnotationScanner; @OnConnect: 监听客户端连接 @OnDisconnect：监听客户端断开连接 @OnEvent (value="text")：用于监听客户端发送的消息,value的值就是客户端请求的唯一标识，如：socket.emit('text','要发送的消息');

注意：SocketIOMessageEventHandler 继承了Observable，Observable是JDK自带的观察者模式中的类，继承这个类的类说明是被观察者，

新建处理消息的逻辑 package com.zh.service; import org.springframework.beans.factory.InitializingBean; import java.util.Map; import java.util.Observer; /** * @Description: * @ClassName TestSocketIOService * @date: 2021.09.09 14:07 * @Author: zhanghang */ public interface TestSocketIOService extends Observer, InitializingBean { /** * description: 给容器内所有的客户端发送通知 * date: 2021年-09月-09日 14:08 * author: zhanghang * * @param msg * @return void */ void sendMessageToAllUser(Map msg); /** * description: 给指定用户发送通知 * date: 2021年-09月-09日 14:09 * author: zhanghang * * @param username * @param msg * @return void */ void sendMessage(String username, Map msg); } package com.zh.service.impl; import com.corundumstudio.socketio.AckRequest; import com.corundumstudio.socketio.SocketIOClient; import com.corundumstudio.socketio.annotation.OnEvent; import com.zh.service.TestSocketIOService; import com.zh.socket.handler.SocketIOMessageEventHandler; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Observable; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; /** * @Description: * @ClassName TestSocketIOServiceImpl * @date: 2021.09.09 14:09 * @Author: zhanghang */ @Service @Slf4j public class TestSocketIOServiceImpl implements TestSocketIOService { // 使用ConcurrentMap 存储客户端 public static ConcurrentMap connectMap = new ConcurrentHashMap(); // 发送消息的通道 public final static String SEND_CHANNEL = "send_channel"; @Autowired private SocketIOMessageEventHandler socketIOMessageEventHandler; /** * description: 测试通道连接 * date: 2021年-09月-09日 14:26 * author: zhanghang * * @param client * @param request * @return void */ @OnEvent(value = "test-channel-connetListener") public void startOrderDetailChangeListener(SocketIOClient client, AckRequest request, String username) { // 数据校验 if (null == username) { return; } if (username.length() == 0){ return; } String sessionId = client.getSessionId().toString(); log.info("SocketIO-消息通知-新增连接-sessionId:" + client.getSessionId()); connectMap.put(username+"~"+sessionId, client); } /** * description: 给容器内所有的客户端发送通知 * date: 2021年-09月-09日 14:10 * author: zhanghang * * @param msg * @return void */ @Override public void sendMessageToAllUser(Map msg) { if (connectMap.isEmpty()){ return; } connectMap.entrySet().forEach(entry -> { entry.getValue().sendEvent(SEND_CHANNEL,msg); }); } /** * description: 给指定用户发送通知 * date: 2021年-09月-09日 14:10 * author: zhanghang * * @param username * @param msg * @return void */ @Override public void sendMessage(String username, Map msg) { SocketIOClient socketClient = getSocketClientByUsername(username); if ( null != socketClient){ socketClient.sendEvent(SEND_CHANNEL,msg); } } /** * description: 根据用户找到对应的客户端 * date: 2021年-09月-09日 14:33 * author: zhanghang * * @param * @return com.corundumstudio.socketio.SocketIOClient */ public SocketIOClient getSocketClientByUsername(String username){ SocketIOClient client = null; if (null == username){ return client; } if (connectMap.isEmpty()){ return client; } for (String key : connectMap.keySet()) { if (username.equals(key.split("~")[0])){ client = connectMap.get(key); } } return client; } /** * description: 观察者模式中的通知 * date: 2021年-09月-09日 14:10 * author: zhanghang * * @param o * @param arg * @return void */ @Override public void update(Observable o, Object arg) { if (!(o instanceof SocketIOMessageEventHandler)) { return; } Map map = new HashMap(); if (arg instanceof Map){ map = (Map) arg; } log.info("TestSocketIOServiceImpl{}客户端接收到通知："+map.toString()); // Map map = (Map) arg; // String type = MapUtil.getStr(map, "type"); Object type = map.get("type"); if (null == type) { return; } if (type.equals("disconnect")) { // 断开连接 this.disconnect(map); } } /** * description: 断开连接 * date: 2021年-09月-09日 14:23 * author: zhanghang * * @param map * @return void */ private void disconnect(Map map) { // String sessionId = MapUtil.getStr(map, "sessionId"); Object sessionId = map.get("sessionId"); if(null == sessionId ){ return; } List keyList = connectMap.keySet().parallelStream().filter(k->k.split("~")[1].equals(sessionId.toString())).collect(Collectors.toList()); if(null != keyList && keyList.size() > 0 ){ connectMap.remove(keyList.get(0)); } } /** * description: 注册进观察者模式 * date: 2021年-09月-09日 14:10 * author: zhanghang * * @param * @return void */ @Override public void afterPropertiesSet() throws Exception { // spring 为bean提供了两种初始化Bean的方法，1，在配置文件中指定init-metho方法。2，实现InitializingBean接口，实现afterPropertiesSet()方法 // 只要实现了 InitializingBean 接口，Spring 就会在类初始化时自动调用该afterPropertiesSet()方法 // 将当前对象注册进观察者模式中 socketIOMessageEventHandler.addObserver(this); } }

注意：这个类实现了Observer, InitializingBean 两个接口， Observer：是JDK 自带观察者模式中的观察者类， InitializingBean ：是Spring 初始化实现方式两种的其中一种，实现InitializingBean 接口，从写afterPropertiesSet()方法，spring会在初始化这个类的时候，自动先调用afterPropertiesSet()方法。另一种实现方式就是在配置类中指定init-method 方法。

@RestController @Slf4j public class TestServletController { // @Autowired // private SocketIOService socketIOService; @Autowired private TestSocketIOService testSocketIOService; @GetMapping("/myServlet") public String testMyHttpServlet(){ return "hello servlet"; } @PostMapping("/myServlet1") public String testMyHttpServlet1(){ return "hello servlet"; } @GetMapping("/sendMsg") public String sendMsg(@RequestParam("userId") String userId,@RequestParam("msg") String msg){ return "hello socketIO"; } @GetMapping("/testSendMsg") public String testSendMsg(@RequestParam("username") String username,@RequestParam("msg") String msg){ Map map = new HashMap(); map.put("msg",msg); testSocketIOService.sendMessage(username, map); return "hello socketIO"; } } 客户端代码

客户端使用html来实现

DOCTYPE html> NETTY SOCKET.IO DEMO body { padding: 20px; } #console { height: 450px; overflow: auto; } .username-msg { color: orange; } .connect-msg { color: green; } .disconnect-msg { color: red; } 发送数据 var socket; connect(); function connect() { var username = 'zhangsan'; var opts = { query: 'username=' + username }; socket = io.connect('http://localhost:9099', opts); socket.on('connect', function () { console.log("连接成功"); serverOutput('连接成功'); }); socket.on('send_channel', function (data) { let msg= JSON.stringify(data) output('' + msg + ' '); console.log(data); }); socket.on('disconnect', function () { serverOutput('' + '已下线! '); }); } function output(message) { //var element = $("" + " " + message + ""); $('#console').text(message); } function resultOutput(message) { var element = $("" + " " + message + ""); $('#console').prepend(element); } function serverOutput(message) { var element = $("" + message + ""); $('#console').prepend(element); } function send() { console.log('发送数据'); socket.emit('test-channel-connetListener','zhangsan'); }

【本文地址】

SpringBoot集成SocketIO

SpringBoot集成SocketIO

今日新闻

推荐新闻