package com.project.scoket; import cn.hutool.core.convert.Convert; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.project.common.core.domain.entity.SysUser; import com.project.common.exception.base.BaseException; import com.project.common.utils.JsonUtils; import com.project.common.utils.StringUtils; import com.project.common.utils.spring.SpringUtils; import com.project.system.mapper.SysUserMapper; import com.project.util.CurPoolUtil; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import javax.websocket.*; import javax.websocket.server.PathParam; import javax.websocket.server.ServerEndpoint; import java.util.Date; import java.util.Map; import static com.project.util.CurPoolUtil.webSockets; @Slf4j @Component @ServerEndpoint("/websocket/{userId}/{toUserId}") public class WebSocket { private Session session; /** * 连接建立成功调用的方法 */ @OnOpen public void onOpen(Session session, @PathParam(value = "userId") String userId, @PathParam(value = "toUserId") String toUserId) { this.session = session; webSockets.put(Long.parseLong(userId),this); CurPoolUtil.sessionPool.put(Long.parseLong(userId) , session); log.info("【websocket消息】有新的连接,userId:"+userId+", toUserId:"+toUserId+", 总数为:"+ webSockets.size()); System.out.println("【websocket消息】有新的连接,userId:"+userId+", toUserId:"+toUserId+", 总数为:"+ webSockets.size()); } /** * 连接关闭调用的方法 */ @OnClose public void onClose() { // 断开连接删除用户删除session Long userId = Long.parseLong(this.session.getRequestParameterMap().get("userId").get(0)); CurPoolUtil.sessionPool.remove(userId); webSockets.remove(userId); SysUserMapper userMapper = SpringUtils.getBean(SysUserMapper.class); SysUser sysUser = userMapper.selectUserById(userId); CurPoolUtil.curUserPool.remove(sysUser.getUserName()); System.out.println("【websocket消息】连接断开,usserId:"+userId+",总数为:"+ webSockets.size()); } /** * 收到客户端消息后调用的方法 * 后台收到客户端发送过来的消息 * onMessage 是一个消息的中转站 * @param message 客户端发送过来的消息 */ @OnMessage public void onMessage(String message,@PathParam(value = "userId") String userId, @PathParam(value = "toUserId") String toUserId) { if ("0".equals(toUserId)){ log.info("toUserId------------>"+toUserId+",不能发送消息"); throw new BaseException("不能发送消息"); } } @OnError public void onError(Session session, Throwable error) { log.error("发生错误"); error.printStackTrace(); } // 广播消息 public void sendAllMessage(String message) { Map sockets = webSockets; if (StringUtils.isNotEmpty(sockets)){ for (Long aLong : sockets.keySet()) { try { WebSocket webSocket = webSockets.get(aLong); if (webSocket.session!=null){ webSocket.session.getAsyncRemote().sendText(message); } } catch (Exception e) { e.printStackTrace(); } } } } // 单点消息 (发送文本) public void sendTextMessage(Long userId, String message) { Session session = (Session) CurPoolUtil.sessionPool.get(userId); if (session != null) { try { session.getBasicRemote().sendText(message); } catch (Exception e) { e.printStackTrace(); } } System.out.println("【websocket发送消息】接收者:"+userId+",消息:"+message); System.out.println("【session】session:"+session); log.info("【websocket发送消息】接收者:"+userId+",消息:"+message); } }