关于springcloud:SpringCloud之WebSocket配置的最佳方式

53次阅读

共计 7580 个字符，预计需要花费 19 分钟才能阅读完成。

前言

最近我的项目中有个私信性能，须要用到 websocket，于是在网上找找材料并在实践中总结了一点教训分享给大家。

问题

在操作之前我先抛一个问题：

SpringBoot 我的项目集成 webSocket, 当客户端与服务器端建设连贯的时候，发现 server 对象并未注入而是为 null。
产生起因：spring 治理的都是单例（singleton），和 websocket（多对象）相冲突。
具体解释：我的项目启动时初始化，会初始化 websocket（非用户连贯的），spring 同时会为其注入 service，该对象的 service 不是 null，被胜利注入。然而，因为 spring 默认治理的是单例，所以只会注入一次 service。当客户端与服务器端进行连贯时，服务器端又会创立一个新的 websocket 对象，这时问题呈现了：spring 治理的都是单例，不会给第二个 websocket 对象注入 service，所以导致只有是用户连贯创立的 websocket 对象，都不能再注入了。
像 controller 外面有 service，service 外面有 dao。因为 controller，service，dao 都有是单例，所以注入时不会报 null。然而 websocket 不是单例，所以应用 spring 注入一次后，前面的对象就不会再注入了，会报 NullException。

上面会讲解决办法。

操作

1、引入 websocket 依赖包

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

2、配置 websocket

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry webSocketHandlerRegistry) {
        // webSocket 通道
        // 指定处理器和门路, 如：http://www.baidu.com/service-name/websocket?uid=xxxx
        webSocketHandlerRegistry.addHandler(new WebSocketHandler(), "/websocket")
//                // 指定自定义拦截器
                .addInterceptors(new WebSocketInterceptor())
                // 容许跨域
                .setAllowedOrigins("*");
    }
}

3、增加获取 websocket 地址中的参数类

public class WebSocketInterceptor implements HandshakeInterceptor {

    /**
     * handler 解决前调用,attributes 属性最终在 WebSocketSession 里, 可能通过 webSocketSession.getAttributes().get(key 值)取得
     */
    @Override
    public boolean beforeHandshake(org.springframework.http.server.ServerHttpRequest request, ServerHttpResponse serverHttpResponse, org.springframework.web.socket.WebSocketHandler webSocketHandler, Map<String, Object> map) throws Exception {if (request instanceof ServletServerHttpRequest) {ServletServerHttpRequest serverHttpRequest = (ServletServerHttpRequest) request;
            // 获取申请门路携带的参数
            String uid = serverHttpRequest.getServletRequest().getParameter("uid");
            map.put("uid", uid);
            return true;
        } else {return false;}
    }

    @Override
    public void afterHandshake(org.springframework.http.server.ServerHttpRequest serverHttpRequest, ServerHttpResponse serverHttpResponse, org.springframework.web.socket.WebSocketHandler webSocketHandler, Exception e) {}}

4、增加解决 server 对象 @Autowired 注入为 null 的类

@Component
public class SpringContext implements ApplicationContextAware {

    /**
     * 打印日志
     */
    private Logger logger = LoggerFactory.getLogger(getClass());

    /**
     * 获取上下文对象
     */
    private static ApplicationContext applicationContext;


    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        SpringContext.applicationContext = applicationContext;
        logger.info("set applicationContext");
    }

    /**
     * 获取 applicationContext
     *
     * @return
     */
    public static ApplicationContext getApplicationContext() {return applicationContext;}

    /**
     * 通过 name 获取 bean 对象
     *
     * @param name
     * @return
     */
    public static Object getBean(String name) {return getApplicationContext().getBean(name);
    }

    /**
     * 通过 class 获取 bean 对象
     *
     * @param clazz
     * @param <T>
     * @return
     */
    public static <T> T getBean(Class<T> clazz) {return getApplicationContext().getBean(clazz);
    }

    /**
     * 通过 name，clazz  获取指定的 bean 对象
     *
     * @param name
     * @param clazz
     * @param <T>
     * @return
     */
    public static <T> T getBean(String name, Class<T> clazz) {return getApplicationContext().getBean(name, clazz);
    }

}

5、增加 websocket 接管发送音讯类

@Component
public class WebSocketHandler extends AbstractWebSocketHandler {private static Logger log = LoggerFactory.getLogger(WebSocketHandler.class);

    public AccountFeignClient getAccountFeignClient() {return SpringContext.getBean(AccountFeignClient.class);
    }

    public NotifyMailboxService getNotifyMailboxService() {return SpringContext.getBean(NotifyMailboxService.class);
    }

    public NotifyMailboxMessageService getNotifyMailboxMessageService() {return SpringContext.getBean(NotifyMailboxMessageService.class);
    }

    /**
     * 存储 sessionId 和 webSocketSession
     * 须要留神的是，webSocketSession 没有提供无参结构，不能进行序列化，也就不能通过 redis 存储
     * 在分布式系统中，要想别的方法实现 webSocketSession 共享
     */
    private static Map<String, WebSocketSession> sessionMap = new ConcurrentHashMap<>();
    private static Map<String, String> userMap = new ConcurrentHashMap<>();

    /**
     * webSocket 连贯创立后调用
     */
    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        // 获取参数
        String uid = String.valueOf(session.getAttributes().get("uid"));
        String sessionId = session.getId();
        log.info("init websocket uid={},sessionId={}", uid, sessionId);
        userMap.put(uid, sessionId);
        sessionMap.put(sessionId, session);
    }

    /**
     * 前端发送音讯到后盾
     * 接管到音讯会调用
     */
    @Override
    public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception {
        // A 用户发送前端音讯到后盾，后盾要保留 A 音讯，并且向 B 用户推送音讯
        if (message instanceof TextMessage) {log.info("message={}", message);
        } else if (message instanceof BinaryMessage) {} else if (message instanceof PongMessage) { } else {log.info("Unexpected WebSocket message type:" + message);
        }
        String uid = String.valueOf(session.getAttributes().get("uid"));
        String messages = (String) message.getPayload();
        ObjectMapper mapper = new ObjectMapper();
        HashMap<String, Object> map = mapper.readValue(messages, HashMap.class);
        String _uid = (String) map.get("uid");
//        String _dialogId = (String) map.get("dialogId");
        String _friendId = (String) map.get("friendId");
        String _message = (String) map.get("message");

        String sessionId = session.getId();
        log.info("sessionId={},uid={},_uid={},_friendId={},_message={}", sessionId, uid, _uid, _friendId, _message);

        if (!StringUtils.hasLength(sessionId) || !StringUtils.hasLength(_uid) || !StringUtils.hasLength(_friendId)) {log.info("sessionId&_uid&_friendId 不能为空");
            session.sendMessage(new TextMessage("error:sessionId&_uid&_friendId 不能为空"));
            return;
        }
        String dialogId = pushMessage(_uid, _friendId, _message);
        if (dialogId != null) {TextMessage textMessage = new TextMessage("dialogId:" + dialogId);
            // 向本人的 ws 推送音讯
            session.sendMessage(textMessage);
            String sessionIdForFriend = userMap.get(_friendId);
            log.info("sessionIdForFriend={}", sessionIdForFriend);
            if (StringUtils.hasLength(sessionIdForFriend)) {WebSocketSession friendSession = sessionMap.get(sessionIdForFriend);
                if (friendSession != null && friendSession.isOpen())
                    // 向敌人推送音讯
                    friendSession.sendMessage(textMessage);
            }
        }
    }

    /**
     * 连贯出错会调用
     */
    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) {String uid = String.valueOf(session.getAttributes().get("uid"));
        String sessionId = session.getId();
        log.info("CLOSED uid= ={},sessionId={}", uid, sessionId);
        sessionMap.remove(sessionId);
    }

    /**
     * 连贯敞开会调用
     */
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {String uid = String.valueOf(session.getAttributes().get("uid"));
        String sessionId = session.getId();
        log.info("CLOSED uid= ={},sessionId={}", uid, sessionId);
        sessionMap.remove(sessionId);
    }

    @Override
    public boolean supportsPartialMessages() {return false;}

    /**
     * 后盾发送音讯到前端
     * 封装办法发送音讯到客户端
     */
    public static void sendMessage(String uid, String dialogId) {log.info("发送音讯到:");
    }

    /**
     * @param uid
     * @param friendId
     * @param message
     * @return
     */
    private String pushMessage(String uid, String friendId, String message) {log.info("uid={},friendId={},message={}", uid, friendId, message);
        NotifyMailboxService notifyMailboxService = getNotifyMailboxService();
        NotifyMailboxMessageService notifyMailboxMessageService = getNotifyMailboxMessageService();
        try {NotifyMailbox notifyMailbox = notifyMailboxService.queryBy(uid, friendId);
            
        } catch (Exception e) {log.info("exception msg={}", e.getMessage());
            return null;
        }

    }
}

websocket 前端状态码 readyState

0        CONNECTING        连贯尚未建设
1        OPEN            WebSocket 的链接曾经建设
2        CLOSING            连贯正在敞开
3        CLOSED            连贯曾经敞开或不可用


总结

1、server 对象并未注入而是为 null，所以要通过增加下面的 SpringContext 类，并通过上面这种形式援用

public AccountFeignClient getAccountFeignClient() {return SpringContext.getBean(AccountFeignClient.class);
}

public NotifyMailboxService getNotifyMailboxService() {return SpringContext.getBean(NotifyMailboxService.class);
}

public NotifyMailboxMessageService getNotifyMailboxMessageService() {return SpringContext.getBean(NotifyMailboxMessageService.class);
}

2、websocket 的连贯和敞开的 session 对应问题，应用下面代码就没问题，否则会呈现连贯上的问题。

3、WebSocketInterceptor会获取 https://www.baidu.com/service-name/websocket?uid=xxx 中的 uid 并注入到 session 中，因而 WebSocketHandler 类能力获取到 session 中的 uid 参数。

援用

WebSocket 教程
webSocket 中应用 @Autowired 注入对应为 null

正文完
springcloud
发表至： spring-cloud
2022-01-14
 0
关于spring-cloud:wanxin金融
关于springcloud:Gateway集成Netty服务
关于spring-cloud:SpringCloud升级之路20200x版24测试Spring-Cloud-LoadBalancer
关于springcloud:SpringCloud源码解析-Eureka原理探究