使用websocket给前台推送消息

Posted 2023-02-02 missccy

1.pom配置

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

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;

/**
 * @类名 WebSocketConfig
 * @描述 WebSocket配置类
 * @版本 1.0
 * @创建人 lying
 * @创建时间 2020/5/27 14:54
 * @修改人 lying
 * @修改时间 2020/5/27 14:54
 **/
@Configuration
public class WebSocketConfig 

    /**
      * @方法名 serverEndpointExporter
      * @描述 注入ServerEndpointExporter    这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint
      * @参数 [] 
      * @返回值 org.springframework.web.socket.server.standard.ServerEndpointExporter
      * @创建人 lying
      * @创建时间 2020/5/27 14:56
      * @修改人 lying
      * @修改时间 2020/5/27 14:56      
      */
    @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;

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

    /**
     * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
     */
    //private static CopyOnWriteArraySet<WebSocket> webSocketSet = new CopyOnWriteArraySet<>();


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

    private static String merchantId="";

//    public static CopyOnWriteArraySet<WebSocket> getWebSocketSet() 
//        return webSocketSet;
//    

    private static ConcurrentHashMap<String,WebSocket> webSocketMap = new ConcurrentHashMap<>();


    public static ConcurrentHashMap<String, WebSocket> getWebSocketMap() 
        return webSocketMap;
    

    public static void setWebSocketMap(ConcurrentHashMap<String, WebSocket> webSocketMap) 
        WebSocket.webSocketMap = webSocketMap;
    

//    public static void setWebSocketSet(CopyOnWriteArraySet<WebSocket> webSocketSet) 
//        WebSocket.webSocketSet = webSocketSet;
//    

    /**
     * 从数据库查询相关数据信息,可以根据实际业务场景进行修改
     */
    @Resource
    private MerchantNoticeInfoService merchantNoticeInfoService;
    private static MerchantNoticeInfoService merchantNoticeInfoServiceMapper;

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

    /**
     * 连接建立成功调用的方法
     *
     * @param session 会话
     */
    @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));
        
    

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息
     */
    @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();
            
        
    

    /**
     * 发生错误时调用
     *
     * @param session 会话
     * @param session 会话
     * @param error   错误信息
     */
    @OnError
    public void onError(Session session, Throwable error) 
        logger.error("连接异常！");
        error.printStackTrace();
    

    /**
     * 发送信息
     *
     * @param message 消息
     */
    public void sendMessage(String message) throws IOException 
        if (session != null&&session.isOpen())
            this.session.getBasicRemote().sendText(message);
        
    

    /**
     * 自定义消息推送、可群发、单发
     *
     * @param 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<String, WebSocket> webSocketMap = WebSocket.getWebSocketMap();
        for (Map.Entry<String, WebSocket> entry : webSocketMap.entrySet()) 
            Integer count = merchantNoticeInfoMapper.getUnreadCount(new BigInteger(entry.getKey()));
            sendMessage(entry.getKey(), count + "");