乐趣区

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

作者：

jiezi

在

前言

最近我的项目中有个私信性能，须要用到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

评论

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

这个站点使用 Akismet 来减少垃圾评论。了解你的评论数据如何被处理。

更多文章