springboot websocket消息推送和查询服务

weblog 精帖
693

某些情形中如果想要实现前端页面刷新,那么一个比较好的办法就是用websocket实现。应该是比ajax轮询要好吧~。既然是websocket主动推送消息,那么服务端查询和推送消息的时机就很重要,也就是说当有websocket服务连接的时候才去查询和发送,当没有websocket连接的时候,就不需要查询和推送(总不能一直查询,再看有没有链接,有就发送,没有就算了,这样显然是不行的)。

那么怎么做才能更好呢~ 下面就说一个 个人感觉还行的方案

在项目启动的时候,开启一个查询服务的线程,在该线程中调用查询服务的方法,方法中写一个while循环,循环时判断当websocket的连接数小于等于0时,让该线程处于等待状态,线程执行停止。直到有websocket连接成功时,唤醒此等待的线程继续在循环中查询,过程中可以设置循环的间隔时间,比如2秒查询一次。

以springboot项目为例,整体的结构为

项目启动时执行
package com.example.demo.service;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
/**
 *项目启动执行
 */
@Component
@Order(value = 1)
public class StartRunner implements ApplicationRunner {
	/**
	 * 查询服务
	 */
	@Autowired
	private WebDataService webDataService;
    @Override
    public void run(ApplicationArguments applicationArguments){
    	/**
    	 * 开启一个查询服务的线程
    	 */
    	new Thread(new Runnable() {
			@Override
			public void run() {
				webDataService.test();
			}
		}).start();
    }
}

 注意的是:不要再StartRunner 的run方法中直接调用webDateService.test()方法,否则该方法后面的代码将不会执行。

查询服务
package com.example.demo.service;
import org.springframework.stereotype.Service;
/***
 * 查询服务
 * @author jiajia
 *
 */
@Service
public class WebDataService {
	private Object obj=new Object();
	public void test() {
		System.out.println("开始查找");
		while(true) {
			if(WebSocketServer.getOnlineCount()<=0) {
				try {
					synchronized (obj) {
						obj.wait();
					}
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			System.out.println("找到消息");
			/**
			 * 查找数据库或其他
			 * 发送消息入口
			 */
			WebSocketServer.sendMessageGroup("找到消息");
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}
	public void ok() {
		synchronized (obj) {
			obj.notify();
		}
	}
}
websocket配置 
package com.example.demo.service;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
/**
 * websocket配置
 * @author jiajia
 */
@Configuration  
public class WebSocketConfig {  
    @Bean  
    public ServerEndpointExporter serverEndpointExporter() {  
        return new ServerEndpointExporter();  
    }  
}

 

websocket服务
package com.example.demo.service;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import org.springframework.stereotype.Component;
import com.example.demo.util.SpringUtil;
/**
 * springboot集成websocket
 * @author jiajia
 */
@ServerEndpoint("/websocket/{id}")
@Component
public class WebSocketServer {
    public static final Object lock=new Object(); 
    /**
     * 记录在线人数
     */
    private static int onlineCount = 0;
    /**
     * WebSocketServer对象集合
     */
    private static final Map<String,WebSocketServer> webSocketMap = new HashMap<String,WebSocketServer>();
    /**
     * 回话
     */
    private Session session;
    /**
     * id标识
     */
    private String id;
    
    private static WebDataService webDataService=null;
     
    /**
     * 连接成功时调用
     * @param session
     * @param sid
     */
    @OnOpen
    public void onOpen(Session session,@PathParam("id") String id) {
    	if(webDataService==null) {
    		webDataService=SpringUtil.getBean(WebDataService.class);
    	}
        this.session = session;
        this.id=id;
        webSocketMap.put(id,this);
        addOnlineCount();
        webDataService.ok();
        System.err.println("连接成功"+id);
    }
 
    /**
     * 连接关闭时调用
     */
    @OnClose
    public void onClose() {
    	webSocketMap.remove(this.id);
        subOnlineCount();
        System.err.println("关闭连接:"+this.id);
    }
 
    /***
     * 接受客户端发送消息时调用
     * @param message 消息
     * @param session 
     */
    @OnMessage
    public void onMessage(String message, Session session) {
    	System.err.println("收到消息:"+message);
    }
     
    /**
     * 群发消息
     * @param message
     * @return
     */
    public static void sendMessageGroup(String message) {
    	Iterator<Map.Entry<String, WebSocketServer>> it = webSocketMap.entrySet().iterator();  
        while(it.hasNext()){  
            try {
            	Map.Entry<String, WebSocketServer> entry = it.next();
				entry.getValue().session.getBasicRemote().sendText(message);
			} catch (Exception e1) {
			}
        }
    }
 
    /**
     * 连接发生异常时调用
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        System.err.println("链接异常:"+this.id);
        subOnlineCount();
    }
    
    /**
     * 获取在线人数
     * @return
     */
    public static synchronized int getOnlineCount() {
        return onlineCount;
    }
 
    /**
     * 在线人数+1
     */
    private static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }
 
    /**
     * 在线人数-1
     */
    private static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount--;
    }
}
 如何获取ioc容器管理的对象
package com.example.demo.util;

import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
/***
 * 获取ioc容器管理的对象
 * @author jiajia
 *
 */
@Component
public class SpringUtil implements ApplicationContextAware{
     
    private static ApplicationContext applicationContext;
     
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
       if(SpringUtil.applicationContext == null) {
           SpringUtil.applicationContext = applicationContext;
       }
    }
     
    //获取applicationContext
    public static ApplicationContext getApplicationContext(){
        return applicationContext;
    }
 
    //通过name获取 Bean.
    public static Object getBean(String name){
        return getApplicationContext().getBean(name);
    }
 
    //通过class获取Bean.
    public static <T> T getBean(Class<T> clazz){
        return getApplicationContext().getBean(clazz);
    }
 
    //通过name,以及Clazz返回指定的Bean
    public static <T> T getBean(String name,Class<T> clazz){
        return getApplicationContext().getBean(name, clazz);
    }
}
 前端案例
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>Insert title here</title>
 <script src="https://code.jquery.com/jquery-3.3.1.min.js" type="text/javascript"></script>
</head>
<body>
	
</body>
	<script type="text/javascript">
		function S4() {
	        return (((1+Math.random())*0x10000)|0).toString(16).substring(1);
	    }
		var socket;  
	    if(typeof(WebSocket) == "undefined") {  
	        console.log("您的浏览器不支持WebSocket");  
	    }else{  
	        console.log("您的浏览器支持WebSocket");  
	            socket = new WebSocket("ws://"+window.location.host+"/websocket/"+S4());
	            socket.onopen = function() {  
	                console.log("socket连接成功");  
	            };  
	            socket.onmessage = function(event) {  
	            	console.log("收到消息:"+event.data);
	            };  
	            socket.onclose = function() {  
	                console.log("socket已关闭");  
	            };  
	            socket.onerror = function() {  
	                alert("socket发生了错误");  
	            } 
	    }
	    window.onbeforeunload = function(){ socket.close(); }
	</script>

</html>

 

猜你喜欢