Springboot整合websocket实现一对一消息推送和广播消息推送

https://www.jianshu.com/p/60799f1356c5

https://blog.csdn.net/Ouyzc/article/details/79994401

maven依赖

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

项目实战

1，设置webSocket终端服务

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();
    }  
}  

2，消息处理

import com.chitic.module.core.enums.ChiticResponseCode;
import com.chitic.module.core.exception.ChiticException;
import com.chitic.module.core.util.SpringUtils;
import com.chitic.module.sso.api.common.UserCacheInfo;
import com.chitic.module.sso.config.TokenService;
import org.apache.commons.lang3.StringUtils;
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.concurrent.ConcurrentHashMap;

@ServerEndpoint(value = "/socketServer/{clientId}")
@Component
public class SocketServer {

    private Session session;
    public static ConcurrentHashMap<String, Session> sessionPool = new ConcurrentHashMap<>();
    public static ConcurrentHashMap<String, String> sessionIds = new ConcurrentHashMap<>();

    public void setSession(Session session) {
        this.session = session;
    }

    private String token(String clientId) {
        if (StringUtils.isNotBlank(clientId)) {
            String[] split = clientId.split("_");
            return split[0];
        }
        return null;
    }

    @OnOpen
    public void open(Session session, @PathParam(value = "clientId") String clientId) {
        this.session = session;
        TokenService tokenService = SpringUtils.getBean(TokenService.class);
        String token = token(clientId);
        if (null == token) {
            throw ChiticException.of(ChiticResponseCode.ACCESS_DENY);
        }
        UserCacheInfo userCacheInfo = tokenService.getByToken(token);
        if (null == userCacheInfo) {
            throw ChiticException.of(ChiticResponseCode.ACCESS_DENY);
        }
        sessionPool.put(clientId, session);
        sessionIds.put(session.getId(), clientId);
    }

    @OnMessage
    public void onMessage(String message) {
        String sessionId = session.getId();
        sendMessage(message, sessionIds.get(sessionId));
    }

    @OnClose
    public void onClose() {
        sessionPool.remove(sessionIds.get(session.getId()));
        sessionIds.remove(session.getId());
    }

    @OnError
    public void onError(Session session, Throwable error) {
        error.printStackTrace();
    }

    public synchronized static void sendMessage(String message, String clientId) {
        Session s = sessionPool.get(clientId);
        if (s != null) {
            try {
                s.getBasicRemote().sendText(message);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

}

3，发送消息（可以在MQ如kafka生产消息）

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.websocket.Session;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;

/*
  * @Author 
  * @Description websocket推送
  * @Date 8:46 2019/7/17
  * @Param
  * @return
  **/
@Service
@AllArgsConstructor
@Slf4j
public class DataDevNowWsBizService {

    private final WaterUnitRepository waterUnitRepository;private String houseId(String clientId) {
        if (StringUtils.isNotBlank(clientId)) {
            String[] split = clientId.split("_");
            return split[split.length - 1];
        }
        return null;
    }

    public void pushWs(HubDataDevKafkaResponse data) {
        ConcurrentHashMap<String, Session> sessionPool = SocketServer.sessionPool;
        if (CollectionUtils.isEmpty(sessionPool.keySet())) {
            return;
        }
        sessionPool.keySet().forEach(r -> {
            String house = houseId(r);
            if (null != house && house.equals("" + houseId)) {
                Session session = sessionPool.get(r);
                SocketServer socketServer = new SocketServer();

                socketServer.setSession(session);
                socketServer.onMessage(JacksonUtil.toString(ChiticWsResponse.of(r, WsDataType.POINT.getCode(), wsPointResponse)));
                socketServer.onMessage(JacksonUtil.toString(ChiticWsResponse.of(r, WsDataType.PUMPS.getCode(), dataDevNowWsResponsePumps)));

 } }); } }

第一种

常量

/**
 * @Description TODO webSocket相关常量
 * @Author GX
 * @Date 2019/6/12 11:16
 * @Version V1.0
 **/
public interface SocketConstant {

    /**
     * 链接地址
     */
    String WEBSOCKETPATHPERFIX = "/ws-push";
    String WEBSOCKETPATH = "/endpointWisely";
    /**
     * 消息代理路径
     */
    String WEBSOCKETBROADCASTPATH = "/topic";
    /**
     * 前端发送给服务端请求地址
     */
    String FORETOSERVERPATH = "/welcome";
    /**
     * 服务端生产地址,客户端订阅此地址以接收服务端生产的消息
     */
    String PRODUCERPATH = "/topic/getResponse";
    /**
     * 点对点消息推送地址前缀
     */
    String P2PPUSHBASEPATH = "/user";
    /**
     * 点对点消息推送地址后缀,最后的地址为/user/用户识别码/msg
     */
    String P2PPUSHPATH = "/msg";
}

接收前端消息实体

import lombok.Data;
/**
 * @Description TODO 接收前端消息实体
 * @Author GX
 * @Date 2019/6/12 11:21
 * @Version V1.0
 **/
@Data
public class WiselyMessage {

    private String name;

}

后台发送消息实体

import lombok.AllArgsConstructor;
import lombok.Data;

/**
 * @Description TODO 后台发送消息实体
 * @Author GX
 * @Date 2019/6/12 11:21
 * @Version V1.0
 **/
@Data
@AllArgsConstructor
public class WiselyResponse {

    private String responseMessage;

}

配置websocket

import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;

/**
 * @EnableWebSocketMessageBroker注解用于开启使用STOMP协议来传输基于代理（MessageBroker）的消息，
 * 这时候控制器（controller）开始支持@MessageMapping,就像是使用@requestMapping一样。
 * @Description TODO 配置类
 * @Author GX
 * @Date 2019/6/12 11:21
 * @Version V1.0
 **/
@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void registerStompEndpoints(StompEndpointRegistry stompEndpointRegistry) {
        //注册一个Stomp的节点（endpoint）,并指定使用SockJS协议。
        //TODO 注意： 涉及到跨域问题  需加.setAllowedOrigins("*") 
        stompEndpointRegistry.addEndpoint(SocketConstant.WEBSOCKETPATH).setAllowedOrigins("*").withSockJS();
        //TODO 注意： 不涉及到跨域问题  .setAllowedOrigins("*")可去掉，如下 
        //stompEndpointRegistry.addEndpoint(SocketConstant.WEBSOCKETPATH).withSockJS();
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        //服务端发送消息给客户端的域,多个用逗号隔开
        registry.enableSimpleBroker(SocketConstant.WEBSOCKETBROADCASTPATH, SocketConstant.P2PPUSHBASEPATH);
        //定义一对一推送的时候前缀
        registry.setUserDestinationPrefix(SocketConstant.P2PPUSHBASEPATH);
        //定义websoket前缀
        registry.setApplicationDestinationPrefixes(SocketConstant.WEBSOCKETPATHPERFIX);
    }
}

消息逻辑处理，发送消息

package com.chitic.supplywater.common.config.webSocket;

import lombok.AllArgsConstructor;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Service;

import java.util.List;

@Service
@AllArgsConstructor
public class WebSocketService {

    //@Autowired
    private final SimpMessagingTemplate template;

    /**
     * 广播
     * 发给所有在线用户
     *
     * @param msg
     */
    public void sendMsg(WiselyResponse msg) {
        template.convertAndSend(SocketConstant.PRODUCERPATH, msg);
    }

    /**
     * 发送给指定用户
     * @param users
     * @param msg
     */
    public void send2Users(List<String> users, WiselyResponse msg) {
        users.forEach(userName -> template.convertAndSendToUser(userName, SocketConstant.P2PPUSHPATH, msg));
    }
}

控制器controller

import lombok.AllArgsConstructor;
import org.apache.commons.compress.utils.Lists;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Controller;

import java.util.List;

@Controller
@AllArgsConstructor
public class WsController {

    //@Resource
    private final WebSocketService webSocketService;

    @MessageMapping(SocketConstant.FORETOSERVERPATH)//@MessageMapping和@RequestMapping功能类似，用于设置URL映射地址，浏览器向服务器发起请求，需要通过该地址。
    @SendTo(SocketConstant.PRODUCERPATH)//如果服务器接受到了消息，就会对订阅了@SendTo括号中的地址传送消息。
    public WiselyResponse say(WiselyMessage message) throws Exception {
        List<String> users = Lists.newArrayList();
        //此处写死只是为了方便测试,此值需要对应页面中订阅个人消息的userId, 实际开发中，可以指定将此消息发送给谁
        users.add("aa");
        users.add("bb");
        webSocketService.send2Users(users, new WiselyResponse("admin hello"));
        return new WiselyResponse("Welcome, " + message.getName() + "!");
    }

}

前端页面

<!DOCTYPE html>
<html xmlns:th="http://www.thymeleaf.org">
<head>
    <meta charset="UTF-8" />
    <title>Spring Boot+WebSocket+广播式</title>

</head>
<body onload="disconnect()">
<noscript><h2 style="color: #ff0000">貌似你的浏览器不支持websocket</h2></noscript>
<div>
    <div>
        <button id="connect" onclick="connect();">连接</button>
        <button id="disconnect" disabled="disabled" onclick="disconnect();">断开连接</button>
    </div>
    <div id="conversationDiv">
        <label>输入你的名字</label><input type="text" id="name" />
        <button id="sendName" onclick="sendName();">发送</button>
        <p id="response"></p>
        <p id="response1"></p>
    </div>
</div>
<!--<script th:src="@{sockjs.min.js}"></script>
<script th:src="@{stomp.min.js}"></script>
<script th:src="@{jquery.js}"></script>-->
<script src="https://cdn.bootcss.com/sockjs-client/1.1.4/sockjs.min.js"></script>
<script src="https://cdn.bootcss.com/stomp.js/2.3.3/stomp.min.js"></script>
<script src="https://cdn.bootcss.com/jquery/3.2.1/jquery.min.js"></script>
<script th:inline="javascript">
    var stompClient = null;
    //此值有服务端传递给前端,实现方式没有要求
    var userId = "aa";

    function setConnected(connected) {
        document.getElementById('connect').disabled = connected;
        document.getElementById('disconnect').disabled = !connected;
        document.getElementById('conversationDiv').style.visibility = connected ? 'visible' : 'hidden';
        $('#response').html();
    }

    function connect() {
        var socket = new SockJS('http://localhost:9000/endpointWisely'); //1连接SockJS的endpoint是“endpointWisely”，与后台代码中注册的endpoint要一样。
        stompClient = Stomp.over(socket);//2创建STOMP协议的webSocket客户端。
        stompClient.connect({}, function(frame) {//3连接webSocket的服务端。
            setConnected(true);
            console.log('开始进行连接Connected: ' + frame);
            //4通过stompClient.subscribe（）订阅服务器的目标是'/topic/getResponse'发送过来的地址，与@SendTo中的地址对应。
            stompClient.subscribe('/topic/getResponse', function(respnose){
                showResponse(JSON.parse(respnose.body).responseMessage);
            });
            //4通过stompClient.subscribe（）订阅服务器的目标是'/user/' + userId + '/msg'接收一对一的推送消息,其中userId由服务端传递过来,用于表示唯一的用户,通过此值将消息精确推送给一个用户
            stompClient.subscribe('/user/' + userId + '/msg', function(respnose){
                console.log(respnose);
                showResponse1(JSON.parse(respnose.body).responseMessage);
            });
        });
    }


    function disconnect() {
        if (stompClient != null) {
            stompClient.disconnect();
        }
        setConnected(false);
        console.log("Disconnected");
    }

    function sendName() {
        var name = $('#name').val();
        //通过stompClient.send（）向地址为"/welcome"的服务器地址发起请求，与@MessageMapping里的地址对应。因为我们配置了registry.setApplicationDestinationPrefixes(Constant.WEBSOCKETPATHPERFIX);所以需要增加前缀/ws-push/
        stompClient.send("/ws-push/welcome", {}, JSON.stringify({ 'name': name }));
    }

    function showResponse(message) {
        var response = $("#response");
        response.html(message);
    }
    function showResponse1(message) {
        var response = $("#response1");
        response.html(message);
    }
</script>
</body>
</html>

 第二种

设置WebSocket的配置类

package com.chitic.supplywater.common.config.webSocket;

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

/**
 * 设置webSocket终端服务
 * @Author: gx
 * @Date: 2019/6/10 16:39
 */

@Configuration
public class WebSocketConfig {

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

}

WebSocket消息处理类

package com.chitic.supplywater.common.config.webSocket;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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.concurrent.CopyOnWriteArraySet;

@ServerEndpoint("/webSocket/{sid}")
@Component
public class WebSocketServer {

    static Logger logger= LoggerFactory.getLogger(WebSocketServer.class);
    /**
     * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
     */
    private static int onlineCount = 0;
    //concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
    private static CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet<WebSocketServer>();

    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    private Session session;

    //接收sid
    private String sid="";
    /**
     * 连接建立成功调用的方法*/
    @OnOpen
    public void onOpen(Session session,@PathParam("sid") String sid) {
        this.session = session;
        //加入set中
        webSocketSet.add(this);
        //在线数加1
        addOnlineCount();
        logger.info("有新窗口开始监听:识别码为"+sid+",当前在线人数为" + getOnlineCount());
        this.sid=sid;
        try {
            sendMessage("连接成功");
        } catch (IOException e) {
            logger.error("websocket IO异常");
        }
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        webSocketSet.remove(this);  //从set中删除
        subOnlineCount();           //在线数减1
        logger.info("有一连接关闭！当前在线人数为" + getOnlineCount());
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息*/
    @OnMessage
    public void onMessage(String message, Session session) {
        logger.info("收到来自窗口"+sid+"的信息:"+message);
        //群发消息
        for (WebSocketServer item : webSocketSet) {
            try {
                item.sendMessage(message);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     *
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        logger.error("发生错误");
        error.printStackTrace();
    }
    /**
     * 实现服务器主动推送
     */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }


    /**
     * 群发自定义消息
     * */
    public static void sendInfo(String message,@PathParam("sid") String sid) throws IOException {
        logger.info("推送消息到窗口"+sid+"，推送内容:"+message);
        for (WebSocketServer item : webSocketSet) {
            try {
                //这里可以设定只推送给这个sid的，为null则全部推送
                if(sid==null) {
                    item.sendMessage(message);
                }else if(item.sid.equals(sid)){
                    item.sendMessage(message);
                }
            } catch (IOException e) {
                continue;
            }
        }
    }

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount--;
    }
}

前端代码

<!DOCTYPE HTML>
<html>
<head>
    <title>My WebSocket</title>
</head>
<body>
Welcome<br/>
<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;
    //判断当前浏览器是否支持WebSocket
    if ('WebSocket' in window) {
        websocket = new WebSocket("ws://localhost:9000/webSocket/1");
    }
    else {
        alert('Not support websocket')
    }
    //连接发生错误的回调方法
    websocket.onerror = function () {
        setMessageInnerHTML("error");
    };
    //连接成功建立的回调方法
    websocket.onopen = function (event) {
        setMessageInnerHTML("open");
    }
    //接收到消息的回调方法
    websocket.onmessage = function (event) {
        setMessageInnerHTML(event.data);
    }
    //连接关闭的回调方法
    websocket.onclose = function () {
        setMessageInnerHTML("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>

其中： sid可以控制将消息推送给那个用户