React+SpringBoot通过WebSocket实时统计在线人数

一、基本概念

WebSocket 是一种网络通信协议，如果服务器有连续的状态变化，客户端要获知就非常麻烦。大多数 Web 应用程序将通过频繁的异步请求实现长轮询。轮询的效率低，非常浪费资源（因为必须不停连接，或者 HTTP 连接始终打开）所以这里使用WebSocket 通过登录后跳转到首页，向后台WebSocket 建立长链接来达到"即使通讯"，随着用户页面打开或关闭后台群发消息来实时更改页面显示的人数，当然这里目前不涉及登录后的上线下线以及帐号登录挤掉功能，如果需要可以通过发送消息来改变。

二、SpringBoot 后台实现WebSocket

pom.xml

	<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
        </dependency>

WebSocketConfig.java

package com.kero99.socket;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
/**
 * 开启WebSocket支持
 * @author ygc
 */
@Configuration  
public class WebSocketConfig {  
    @Bean  
    public ServerEndpointExporter serverEndpointExporter() {  
        return new ServerEndpointExporter();  
    }  
} 

WebSocketController.java

package com.kero99.socket;


import java.io.IOException;

import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

/**
 * WebSocket服务器端推送消息示例Controller
 * 
 * @author ygc
 *
 */
@RestController
@RequestMapping("/scoket")
public class WebSocketController {
//	@Autowired
//	private RedisOperator redisOperator;
	@RequestMapping(value="/sendAll", method=RequestMethod.GET)
	/**
	 * 群发消息内容
	 * @param message
	 * @return
	 */
	String sendAllMessage(@RequestParam(required=true) String message){
		try {
			WebSocketServer.BroadCastInfo(message);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return "ok";
	}
	@RequestMapping(value="/sendOne", method=RequestMethod.GET)
	/**
	 * 指定会话ID发消息
	 * @param message 消息内容
	 * @param id 连接会话ID
	 * @return
	 */
	String sendOneMessage(@RequestParam(required=true) String message,@RequestParam(required=true) String id){
		try {
			WebSocketServer.SendMessage(id,message);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return "ok";
	}
}

WebSocketServer.java

package com.kero99.socket;

import java.io.IOException;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicInteger;

import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

/**
 * WebSocket服务端
 * @author ygc
 */
@ServerEndpoint(value = "/websocket")
@Component
public class WebSocketServer {
	
	private final static Logger log = LoggerFactory.getLogger(WebSocketServer.class);
	private static final AtomicInteger OnlineCount = new AtomicInteger(0);
	// concurrent包的线程安全Set，用来存放每个客户端对应的Session对象。
	private static CopyOnWriteArraySet<Session> SessionSet = new CopyOnWriteArraySet<Session>();
	

	/**
	 * 连接建立成功调用的方法
	 * @throws IOException 
	 */
	@OnOpen
	public void onOpen(Session session) throws IOException {
		SessionSet.add(session); 
		int personCount = OnlineCount.incrementAndGet(); // 在线数加1
		System.out.println("有连接加入，当前连接数为："+personCount);
		log.info("有连接加入，当前连接数为：{}", personCount);
//		SendMessage(session, "连接成功,当前连接人数为:"+personCount);
//		SendMessage(session,String.valueOf(personCount));
		BroadCastInfo(String.valueOf(OnlineCount.get()));
	}
	
	/**
	 * 连接关闭调用的方法
	 * @throws IOException 
	 */
	@OnClose
	public void onClose(Session session) throws IOException {
		int personCount = OnlineCount.decrementAndGet();
		System.out.println("有连接关闭，当前连接数为："+personCount);
		log.info("有连接关闭，当前连接数为：{}", personCount);
		SessionSet.remove(session);	
	}

	/**
	 * 收到客户端消息后调用的方法
	 * 
	 * @param message
	 *            客户端发送过来的消息
	 * @throws IOException 
	 */
	@OnMessage
	public void onMessage(String message, Session session) throws IOException {
		log.info("来自客户端的消息：{}",message);
//		System.out.println("来自客户端的消息:"+message);
//		SendMessage(session, "收到消息，消息内容："+message);
		if(message.equals("管理平台")) {
			System.out.println("收到平台类型:"+message);
		}
//		if(message.equals("新增人数")) {
//			System.out.println("打开页面:"+message);
//			BroadCastInfo(String.valueOf(OnlineCount.get()+1));
//		}
		if(message.equals("关闭页面")) {
			System.out.println("收到关闭页面:"+message);
			//在线数加-1
			BroadCastInfo(String.valueOf(OnlineCount.get()-1));
		}		
	}

	/**
	 * 出现错误
	 * @param session
	 * @param error
	 */
	@OnError
	public void onError(Session session, Throwable error) {
		log.error("发生错误：{}，Session ID： {}",error.getMessage(),session.getId());
		System.out.println("发生错误：{}，Session ID： "+error.getMessage()+session.getId());
		error.printStackTrace();
	}

	/**
	 * 发送消息，实践表明，每次浏览器刷新，session会发生变化。
	 * @param session
	 * @param message
	 */
	public static void SendMessage(Session session, String message) {
		try {
			session.getBasicRemote().sendText(message);
//			session.getBasicRemote().sendText(String.format("%s (From Server，Session ID=%s)",message,session.getId()));
		} catch (IOException e) {
			log.error("发送消息出错：{}", e.getMessage());
			System.out.println("发送消息出错：{}"+e.getMessage());
			e.printStackTrace();
		}
	}

	/**
	 * 群发消息
	 * @param message
	 * @throws IOException
	 */
	public static void BroadCastInfo(String message) throws IOException {
		for (Session session : SessionSet) {
			if(session.isOpen()){
				SendMessage(session, message);
			}
		}
	}

	/**
	 * 指定Session发送消息
	 * @param sessionId
	 * @param message
	 * @throws IOException
	 */
	public static void SendMessage(String sessionId,String message) throws IOException {
		Session session = null;
		for (Session s : SessionSet) {
			if(s.getId().equals(sessionId)){
				session = s;
				break;
			}
		}
		if(session!=null){
			SendMessage(session, message);
		}
		else{
			log.warn("没有找到你指定ID的会话：{}",sessionId);
			System.out.println("没有找到你指定ID的会话:"+sessionId);
		}
	}
	
}

三、React+Umi+Antd 实现前端 WebSocket 通讯

js 需要安装 socket.io

命令: npm  socket.io 或者 yarn add socket.io

  componentDidMount() { 
 let ws = new WebSocket("ws://localhost:12935/20191108_V1.0_xdnx/websocket");
    if (typeof (WebSocket) == "undefined") {
      console.log("遗憾：您的浏览器不支持WebSocket");
    } else {
      console.log("恭喜：您的浏览器支持WebSocket");
      ws.onopen = (evt)=> {
        console.log("Connection open ...");
        ws.send("管理平台");
        ws.send("新增人数");
      };

      ws.onmessage = (evt)=> {
        console.log( "Received Message: " + evt.data);
        // alert(evt.data)
        //this.state.messageData 为接受数据的变量
        let messageData=this.state.messageData;
        this.setState({
          messageData:evt.data
        })
        // ws.close();
      };
      ws.onclose = (evt)=> {
        // alert(evt.data)
        console.log("Connection closed.");
        // ws.close();
      };
      ws.onerror = (evt)=> {
        console.log("error")
      };
      window.onbeforeunload = (event)=> {
        console.log("关闭WebSocket连接！");
        ws.send("关闭页面");
        event.close();
      }
  
  }
｝

render html

这里用的antd的统计数值控件

          <div  style={{
            background: '#ececec',
            padding: '10px',
            width:'20%',
            float:'left',marginTop:'20px'
          }}>
            <Row gutter={16}>
              <Col span={12}>
                <Card>
                  <Statistic
                    title="管理平台当前在线人数"
                    value={this.state.messageData}
                    precision={0}
                    valueStyle={{ color: '#3f8600' }}
                    suffix="人"
                  />
                </Card>
              </Col>

            </Row>
          </div>

四、实现结果