使用websocket给前台推送消息

1.pom配置

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

2.配置文件WebSocketConfig

package com.jingmai.video.live.operation.config;

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

3.发送消息类

merchantId为前台传来的参数

package com.jingmai.video.live.operation.config;

import com.jingmai.video.live.common.utils.RedisUtil;
import com.jingmai.video.live.operation.service.MerchantNoticeInfoService;
import lombok.Data;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.math.BigInteger;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import javax.websocket.*;
import com.alibaba.fastjson.JSON;


@ServerEndpoint(value = "/webSocket/{merchantId}")//主要是将目前的类定义成一个websocket服务器端, 注解的值将被用于监听用户连接的终端访问URL地址,客户端可以通过这个URL来连接到WebSocket服务器端
@Component
@EnableScheduling// cron定时任务
@Data
public class WebSocket {
    private static final Logger logger = LoggerFactory.getLogger(WebSocket.class);

    
    //private static CopyOnWriteArraySet webSocketSet = new CopyOnWriteArraySet<>();


    
    private Session session;

    private static String merchantId="";

//    public static CopyOnWriteArraySet getWebSocketSet() {
//        return webSocketSet;
//    }

    private static ConcurrentHashMap webSocketMap = new ConcurrentHashMap<>();


    public static ConcurrentHashMap getWebSocketMap() {
        return webSocketMap;
    }

    public static void setWebSocketMap(ConcurrentHashMap webSocketMap) {
        WebSocket.webSocketMap = webSocketMap;
    }

//    public static void setWebSocketSet(CopyOnWriteArraySet webSocketSet) {
//        WebSocket.webSocketSet = webSocketSet;
//    }

    
    @Resource
    private MerchantNoticeInfoService merchantNoticeInfoService;
    private static MerchantNoticeInfoService merchantNoticeInfoServiceMapper;

    @PostConstruct
    public void init() {
        WebSocket.merchantNoticeInfoServiceMapper = this.merchantNoticeInfoService;
    }

    
    @OnOpen
    public void onOpen(@PathParam(value = "merchantId") String merchantId, Session session) throws Exception {
        merchantId = merchantId;//接收到发送消息的人员编号
        this.session = session;
        //webSocketSet.add(this);
        webSocketMap.put(merchantId,this);

        //根据商户id获取未读消息条数
        if(StringUtils.isNotEmpty(merchantId)&&!"undefined".equals(merchantId)){
            BigInteger merchantIdBig=new BigInteger(merchantId);
            int nowOnline = merchantNoticeInfoServiceMapper.getUnreadCount(merchantIdBig);
            this.sendMessage(JSON.toJSONString(nowOnline));
        }
    }

    
    @OnMessage
    public void onMessage(String message, Session session) throws IOException {
        logger.info("参数信息：{}", message);
        //群发消息
        for(String key : webSocketMap.keySet()){
            String merchantIdStr=key;
            WebSocket web=webSocketMap.get(key);
            try {
                web.sendMessage(JSON.toJSONString(message));
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    
    @OnClose
    public void onClose() {
        if (!merchantId.equals("")) {
            webSocketMap.remove(merchantId);  //从set中删除
        }

        if (session != null) {
            try {
                session.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    
    @OnError
    public void onError(Session session, Throwable error) {
        logger.error("连接异常！");
        error.printStackTrace();
    }

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

    
    public static void sendInfo(String message) throws IOException {
        logger.info("信息:{}", message);
        for(String key : webSocketMap.keySet()){
            String merchantIdStr=key;
            WebSocket web=webSocketMap.get(key);
            try {
                web.sendMessage(JSON.toJSONString(message));
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

4.调用

//根据meerchatId推送
    private void sendMessage(String merchantId, String message) {
        try {
            WebSocket webSocket = WebSocket.getWebSocketMap().get(merchantId);
            if (webSocket != null) {
                webSocket.sendMessage(message);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    //循环所有推送
    private void sendBroadcast() {
        ConcurrentHashMap webSocketMap = WebSocket.getWebSocketMap();
        for (Map.Entry entry : webSocketMap.entrySet()) {
            Integer count = merchantNoticeInfoMapper.getUnreadCount(new BigInteger(entry.getKey()));
            sendMessage(entry.getKey(), count + "");
        }
    }