基于springboot实现的websoket服务

package top.itjie.java.demo.websoket.server; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import top.itjie.java.demo.websoket.dao.UserDao; import top.itjie.java.demo.websoket.pojo.*; import top.itjie.java.demo.websoket.utils.JsonUtil; import javax.websocket.Session; import java.io.IOException; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @Service("wsService") public class WsService { @Autowired private UserDao userDao; private Logger logger = LoggerFactory.getLogger(this.getClass()); private final ConcurrentHashMap<Session, UserStatus> sessionMap = new ConcurrentHashMap<>(); public ConcurrentHashMap getSessionMap(){ return sessionMap; } /** * 处理消息 */ public void handleMessage(Session session, String message) throws IOException { logger.info("进入处理消息"); Map map = JsonUtil.string2Obj(message, Map.class); int type = map == null? 0 : (map.get("type") == null? 0 : (int) map.get("type")); // type = 1, 处理登录 if(type == 1){ LoginReq loginReq = JsonUtil.string2Obj(message, LoginReq.class); handleLogin(session, loginReq); } } /** * 设置SessionMap * @param session 会话 * @param userStatus 用户状态 */ public void putSession(Session session, UserStatus userStatus){ sessionMap.put(session, userStatus); logger.info("现在共有: {} 个活跃连接.", sessionMap.size()); } /** * 移除会话 * @param session 会话 */ public void removeSession(Session session){ sessionMap.remove(session); } /** * 定时任务, 给登录成功的用户发随机数 * @throws IOException */ @Scheduled(fixedRate = 2000) public void scheduleTask() throws IOException { logger.info("进入发送随机数"); logger.info("[连接检测] 当前活跃用户数: {}", sessionMap.size()); for (Session session : sessionMap.keySet()){ Heartbeat heartbeat = new Heartbeat(2, "心跳"); session.getBasicRemote().sendText(JsonUtil.obj2String(heartbeat)); logger.info("[发送心跳] userId:{}", sessionMap.get(session).getUserId()); // 给登录成功的用户发随机数 if(sessionMap.get(session).getStatus() == 1){ // 生成一个随机数保留两位小数 double random = (double) Math.round(Math.random() * 100 * 100) / 100; DataResp dataResp = new DataResp(3, random, System.currentTimeMillis()); session.getBasicRemote().sendText(JsonUtil.obj2String(dataResp)); logger.info("[发送消息] userId:{}, message: {}", sessionMap.get(session).getUserId(), dataResp); } } } void handleLogin(Session session, LoginReq loginReq) throws IOException { // 用户不存在 if(!userDao.checkUserId(loginReq.getUserId())){ LoginResp loginResp = new LoginResp(1, 0, "用户不存在!"); session.getBasicRemote().sendText(JsonUtil.obj2String(loginResp)); logger.info("[用户登录] loginReq: {}, 登录失败, 用户不存在!", loginReq); return; } // 密码错误 if(!userDao.passwordValid(loginReq.getUserId(), loginReq.getPassword())) { LoginResp loginResp = new LoginResp(1, 0, "密码不正确!"); session.getBasicRemote().sendText(JsonUtil.obj2String(loginResp)); logger.info("[用户登录] loginReq: {}, 登录失败, 密码不正确!", loginReq); return; } LoginResp loginResp = new LoginResp(1, 1, "登录成功!"); session.getBasicRemote().sendText(JsonUtil.obj2String(loginResp)); // 登录成功更新该会话状态 UserStatus userStatus = sessionMap.get(session); userStatus.setUserId(loginReq.getUserId()); userStatus.setStatus(1); logger.info("[用户登录] loginReq: {}, 登录成功!", loginReq); } }