SpringBoot项目整合WebSocket

这一个项目开始前需要用到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 的事件进行处理，这里可以同原生注解的那几个注解连起来看

  1. afterConnectionEstablished 方法是在 socket 连接成功后被触发，同原生注解里的 @OnOpen 功能
  2. afterConnectionClosed方法是在 socket 连接关闭后被触发，同原生注解里的 @OnClose 功能
  3. 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";
    }
}
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 206,839评论 6赞 482
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 88,543评论 2赞 382
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 153,116评论 0赞 344
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 55,371评论 1赞 279
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 64,384评论 5赞 374
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 49,111评论 1赞 285
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,416评论 3赞 400
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 37,053评论 0赞 259
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 43,558评论 1赞 300
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,007评论 2赞 325
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,117评论 1赞 334
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,756评论 4赞 324
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,324评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,315评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,539评论 1赞 262
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 45,578评论 2赞 355
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,877评论 2赞 345