WebSocket+SpringBoot消息主动推送

通常在web前端通知客户的时候，一种方式是通过Http/Https轮询来实现，另外一种方式是通过WebSocket 这种轻量Tcp连接来实现，我们这边介绍的就是消息推送。

Http+WebSocket 原理解析

前端实现webSocket 的方式如下两种方式仅供参考：

• 第一种一种是使用sockjs。
• 第二种是使用h5的标准。使用Html5标准自然更方便简单，所以记录的是配合h5的使用方法。
  前端代码
  index.html

    
    


    Welcome

    Send    Close
    

    

后端实现

• 包的引入

        
            org.springframework.boot
            spring-boot-starter-websocket
        

• 配置类（WebSocketConfig）
  使用@ServerEndpoint创立websocket endpoint [配置端点类以及具体实现：WebSocketServer] 首先要注入ServerEndpointExporter，这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint。 要注意，如果使用独立的servlet容器，而不是直接使用springboot的内置容器，就不要注入ServerEndpointExporter，因为它将由容器自己提供和管理。
• 配置类的实现

package com.dianxin.msg.websocketconfig;

import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import javax.websocket.server.ServerEndpointConfig;

public class MyEndpointConfigure extends ServerEndpointConfig.Configurator implements ApplicationContextAware {
    private static volatile BeanFactory context;
 
	@Override
	public  T getEndpointInstance(Class clazz) throws InstantiationException {
		return context.getBean(clazz);
	}
 
	@Override
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
		MyEndpointConfigure.context = applicationContext;
	}
}

package com.dianxin.msg.websocketconfig;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;


@Configuration
public class WebSocketConfig {
    
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }

    
    @Bean
    public MyEndpointConfigure newConfigure() {
        return new com.dianxin.msg.websocketconfig.MyEndpointConfigure();
    }
}

Service 层实现

package com.dianxin.msg.service;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;


@Component
@ServerEndpoint(value = "/socket/{name}/{token}")
public class WebSocketServer {

    //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    private static AtomicInteger online = new AtomicInteger();

    //concurrent包的线程安全Set，用来存放每个客户端对应的WebSocketServer对象。
    private static Map sessionPools = new HashMap<>();

    
    public void sendMessage(Session session, String message) throws IOException{
        if(session != null){
            session.getBasicRemote().sendText(message);
        }
    }


    
    @OnOpen
    public void onOpen(Session session, @PathParam(value = "name") String userName, @PathParam(value = "token") String token){
        sessionPools.put(userName, session);
        addOnlineCount();
        System.out.println(userName + "加入webSocket！当前人数为" + online);
        try {
            sendMessage(session, "欢迎" + userName + "加入连接！");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    
    @OnClose
    public void onClose(@PathParam(value = "name") String userName){
        sessionPools.remove(userName);
        subOnlineCount();
        System.out.println(userName + "断开webSocket连接！当前人数为" + online);
    }

    
    @OnMessage
    public void onMessage(String message) throws IOException{
        for (Session session: sessionPools.values()) {
            try {
                sendMessage(session, message);
            } catch(Exception e){
                e.printStackTrace();
                continue;
            }
        }
    }

    
    @OnError
    public void onError(Session session, Throwable throwable){
        System.out.println("发生错误");
        throwable.printStackTrace();
    }

    
    public void sendInfo(String userName, String message){
        Session session = sessionPools.get(userName);
        try {
            sendMessage(session, message);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    
    public Session getSession(String userName){
        Session session = sessionPools.get(userName);
        return session;
    }

    public static void addOnlineCount(){
        online.incrementAndGet();
    }

    public static void subOnlineCount() {
        online.decrementAndGet();
    }
}

Controller 层实现(用来测试)

package com.dianxin.msg.controller;


import com.dianxin.msg.service.WebSocketServer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.io.IOException;



@RequestMapping("/webSocket")
@RestController
public class WebSocketController {


    @Autowired
    private WebSocketServer webSocketServer;

    
    @RequestMapping(value = "/socket", method = RequestMethod.GET)
    public void testSocket1(@RequestParam String userName, @RequestParam String message){
        webSocketServer.sendInfo(userName, message);
    }


    
    @RequestMapping(value = "/socket/all", method = RequestMethod.GET)
    public void pushAllSocket(@RequestParam String message){
        try {
            webSocketServer.onMessage(message);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}