首页 > 代码库 > Spring+Websocket实现消息的推送

Spring+Websocket实现消息的推送

2024-07-18 09:57:15   219人阅读

Websocet服务端实现

WebSocketConfig.java 

@Configuration
@EnableWebMvc
@EnableWebSocket
public class WebSocketConfig extends WebMvcConfigurerAdapter implements WebSocketConfigurer {
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(systemWebSocketHandler(),"/webSocketServer").addInterceptors(new WebSocketHandshakeInterceptor());

        registry.addHandler(systemWebSocketHandler(), "/sockjs/webSocketServer").addInterceptors(new WebSocketHandshakeInterceptor())
                .withSockJS();
    }

    @Bean
    public WebSocketHandler systemWebSocketHandler(){
        return new SystemWebSocketHandler();
    }

}
不要忘记在springmvc的配置文件中配置对此类的自动扫描 

<context:component-scan base-package="com.ldl.origami.websocket" />

@Configuration

@EnableWebMvc
@EnableWebSocket
这三个大致意思是使这个类支持以@Bean的方式加载bean，并且支持springmvc和websocket，不是很准确大致这样，试了一下@EnableWebMvc不加也没什么影响，@Configuration本来就支持springmvc的自动扫描 


registry.addHandler(systemWebSocketHandler(),"/webSocketServer").addInterceptors(new WebSocketHandshakeInterceptor())
用来注册websocket server实现类，第二个参数是访问websocket的地址 



registry.addHandler(systemWebSocketHandler(), "/sockjs/webSocketServer").addInterceptors(new WebSocketHandshakeInterceptor())
                .withSockJS();
    }
这个是使用Sockjs的注册方法


首先SystemWebSocketHandler.java 


public class SystemWebSocketHandler implements WebSocketHandler {

    private static final Logger logger;

    private static final ArrayList<WebSocketSession> users;

    static {
        users = new ArrayList<>();
        logger = LoggerFactory.getLogger(SystemWebSocketHandler.class);
    }

    @Autowired
    private WebSocketService webSocketService;

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        logger.debug("connect to the websocket success......");
        users.add(session);
        String userName = (String) session.getAttributes().get(Constants.WEBSOCKET_USERNAME);
        if(userName!= null){
            //查询未读消息
            int count = webSocketService.getUnReadNews((String) session.getAttributes().get(Constants.WEBSOCKET_USERNAME));

            session.sendMessage(new TextMessage(count + ""));
        }
    }

    @Override
    public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception {

        //sendMessageToUsers();
    }

    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        if(session.isOpen()){
            session.close();
        }
        logger.debug("websocket connection closed......");
        users.remove(session);
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception {
        logger.debug("websocket connection closed......");
        users.remove(session);
    }

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

    /**
     * 给所有在线用户发送消息
     *
     * @param message
     */
    public void sendMessageToUsers(TextMessage message) {
        for (WebSocketSession user : users) {
            try {
                if (user.isOpen()) {
                    user.sendMessage(message);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 给某个用户发送消息
     *
     * @param userName
     * @param message
     */
    public void sendMessageToUser(String userName, TextMessage message) {
        for (WebSocketSession user : users) {
            if (user.getAttributes().get(Constants.WEBSOCKET_USERNAME).equals(userName)) {
                try {
                    if (user.isOpen()) {
                        user.sendMessage(message);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
                break;
            }
        }
    }
}



相关内容大家一看就能明白，就不多解释了


然后WebSocketHandshakeInterceptor.java 


public class WebSocketHandshakeInterceptor implements HandshakeInterceptor {

    private static Logger logger = LoggerFactory.getLogger(HandshakeInterceptor.class);
    @Override
    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object
                > attributes) throws Exception {
        if (request instanceof ServletServerHttpRequest) {
            ServletServerHttpRequest servletRequest = (ServletServerHttpRequest) request;
            HttpSession session = servletRequest.getServletRequest().getSession(false);
            if (session != null) {
                //使用userName区分WebSocketHandler，以便定向发送消息
                String userName = (String) session.getAttribute(Constants.SESSION_USERNAME);
                attributes.put(Constants.WEBSOCKET_USERNAME,userName);
            }
        }
        return true;
    }

    @Override
    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception exception) {

    }
}

这个的主要作用是取得当前请求中的用户名，并且保存到当前的WebSocketHandler中，以便确定WebSocketHandler所对应的用户，具体可参考HttpSessionHandshakeInterceptor

用户登录建立websocket连接

index.jsp 

<script type="text/javascript" src=http://www.mamicode.com/"http://localhost:8080/Origami/websocket/sockjs-0.3.min.js"></script>> 

 
 
 

 使用sockjs时要注意 
 

 1、这两个的写法 
 

<script type="text/javascript" src=http://www.mamicode.com/"http://localhost:8080/Origami/websocket/sockjs-0.3.min.js"></script>
 

websocket = new SockJS("http://localhost:8080/Origami/sockjs/webSocketServer");
 2、web.xml中 

<web-app version="3.0" xmlns="http://java.sun.com/xml/ns/javaee"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_1.xsd">
 

version
 

web-app_3_1.xsd
 这两个的版本都要是3.0+ 

 
 
 

 然后在这个servlet中加入 
 

<async-supported>true</async-supported>
 

 
 
 

<servlet>
		<servlet-name>appServlet</servlet-name>
		<servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
		<init-param>
			<param-name>contextConfigLocation</param-name>
			<param-value>classpath*:servlet-context.xml</param-value>
		</init-param>
		<load-on-startup>1</load-on-startup>
        <async-supported>true</async-supported>
	</servlet>
 

 然后所有的filter中也加入 
 

<async-supported>true</async-supported>
 

 
 
 

 3、添加相关依赖 
 

<dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-annotations</artifactId>
            <version>2.3.0</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-core</artifactId>
            <version>2.3.1</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.3.3</version>
        </dependency>
 好了，现在websocket可以正常建立起来了 

 
 
 

 
 
 


 返回用户未读的消息 
 

 当连接建立后，会进入SystemWebSocketHandler的afterConnectionEstablished方法，代码看上边，取出WebSocketHandshakeInterceptor中保存的用户名 
 

 查询信息后使用session.sendMessage(new TextMessage(count + ""));返回给用户，从哪来回哪去 
 

 
  
 


 服务端推送消息给用户 
 

   
 

@Controller
public class AdminController {

    static Logger logger = LoggerFactory.getLogger(AdminController.class);

    @Autowired(required = false)
    private AdminService adminService;

    @Bean
    public SystemWebSocketHandler systemWebSocketHandler() {
        return new SystemWebSocketHandler();
    }


    @RequestMapping("/auditing")
    @ResponseBody
    public String auditing(HttpServletRequest request){
        //无关代码都省略了
        int unReadNewsCount = adminService.getUnReadNews(username);
        systemWebSocketHandler().sendMessageToUser(username, new TextMessage(unReadNewsCount + ""));
        return result;
    }
}
 

 在这里可以使用sendMessageToUser给某个用户推送信息，也可以使用sendMessageToUsers给所有用户推送信息


联系
我们