SpringBoot项目整合WebSocket

作者: lannisiter | 来源:发表于2020-11-01 19:25 被阅读0次

这一个项目开始前需要用到WebSocket来进行双向通信，在网上看了两天的资料，感觉还是挺乱的，整合的方式有好几种，每一种方式各有不同，我在这做一个讲解，大多数web项目只涉及到http通信，仅仅是请求-响应的模式，初学者接触全双工通信可能会不太清楚服务器端是如何主动给客户端发送消息的，其实这里很简单，服务器端只需要保存住与客户端建立的连接session即可与客户端进行通信。用代码来描述就是客户端主动建立连接之后，服务器只需要用一个map来保存这个session就可以了，我在下面的代码中会给出一个通用的管理session的类。

一、WebSocket原生注解

第一种是使用WebSocket原生的注解来开发，这种比较简单。

pom.xml

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

配置类WebSocketConfig，这里开启了配置之后springboot才会去扫描对应的注解。

@Configuration
@EnableWebSocket
public class WebSocketConfig {

    @Bean
    public ServerEndpointExporter serverEndpoint() {
        return new ServerEndpointExporter();
    }
}

处理消息类WsServerEndpoint

@ServerEndpoint("/myWs")
@Component
public class WsServerEndpoint {
    /**
     * 连接成功
     * @param session
     */
    @OnOpen
    public void onOpen(Session session) {
        System.out.println("连接成功");
    }

    /**
     * 连接关闭
     * @param session
     */
    @OnClose
    public void onClose(Session session) {
        System.out.println("连接关闭");
    }

    /**
     * 接收到消息
     * @param text
     */
    @OnMessage
    public String onMsg(String text) throws IOException {
        return "servet 发送：" + text;
    }
}

这些注解都是属于jdk自带的，并不是spring提供的，具体位置是在javax.websocket下，需要注意的是接收参数中的session，这是我们需要保存的，后面如果要对客户端发送消息的话使用session.getBasicRemote().sendText(XXX)

@ServerEndpoint

通过这个 spring boot 就可以知道你暴露出去的 ws 应用的路径，有点类似我们经常用的@RequestMapping。比如你的启动端口是8080，而这个注解的值是ws，那我们就可以通过 ws://127.0.0.1:8080/ws 来连接你的应用。

@OnOpen

当 websocket 建立连接成功后会触发这个注解修饰的方法。

@OnClose

当 websocket 建立的连接断开后会触发这个注解修饰的方法。

@OnMessage

当客户端发送消息到服务端时，会触发这个注解修改的方法，它有一个 String 入参表明客户端传入的值。

@OnError

当 websocket 建立连接时出现异常会触发这个注解修饰的方法。

二、Spring封装的WebSocket

spring同样也为我们提供了WebSocket的封装，这种方式可以自己配置拦截器，在tcp握手之前对请求进行一次处理，可以避免一些恶意的连接。

pom.xml

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

配置类WebSocketConfig，通过实现 WebSocketConfigurer 类并覆盖相应的方法进行 websocket 的配置。我们主要覆盖 registerWebSocketHandlers 这个方法。通过向 WebSocketHandlerRegistry 设置不同参数来进行配置。其中 addHandler方法添加我们的 ws 的 handler 处理类，第二个参数是你暴露出的 ws 路径。addInterceptors添加我们写的拦截器。setAllowedOrigins这个是关闭跨域校验。

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {

    @Autowired
    private HttpAuthHandler httpAuthHandler;
    @Autowired
    private MyInterceptor myInterceptor;

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry
                .addHandler(httpAuthHandler, "myWS")
                .addInterceptors(myInterceptor)
                .setAllowedOrigins("*");
    }
}

接下来实现处理器和拦截器

HttpAuthHandler用于处理ws的消息，通过继承 TextWebSocketHandler 类并覆盖相应方法，可以对 websocket 的事件进行处理，这里可以同原生注解的那几个注解连起来看

afterConnectionEstablished 方法是在 socket 连接成功后被触发，同原生注解里的 @OnOpen 功能
afterConnectionClosed方法是在 socket 连接关闭后被触发，同原生注解里的 @OnClose 功能
handleTextMessage方法是在客户端发送信息时触发，同原生注解里的 @OnMessage 功能

@Component
public class HttpAuthHandler extends TextWebSocketHandler {
    /**
     * socket 建立成功事件
     * @param session
     * @throws Exception
     */
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        Object token = session.getAttributes().get("token");
        if (token != null) {
            // 用户连接成功，放入在线用户缓存
            WsSessionManager.add(token.toString(), session);
        } else {
            throw new RuntimeException("用户登录已经失效!");
        }
    }
    /**
     * 接收消息事件
     * @param session
     * @param message
     * @throws Exception
     */
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        // 获得客户端传来的消息
        String payload = message.getPayload();
        Object token = session.getAttributes().get("token");
        System.out.println("server 接收到 " + token + " 发送的 " + payload);
        session.sendMessage(new TextMessage("server 发送给 " + token + " 消息 " + payload + " " +    LocalDateTime.now().toString()));
    }

    /**
     * socket 断开连接时
     * @param session
     * @param status
     * @throws Exception
     */
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        Object token = session.getAttributes().get("token");
        if (token != null) {
            // 用户退出，移除缓存
            WsSessionManager.remove(token.toString());
        }
    }
}

MyInterceptor用来拦截ws请求，通过实现 HandshakeInterceptor 接口来定义握手拦截器，注意这里与上面 Handler 的事件是不同的，这里是建立握手时的事件，分为握手前与握手后，而 Handler 的事件是在握手成功后的基础上建立 socket 的连接。所以在如果把认证放在这个步骤相对来说最节省服务器资源。它主要有两个方法 beforeHandshake 与 afterHandshake，顾名思义一个在握手前触发，一个在握手后触发。

@Component
public class MyInterceptor implements HandshakeInterceptor {
    /**
     * 握手前
     * @param request
     * @param response
     * @param wsHandler
     * @param attributes
     * @return
     * @throws Exception
     */
    @Override
    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object> attributes) throws Exception {
        System.out.println("握手开始");
        // 获得请求参数
        HashMap<String, String> paramMap = HttpUtil.decodeParamMap(request.getURI().getQuery(), "utf-8");
        String uid = paramMap.get("token");
        if (StrUtil.isNotBlank(uid)) {
            // 放入属性域
            attributes.put("token", uid);
            System.out.println("用户 token " + uid + " 握手成功！");
            return true;
        }
        System.out.println("用户登录已失效");
        return false;
    }
    /**
     * 握手后
     * @param request
     * @param response
     * @param wsHandler
     * @param exception
     */
    @Override
    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception exception) {
        System.out.println("握手完成");
    }
}

WsSessionManager，这里简单通过 ConcurrentHashMap来实现了一个 session 池，用来保存已经登录的WebSocket 的 session。前文提过，服务端发送消息给客户端必须要通过这个 session。

@Slf4j
public class WsSessionManager {
    /**
     * 保存连接 session 的地方
     */
    private static ConcurrentHashMap<String, WebSocketSession> SESSION_POOL = new ConcurrentHashMap<>();

    /**
     * 添加 session
     * @param key
     */
    public static void add(String key, WebSocketSession session) {
        // 添加 session
        SESSION_POOL.put(key, session);
    }

    /**
     * 删除 session,会返回删除的 session
     * @param key
     * @return
     */
    public static WebSocketSession remove(String key) {
        // 删除 session
        return SESSION_POOL.remove(key);
    }

    /**
     * 删除并同步关闭连接
     * @param key
     */
    public static void removeAndClose(String key) {
        WebSocketSession session = remove(key);
        if (session != null) {
            try {
                // 关闭连接
                session.close();
            } catch (IOException e) {
                // todo: 关闭出现异常处理
                e.printStackTrace();
            }
        }
    }
    /**
     * 获得 session
     * @param key
     * @return
     */
    public static WebSocketSession get(String key) {
        // 获得 session
        return SESSION_POOL.get(key);
    }
}

三、STOMP

stomp是WebSocket的一个子协议，SpringBoot官方也有整合stomp的例子，这也是我现在所用到的整合方式，这种方式功能更加强大，可以使用消息代理，对于发送的消息可以使用类似springMvc的处理方式，同时消息的发送变成了订阅的模式，可以很方便的进行群发。

pom.xml

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

配置类WebSocketConfig，这里注意看我的注释，我结合每一步的代码来进行讲解。

/**
 *
 * @author lannisiter
 * @date 2020/8/13 21:21
 **/
@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Bean
    public WebSocketInterceptor getWebSocketInterceptor() {
        return new WebSocketInterceptor();
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        // 配置客户端尝试连接地址
        registry.
                addEndpoint("/ws").     // 设置连接节点，前端请求的建立连接的地址就是 http://ip:端口/ws
                addInterceptors(getWebSocketInterceptor()).     // 设置握手拦截器
                setAllowedOrigins("*").     // 配置跨域
                withSockJS();       // 开启sockJS支持，这里可以对不支持stomp的浏览器进行兼容。
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        // 消息代理,这里配置自带的消息代理，也可以配置其它的消息代理
        // 一定要注意这里的参数，可以理解为开启通道,后面如果使用了"/XXX"来作为前缀，这里就要配置,同时这里的"/topic"是默认群发消息的前缀,前端在订阅群发地址的时候需要加上"/topic"
        registry.enableSimpleBroker("/user","/topic");  
        // 客户端向服务端发送消息需有的前缀,需要什么样的前缀在这里配置,但是不建议使用，这里跟下面首次订阅返回会有点冲突，如果不需要首次订阅返回消息，也可以加上消息前缀
        // registry.setApplicationDestinationPrefixes("/");
        // 配置单发消息的前缀 /user，前端订阅一对一通信的地址需要加上"/user"前缀
        registry.setUserDestinationPrefix("/user");
    }
}

WebSocketInterceptor握手拦截器，用来过滤请求，我这里是要求用户先登录才能建立ws连接，所以我拿到请求头参数中的token去验证。

/**
 * @author lanni
 * @date 2020/9/10 17:15
 * WebSocket拦截器
 **/
@Slf4j
public class WebSocketInterceptor implements HandshakeInterceptor {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public boolean beforeHandshake(ServerHttpRequest serverHttpRequest, ServerHttpResponse serverHttpResponse, WebSocketHandler webSocketHandler, Map<String, Object> map) throws Exception {
        log.info("---------握手开始---------");
        log.info("访问路径:"+serverHttpRequest.getURI());
        boolean flag = false;
        // 获得请求参数
        HashMap<String, String> paramMap = (HashMap<String, String>) HttpUtil.decodeParamMap(serverHttpRequest.getURI().getQuery(), CharsetUtil.CHARSET_UTF_8);
        String token = paramMap.get("token");
        if(token != null && !"".equals(token)) {
            //验证token
            String adminId = stringRedisTemplate.opsForValue().get(token);
            if (StrUtil.isNotBlank(adminId)) {
                String realToken = stringRedisTemplate.opsForValue().get(adminId);
                flag =  token.equals(realToken);
            }
        }
        serverHttpResponse.getHeaders().setContentType(new MediaType("application","json", Charset.forName("UTF-8")));
        if (!flag) {
            log.info("用户未登录");
            serverHttpResponse.getBody().write(JSONObject.toJSONBytes(ResultWrapper.wrap(Result.ERROR_UNLOGIN_CODE,Result.ERROR_UNLOGIN_MESSAGE)));
        }
        return flag;
    }

    @Override
    public void afterHandshake(ServerHttpRequest serverHttpRequest, ServerHttpResponse serverHttpResponse, WebSocketHandler webSocketHandler, Exception e) {
        log.info("握手成功...");
        try {
            serverHttpResponse.getBody().write(JSONObject.toJSONBytes(ResultWrapper.success("WebSocket连接成功!")));
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }
}

WSController是ws的控制器，@SubscribeMapping注解可以在客户端首次订阅了对应的地址后直接返回一条消息，订阅地址支持路径参数，接收路径参数需要在参数前加上@DestinationVariable，下面有三种常用的订阅方式，这里一定要注意地址格式，通用群发消息/topic/hello，指定一部分人可以收到的群发消息/topic/state/{classId}，一对一消息/user/{name}/hello，我这里的ResultWrapper.success就是一个封装类，跟springMVC中封装的返回对象完全一致，stomp会把对象解析为json字符串返回给前端。

@MessageMapping是用来接收客户端对某个地址发送的消息，需要注意的是客户端发送的地址，如果在之前的配置类中配置了发送前缀，则必须携带前缀才能发送消息到客户端，如：/app/hello，但是服务器仍然只需要这样写@MessageMapping("/hello")。

@SendTo是用来向客户端发送消息的注解，这里填写的参数就是订阅地址的全名/topic/hello不能省略/topic，返回消息只需要return消息对象即可。

除了注解的方式发送消息，还有一种灵活的方式使用消息模板来发送，simpMessagingTemplate.convertAndSendToUser(一对一)和simpMessagingTemplate.convertAndSend(群发)，注意我参数中填写的方式，这种方式比较推荐使用，可以在任意地方对客户端发送消息，但是这个地方似乎有个坑，发送消息之后会阻塞在这里，不过可以开一个线程去发送消息。

/**
 *
 * @author lannisiter
 * @date 2020/8/13 21:25
 **/
@RestController
public class WSController {
    @Autowired
    private SimpMessagingTemplate simpMessagingTemplate;

    @SubscribeMapping({"/topic/hello"})
    public Result subscribeTime() {
        return ResultWrapper.success("hello!");
    }
    
    @SubscribeMapping({"/topic/info/{classId}"})
    public Result subscribeState(@DestinationVariable String classId) {
        return ResultWrapper.success("班级消息推送订阅成功!");
    }

    @SubscribeMapping({"/user/{name}/hello"})
    public Result subscribeParam(@DestinationVariable String name) {
        return ResultWrapper.success("你好!"+name);
    }
    
    @MessageMapping("/hello")
    @SendTo("/topic/hello")
    public Result hello(RequestMessage requestMessage) {
        System.out.println("接收消息：" + requestMessage);
        return ResultWrapper.success("服务端接收到你发的："+requestMessage);
    }

    @GetMapping("/sendMsgToUser")
    public String sendMsgByUser(String name, String msg) {
        // /user/{name}/hello
        simpMessagingTemplate.convertAndSendToUser(name, "/hello", msg);
        return "success";
    }

    @GetMapping("/sendMsgToAll")
    public String sendMsgByAll(int classId, String msg) {
        // /topic/info/{classId}
        simpMessagingTemplate.convertAndSend("/topic/info/"+classId, msg);
        return "success";
    }
}

SpringBoot项目整合WebSocket

一、WebSocket原生注解

二、Spring封装的WebSocket

三、STOMP

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读