栏目分类:
子分类:
返回
名师互学网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
名师互学网 > IT > 软件开发 > 后端开发 > Java

SpringBoot整合WebSocket,实现后台向前端推送信息

Java 更新时间: 发布时间: IT归档 最新发布 模块sitemap 名妆网 法律咨询 聚返吧 英语巴士网 伯小乐 网商动力

SpringBoot整合WebSocket,实现后台向前端推送信息

maven依赖:
        
            org.springframework.boot
            spring-boot-starter-websocket
        
WebSocketConfig:
@Component
public class WebSocketConfig {
    @Bean
    public ServerEndpointExporter serverEndpointExporter(){
        return new ServerEndpointExporter();
    }
}
WebSocketServer:
  1. 因为WebSocket是类似客户端服务端的形式(采用ws协议),那么这里的WebSocketServer其实就相当于一个ws协议的Controller ;
  2. 直接@ServerEndpoint("/imserver/{userId}") 、@Component启用即可,然后在里面实现@OnOpen开启连接,@onClose关闭连接,@onMessage接收消息等方法。
  3. 新建一个ConcurrentHashMap webSocketMap 用于接收当前userId的WebSocket,方便IM之间对userId进行推送消息。单机版实现到这里就可以。
  4. 集群版 需要借助MQ或者redis发布订阅,把消息发送到第三方;多台服务器同时消费该条消息,哪台服务器上有这个WebSocket那台服务器就推送消息(因为webSocket中session是接口无法序列化)
  5. 前端url ws://172.16.15.44:8080/webSocket/参数 注意 https 需要把 ws 改为 wss
这里采用的是redis发布订阅模式

redis配置请参考 :

https://blog.csdn.net/weixin_46841515/article/details/121190753

@Component
@ServerEndpoint("/webSocket/{userno}")
@Slf4j
public class WebSocketService extends MessageListenerAdapter {
    private Session session;
    @Autowired
    private StringRedisTemplate stringRedisTemplate = SpringUtils.getBean(StringRedisTemplate.class);
    private static ConcurrentHashMap> concurrentHashMap = new ConcurrentHashMap<>();

    @OnOpen
    public void  onOpen (@PathParam(value = "userno") String param, Session session){
        this.session = session;
        ConcurrentHashMap sessionMap = (null==concurrentHashMap.get(param))? new ConcurrentHashMap<>():concurrentHashMap.get(param);
        sessionMap.put(session.getId(),session);
        concurrentHashMap.put(param,sessionMap);
    }

    @OnClose
    public void  onClose (Session session){
        concurrentHashMap.remove(session.getId());
    }

    @OnMessage
    public void onMessage(String message,Session session){
        try {
            //前端发送的心跳数据 记录到redis中 方便删除已断开连接webSocket
            String redisKey = "webScoket_heartbeat_"+session.getId();
            stringRedisTemplate.opsForValue().set(redisKey,String.valueOf(new Date().getTime()),10, TimeUnit.SECONDS);
            String[] split = message.split("_");
            if (0 userAnswerMap,Integer optionId,Integer examId){
        try {
            List collect =  ExamAnswerUtil.getExamAnswerList(userAnswerMap);
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("optionId",optionId);
            jsonObject.put("list",collect);
            String message = jsonObject.toString();
            JSONObject messageJson = new JSONObject();
            messageJson.put("examId",examId.toString());
            messageJson.put("message",message);
            //发送消息到redis webScoket_message频道中(单机版可直接webSocket发送异步消息)
            stringRedisTemplate.convertAndSend("webScoket_message",messageJson.toJSONString());
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    
    @Override
    public void onMessage(Message msgs, byte[] pattern) {
        try {
            //消息内容
            byte[] body=msgs.getBody();
            //订阅房间
            String topic=new String(pattern);
            //获取存在的房间中的用户
            String result= new String(body,"utf-8");
            JSONObject js= JSON.parseObject(result);
            String examId=js.getString("examId");
            String msg=js.getString("message");
            //发送异步消息给大屏
            ConcurrentHashMap map = concurrentHashMap.get(examId);
            if (null != map){
                Iterator> iterator = map.entrySet().iterator();
                while (iterator.hasNext()){
                    Session value = iterator.next().getValue();
                    boolean open = value.isOpen();
                    if (!open){
                        map.remove(value.getId());
                        continue;
                    }
                    if (null == value) continue;
                    String redisKey = "webScoket_heartbeat_"+value.getId();
                    String time = stringRedisTemplate.opsForValue().get(redisKey);
                    if (StringUtil.isBlank(time)) {
                        map.remove(value.getId());
                        continue;
                    }
                    long longTime = Long.parseLong(time);
                    long dateTime = new Date().getTime();
                    long a = dateTime - longTime;
                    if ((dateTime - longTime)>100000){
                        map.remove(value.getId());//删除已断开的连接
                    } else {
                        try {
                            //webSocket发送异步消息
                            value.getAsyncRemote().sendText(msg);
                        }catch (Exception e){
                            continue;
                        }
                    }
                }
            }
        }catch (Exception e){
            log.info("onMessage exception");
            e.printStackTrace();
        }
    }
}
转载请注明:文章转载自 www.mshxw.com
本文地址:https://www.mshxw.com/it/439620.html
我们一直用心在做
关于我们 文章归档 网站地图 联系我们

版权所有 (c)2021-2022 MSHXW.COM

ICP备案号:晋ICP备2021003244-6号