WebSocket 双工通信的案例

Dear 丶 2023-01-01 04:00 205阅读 0赞

*  什么是WebSocket

WebSocket是一种在单个TCP连接上进行全双工通信的协议;

*  描述

WebSocket是一种在单个TCP连接上进行全双工通信的协议, 使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。在WebSocket API中，浏览器和服务器只需要完成一次握手，两者之间就直接可以创建持久性的连接，并进行双向数据传输;

*  优点

很多网站为了实现推送技术，所用的技术都是轮询。轮询是在特定的的时间间隔（如每1秒），由浏览器对服务器发出HTTP请求，然后由服务器返回最新的数据给客户端的浏览器。这种传统的模式带来很明显的缺点，即浏览器需要不断的向服务器发出请求，然而HTTP请求可能包含较长的头部，其中真正有效的数据可能只是很小的一部分，显然这样会浪费很多的带宽等资源。  
而比较新的技术去做轮询的效果是Comet。这种技术虽然可以双向通信，但依然需要反复发出请求。而且在Comet中，普遍采用的长链接，也会消耗服务器资源。在这种情况下，HTML5定义了WebSocket协议，能更好的节省服务器资源和带宽，并且能够更实时地进行通讯；

::::不瞎逼逼了，以上概要来自 百度百科

*  直接上代码 WebSocket server

package carloan.common.socket;
    
    import java.io.IOException;
    import java.util.Iterator;
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    
    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.apache.commons.logging.Log;
    import org.apache.commons.logging.LogFactory;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Component;
    
    /**
     * 
     * * @ServerEndpoint 注解是一个类层次的注解，
     *   它的功能主要是将目前的类定义成一个websocket服务器端,
     *   注解的值将被用于监听用户连接的终端访问URL地址,
     *   客户端可以通过这个URL来连接到WebSocket服务器端
     * ClassName: WebSocket
     * @date 2020年12月28日 下午6:07:43
     * @author llh
     * @version 
     * @since JDK 1.8
     */
    @ServerEndpoint("/websocket/{userId}/ws")
    @Component
    public class WebSocket {
    	
    	private static Log log = LogFactory.getLog(WebSocket.class);
    	// 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    	private volatile static int onlineCount = 0;
    	// 客户端回话集合
    	private static ConcurrentHashMap<String, UserSocket> sessions = new ConcurrentHashMap<String, UserSocket>();
    
    
    	class UserSocket {
    		Session session;
    		String userId;
    
    		UserSocket(String uid, Session s) {
    			session = s;
    			userId = uid;
    		}
    
    		public void sendMessage(String message) {
    		log.error("sendMessage,Session:" + this.session.getId() + ";uid=" + this.userId);
    				if (this.session.isOpen()){
    					this.session.getAsyncRemote().sendText(message);
    				}
    		}
    		
    		public int sendOnLineUserCount(int userCount) {
    			this.sendMessage("setOnLineUserCount," + userCount + "");// 在线人数
    			return 1;
    		}
    		
    		public int sendTodotasknumber(String todotasknumbers) {
    			this.sendMessage("todotasknumbers," + todotasknumbers);// 推送代办的数量
    			return 1;
    		}
    		
    		public int sendMessagenumbers(String messagenumbers) {
    			this.sendMessage("messagenumbers," + messagenumbers);// 推送代办站内信数量
    			return 1;
    		}
    
    	}
    
    	/**
    	 * 连接建立成功调用的方法
    	 * 
    	 * @param session
    	 *  可选的参数。session为与某个客户端的连接会话，需要通过它来给客户端发送数据
    	 */
    	@OnOpen
    	public void onOpen(Session session, @PathParam("userId") String userId) {
    
    		log.error("onOpen,Session:" + session.getId() + ";uid:" + userId );
    
    		int userCount = 0;
    		UserSocket us = new UserSocket(userId, session);
    		sessions.put(session.getId(), us);
    		
    		synchronized (WebSocket.class) {
    			userCount = ++onlineCount;
    		}
    		log.error("onOpen Session Count:" + userCount);
    	}
    
    	/**
    	 * 连接关闭调用的方法
    	 */
    	@OnClose
    	public void onClose(Session session, @PathParam("userId") String userId) {
    		log.error("onClose,Session:" + session.getId());
    		int userCount = 0;
    		if (sessions.containsKey(session.getId())) {
    			sessions.remove(session.getId());
    
    			synchronized (WebSocket.class) {
    				userCount = --onlineCount;
    			}
    		}
    		log.error("onClose Session Count:" + userCount);
    	}
    
    	/**
    	 * 收到客户端消息后调用的方法
    	 * @param message 客户端发送过来的消息
    	 * @param session 可选的参数
    	 * @throws IOException 
    	 */
    	@OnMessage
    	public void onMessage(@PathParam("userId") String userId,Session session,String message) throws IOException {
    		log.error("onMessage,Session:" + session.getId() + ";MSG=" + message);
    		sendMessage(message);
    	}
    
    	/**
    	 * 发生错误时调用
    	 * 
    	 * @param session
    	 * @param error
    	 */
    	@OnError
    	public void onError(Session session, Throwable error) {
    		log.error("onError,Session:" + session.getId() + ";MSG=" + error.toString());
    		
    	}
    	
    	
    	int sendcount=0;
    	@Scheduled(cron = "0/2 * * * * ?")
    	public void sendCount(){
    		if (sendcount!=onlineCount){
    			sendcount=onlineCount;
    			setOnLineUserCount(sendcount);
    		}
    	}
    
    
    	/**
    	 * 发给指定用户
    	 * @param userId 用户id
    	 * @param message
    	 * @throws IOException
    	 */
    	public static void sendMessageByUserId(String userId, String message) throws IOException {
    		
    		Iterator<Map.Entry<String, UserSocket>> entries = sessions.entrySet().iterator();  
    		while (entries.hasNext()) {  
    		  
    		    Map.Entry<String, UserSocket> entry = entries.next();  
    		  
    		    System.out.println("Key = " + entry.getKey() + ", Value = " + entry.getValue());  
    		    
    		    if (entry.getValue().userId != null && entry.getValue().userId.equalsIgnoreCase(userId)) {
    		    	entry.getValue().sendMessage(message);
    			}
    		}  
    	}
    	
    	
    	/**
    	 * 发给所有用户
    	 * @param message
    	 * @throws IOException
    	 */
    	public static void sendMessage(String message) throws IOException {
    
    		Iterator<Map.Entry<String, UserSocket>> entries = sessions.entrySet().iterator();
    		while (entries.hasNext()) {
    
    			Map.Entry<String, UserSocket> entry = entries.next();
    
    			System.out.println("Key = " + entry.getKey() + ", Value = " + entry.getValue());
    
    			if (null != entry.getValue().userId) {
    
    				entry.getValue().sendMessage(message);
    
    			}
    			
    		}
    		
    	}
    
    	/**
    	 * 描述：得到在线人数
    	 * @param:
    	 */
    	public static int getOnLineUserCount() {
    			return WebSocket.onlineCount;
    	}
    
    
    	
    	static boolean  isSending=false;
    
    	/**
    	 * 描述：设置在线人数
    	 * @param:
    	 */
    	public static void setOnLineUserCount(int userCount) {
    
    		try{
    			Iterator<Map.Entry<String, UserSocket>> entries = sessions.entrySet().iterator();  
    			int i=0;
    			log.error("开始发送:"+i);
    			while (entries.hasNext()) {  
    			  
    			    Map.Entry<String, UserSocket> entry = entries.next();  
    			    entry.getValue().sendOnLineUserCount(userCount);
    			    i++;
    			} 
    			log.error("发送数量:"+i);
    		}catch(Exception e){
    			e.printStackTrace();
    		}finally {
    			isSending=false;
    		}
    	}
    
    	
    
    	public static void sendWebSocketTodotasknumber(String userId, String todotasknumbers) throws IOException {
    			
    			Iterator<Map.Entry<String, UserSocket>> entries = sessions.entrySet().iterator();  
    			int i=0;
    			log.error("开始发送:"+i);
    			while (entries.hasNext()) {  
    			  
    			    Map.Entry<String, UserSocket> entry = entries.next();  
    			    if (entry.getValue().userId != null && entry.getValue().userId.equalsIgnoreCase(userId)) {
    			    	entry.getValue().sendTodotasknumber(todotasknumbers);
    				}
    			    i++;
    			}  
    			log.error("开始数量:"+i);
    	}
    
    
    	/**
    	 * 
    	 * 描述：代办站内数量推送
    	 * @param:
    	 */
    	public static void sendWebSocketMessagenumbers(String userId, String messagenumbers) throws IOException {
    			
    			Iterator<Map.Entry<String, UserSocket>> entries = sessions.entrySet().iterator();  
    			while (entries.hasNext()) {  
    			  
    			    Map.Entry<String, UserSocket> entry = entries.next();  
    			    if (entry.getValue().userId != null && entry.getValue().userId.equalsIgnoreCase(userId)) {
    			    	entry.getValue().sendMessagenumbers(messagenumbers);
    				}
    			}  
    	}
    
    	/**
    	 * 根据用户id发送消息（通用）
    	 * @param userId 用户ID
    	 * @param businessType 业务类型
    	 * @param message 消息
    	 */
    	public static void sendMessageCommon(String userId, String businessType, String message){
    
    		for (Map.Entry<String, UserSocket> entry : sessions.entrySet()) {
    
    			UserSocket userSocket = entry.getValue();
    
    			if (null != userSocket.userId && userSocket.userId.equals(userId)){
    				userSocket.sendMessage(businessType + "," + message);
    			}
    		}
    	}
    
    }

*  前端 jsp页面代码

<%@ page language="java" contentType="text/html; charset=UTF-8" pageEncoding="UTF-8"%>
    <%@ taglib uri="http://java.sun.com/jsp/jstl/core" prefix="c"%>
    <!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
    <html>
    <body>
        Welcome：<span>index</span><br/>
        <input id="userId" type="hidden"  value="${userId}"/>
        <input id="host" type="hidden"  value="${host}"/>
        <input id="text" type="text" />
        <button onclick="send()">Send</button>
        <button onclick="closeWebSocket()">Close</button>
        <div id="message"></div>
    </body>
    
    <script type="text/JavaScript">
          var websocket = null;
          var userId = document.getElementById('userId').value;
          var host = document.getElementById('host').value;
          //判断当前浏览器是否支持WebSocket
          if('WebSocket' in window){
              websocket = new WebSocket("ws://"+host+"/websocket/"+userId+"/ws?userId="+userId);
          }else{
              alert('Not support websocket');
          }    
          //连接发生错误的回调方法
          websocket.onerror = function(){
        	  console.log("websocket error");
              //setMessageInnerHTML("error");
          };     
          //连接成功建立的回调方法
          websocket.onopen = function(event){
        	  console.log("websocket open");
              //setMessageInnerHTML("websocket open");
          };      
          //接收到消息的回调方法
          websocket.onmessage = function(){
        	  console.log(event.data);
              var note=event.data;
              if(note.split(',')[0]=='setOnLineUserCount'){//在线人数
     			 //$("#usernumber").html(note.split(',')[1]);
     			setMessageInnerHTML("在线人数:"+note.split(',')[1]);
     		 }else if(note.split(',')[0]=='todotasknumbers'){//代办数量
     			 $("#todotasknumbers").html(note.split(',')[1]);
     		 }else if(note.split(',')[0]=='bidBusiessPolicyOrderUpdate'){//核保审核结果
     			 //$("#todotasknumbers").html(note.split(',')[1]);
     			setMessageInnerHTML("核保审核结果:"+note.split(',')[1]);
     		 }else{
            	  setMessageInnerHTML(event.data);
     		 }
              
              
          };     
          //连接关闭的回调方法
          websocket.onclose = function(){
        	  console.log("websocket close");
              //setMessageInnerHTML("websocket close");
          };       
          //监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
          window.onbeforeunload = function(){
              websocket.close();
          };       
          //将消息显示在网页上
          function setMessageInnerHTML(innerHTML){
              document.getElementById('message').innerHTML += innerHTML + '<br/>';
          } 
          //关闭连接
          function closeWebSocket(){
              websocket.close();
          }  
          //发送消息
          function send(){
              var message = document.getElementById('text').value;
              websocket.send(message);
          }
      </script>
    
    </html>

*  ws.jsp页面跳转 Controller

@Controller
    @RequestMapping("/webSocket")
    public class WebSocketController {
        @Value("${server.host}")
        private String serverUrl;
    	
        @EscapeLogin
        @RequestMapping("/socket")
        public String index(HttpSession session,Model mode){
        	mode.addAttribute("userId", 114540);
        	mode.addAttribute("host", serverUrl);
            return"websocket/ws";
        }
    
    }

*  Socket 安全认证

非spring boot 项目架构 这里集成已过滤器Filter ,没做token 校验，就以userId有效性判断;

package carloan.core.filter;
    
    import java.io.IOException;
    import java.io.PrintWriter;
    import java.nio.charset.StandardCharsets;
    
    import javax.servlet.Filter;
    import javax.servlet.FilterChain;
    import javax.servlet.FilterConfig;
    import javax.servlet.ServletException;
    import javax.servlet.ServletRequest;
    import javax.servlet.ServletResponse;
    import javax.servlet.http.HttpServletRequest;
    import javax.servlet.http.HttpServletResponse;
    
    import org.apache.commons.lang3.StringUtils;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.http.MediaType;
    import org.springframework.web.context.support.SpringBeanAutowiringSupport;
    
    import com.alibaba.fastjson.JSON;
    
    import carloan.common.param.ApiResponse;
    import carloan.permission.domain.PmsUser;
    import carloan.permission.mapper.PmsUserMapper;
    
    /**
     * 
     *
     * ClassName: SessionSocketFilter
     * @Description:WebSocket 过滤认证
     * @date 2020年12月29日 下午4:44:04
     * @author llh
     * @version 
     * @since JDK 1.8
     */
    public class SessionWebSocketFilter extends SpringBeanAutowiringSupport implements Filter {
    
        private static final Logger logger = LoggerFactory.getLogger(SessionWebSocketFilter.class);
        @Autowired
    	private PmsUserMapper pmsUserMapper;
    
        public void init(FilterConfig filterConfig) throws ServletException {
        	logger.error("服务器启动>>>>>>>>>");
        }
    
        public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException,
                ServletException {
            HttpServletRequest httpRequest = (HttpServletRequest) request;
            HttpServletResponse httpResponse = (HttpServletResponse) response;
            logger.error("request url={}",httpRequest.getRequestURI());
            String userId = (String)httpRequest.getParameter("userId");
            if(StringUtils.isEmpty(userId)){
            	handleFailure(httpResponse, ApiResponse.AUTHENTICATION_FAILURE);
            	return;
            }
            PmsUser userInfo = pmsUserMapper.selectByPrimaryKey(Integer.parseInt(userId));
            String phone = pmsUserMapper.selectDepositUserByuserId(Integer.parseInt(userId));
            if(!(null!=userInfo || null!=phone)){
            	handleFailure(httpResponse, ApiResponse.AUTHENTICATION_FAILURE);
            	return;
            }
            chain.doFilter(request, response);
        }
        
        public void destroy() {
        	logger.error("服务器关闭>>>>>>>>>");
        }
    
        private void handleFailure(HttpServletResponse response, ApiResponse apiResponse) throws IOException {
            response.setStatus(HttpServletResponse.SC_OK);
            response.setContentType(MediaType.APPLICATION_JSON_VALUE);
            response.setCharacterEncoding(StandardCharsets.UTF_8.name());
            PrintWriter out = response.getWriter();
            out.println(JSON.toJSONString(apiResponse));
            out.flush();
        }
    
    }

*  web.xml 配置过滤器

<filter>
            <filter-name>sessionWebSocketFilter</filter-name>
            <filter-class>carloan.core.filter.SessionWebSocketFilter</filter-class>
        </filter>
        <filter-mapping>
            <filter-name>sessionWebSocketFilter</filter-name>
            <url-pattern>/websocket/*</url-pattern>
        </filter-mapping>

启动服务 浏览器甩 http://127.0.0.1:7081/webSocket/socket

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWxpaHVpMTk4OA_size_16_color_FFFFFF_t_70_pic_center]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWxpaHVpMTk4OA_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/581bc1562ff749558e02f05e504287f0.png