springboot整合websocket(Stomp协议)

过滤器

public class NecpWebSocketInterceptor implements HandshakeInterceptor {
    Logger logger = LoggerFactory.getLogger(NecpWebSocketInterceptor.class);

    @Override
    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler handle, Map<String, Object> map) throws Exception {
        logger.info("啦啦啦，握手开始啦");
        return true;
    }

    @Override
    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler handle, Exception e) {
        logger.info("啦啦啦，握手成功啦");
    }
}
           

配置

@EnableWebSocketMessageBroker
@Configuration
public class WebSocket implements WebSocketMessageBrokerConfigurer {

    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {

//        long[] hert = {1000L,1000L};
        //这里使用的是内存模式，生产环境可以使用rabbitmq或者其他mq。
        //registry.enableStompBrokerRelay().setRelayHost().setRelayPort() 其他方式。

        // 广播式配置名为 /topic 消息代理 , 这个消息代理必须和 controller 中的 @SendTo 配置的地址前缀一样或者全匹配
        //表示消息得订阅
        config.enableSimpleBroker("/topic","/user","/queue");
//        .setHeartbeatValue(hert) //心跳设置
//        ;

        //一对一发送前缀
        config.setUserDestinationPrefix("/user");

        //后台应用接收浏览器消息端点前缀，这个将直接路由到@MessageMapping上
        config.setApplicationDestinationPrefixes("/ws");
    }

    /**
     * 注册stomp端点，主要是起到连接作用
     * @param registry registry
     */
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {

        端点名称
        registry.addEndpoint("/necp_ws")
                .addInterceptors(new  NecpWebSocketInterceptor())
                //跨域
                .setAllowedOrigins("*")
//                .setHandshakeHandler()

                //使用sockJS
                .withSockJS();
    }


}
           

使用

/**
 * Demo
 * @author lgs
 */
@Controller
public class WsTestController {

    Logger logger = LoggerFactory.getLogger(WsTestController.class);

    @Autowired
    private SimpMessagingTemplate simpMessagingTemplate;


    /**
     *  广播订阅
     * @param name
     * @return
     * @throws InterruptedException
     */
    @MessageMapping("/topic1")
    @SendTo("/topic/getResponse")
    public String sub(String name) throws InterruptedException {
        logger.info("广播注解方式订阅执行了");
        Thread.sleep(1000);
        return "广播注解方式订阅成功了";
    }

    /**
     *  广播订阅
     * @param name
     * @return
     * @throws InterruptedException
     */
    @MessageMapping("/topic2")
    public void sub2(String name) throws InterruptedException {
        logger.info("广播模板方式订阅执行了");
        Thread.sleep(1000);
        simpMessagingTemplate.convertAndSend("/topic/getResponse","广播模板方式订阅成功了");
    }


    /**
     * 单点发送
     * @return
     * @throws Exception
     */
    @MessageMapping("/welcome1")
    @SendToUser//默认发送到/user/queue/welcome1
    public String welcome(String name) throws Exception {

        logger.info(name + "单点订阅注解方式执行了");
        Thread.sleep(1000);

        return "单点订阅注解方式成功了";
    }


    /**
     * 单点发送
     * @return
     * @throws Exception
     */
    @MessageMapping("/welcome2/{userId}/{stationId}")
    public void say2(@DestinationVariable("userId") String userId , @DestinationVariable("stationId") String stationId) throws Exception {

        logger.info(userId + "单点订阅模板方式执行了" + stationId);
        Thread.sleep(1000);
        simpMessagingTemplate.convertAndSendToUser("123","/queue/getResponse","单点订阅模板方式成功了");
    }

    /**
     * 单点发送
     * @return
     * @throws Exception
     */
    @MessageMapping("/welcome2")
    public void say2(String name) throws Exception {

        logger.info(name + "单点订阅模板方式执行了");
        Thread.sleep(1000);
        simpMessagingTemplate.convertAndSendToUser("123","/queue/getResponse","单点订阅模板方式成功了");
    }

    /**
     * 定时广播
     */
    @Scheduled(fixedRate = 10000)
    public void sendTopicMessage() {
        System.out.println("后台广播推送！");
        simpMessagingTemplate.convertAndSend("/topic/getResponse","感谢您订阅了我");
    }

}