redis做websocket分布式消息推送服務(wù)的實(shí)現(xiàn)

更新時(shí)間：2024年12月16日 09:21:25 作者：青龍老賊

本文介紹了使用Redis作為消息隊(duì)列實(shí)現(xiàn)WebSocket分布式消息推送服務(wù)的方案,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧

應(yīng)用場(chǎng)景說(shuō)明：

由于redis并非專業(yè)的MQ中間件，消息的防丟失策略并不完整，存在丟失消息的可能。該方案為在再pc web管理平臺(tái)的右下角彈出，顯示新接收到的消息數(shù)，哪怕沒(méi)有收到這個(gè)通知，也可以自己在消息中心看看。所以對(duì)可靠性要求不高。如果業(yè)務(wù)場(chǎng)景要求可靠性高，還是請(qǐng)使用專業(yè)的MQ中間件。該方案已在多個(gè)實(shí)際項(xiàng)目中運(yùn)行。

流程架構(gòu)：

websocket實(shí)現(xiàn)同一賬戶多點(diǎn)登錄、websocket服務(wù)多節(jié)點(diǎn)部署推送方案。

簡(jiǎn)單架構(gòu)圖

假設(shè)用戶A在兩個(gè)地方登錄，連接到兩個(gè)websocketServer服務(wù)節(jié)點(diǎn)1和2，用戶B連接到2節(jié)點(diǎn)。

websocketServer將websocket session保存在各自的Map<String,Session>中，key為userid，value為websocket Session。節(jié)點(diǎn)1保存了用戶A的websocket session，節(jié)點(diǎn)2保存了用戶A、B的websocket session。

消息生產(chǎn)者發(fā)布消息的時(shí)候?yàn)閖son格式，如：[{"receive"="userid_a"，"msg"="您有1個(gè)未讀消息"},{"receive"="userid_b"，"msg"="您有3個(gè)未讀消息"}]，將消息發(fā)到redis的一個(gè)Channel，如showNewestMsg。

websocketServer中訂閱redis的channel=showNewestMsg，收到消息后根據(jù)消息中receive沖map中找到對(duì)應(yīng)的websocket session，發(fā)消息給客戶端。

核心代碼：

1.該項(xiàng)目為springboot項(xiàng)目，先引入jar包，由于是從實(shí)際項(xiàng)目中抽出來(lái)寫的記錄，可能還缺jar請(qǐng)自行導(dǎo)入。

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

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

<!-- redis -->
<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

<!-- 工具類 -->
<dependency>
	<groupId>cn.hutool</groupId>
	<artifactId>hutool-all</artifactId>
	<version>5.3.6</version>
</dependency>

<dependency>
	<groupId>net.sf.json-lib</groupId>
	<artifactId>json-lib</artifactId>
	<version>2.4</version>
	<classifier>jdk15</classifier>
</dependency>

2.websocket配置

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

/**
 * spring websocket組件初始化
 * @author csf
 * 
 */
//war包啟動(dòng)tomcat7及以下版本要關(guān)閉@Configuration注解,否則將無(wú)法啟動(dòng)websocket服務(wù)
@Configuration
public class WebSocketConfig
{
    @Bean
    public ServerEndpointExporter serverEndpointExporter()
    {
        return new ServerEndpointExporter();
    }
}

注意：war包啟動(dòng)tomcat7及以下版本要關(guān)閉@Configuration注解,否則將無(wú)法啟動(dòng)websocket服務(wù)。

3.websocket服務(wù)端實(shí)現(xiàn)

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import com.kingengine.plug.service.MessageService;

import cn.hutool.core.util.StrUtil;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;

/**
 * WebSocket服務(wù)類
 * @author csf
 * @date 2020年8月10日
 */
@ServerEndpoint("/websocket/{custId}")
@Component
public class WebSocketServer
{
    @Resource
    private MessageService messageService;
    
    Logger log = LoggerFactory.getLogger(this.getClass());
    
    // 當(dāng)前在線連接數(shù)
    private static int onlineCount = 0;
    
    // 存放每個(gè)用戶對(duì)應(yīng)的WebSocket連接對(duì)象,key為custId_HHmmss，確保一個(gè)登錄用戶只建立一個(gè)連接
    private static Map<String, Session> webSocketSessionMap = new ConcurrentHashMap<String, Session>();
    
    // 與某個(gè)客戶端的連接會(huì)話，需要通過(guò)它來(lái)給客戶端發(fā)送數(shù)據(jù)
    private Session session;
    
    // 接收用戶id
    private String custId = "";
    
    private static WebSocketServer webSocketServer;
    
    // 通過(guò)@PostConstruct實(shí)現(xiàn)初始化bean之前進(jìn)行的操作
    @PostConstruct
    public void init()
    {
        // 初使化時(shí)將已靜態(tài)化的webSocketServer實(shí)例化
        webSocketServer = this;
        webSocketServer.messageService = this.messageService;
    }
    
    /**
     * 連接建立成功調(diào)用的方法
     * @param session 連接會(huì)話，由框架創(chuàng)建
     * @param custId 用戶id， 為處理用戶多點(diǎn)登錄都能收到消息，需傳該格式custId_HHmmss
     * @author csf
     * @date 2020年8月10日
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("custId") String custId)
    {
        if (!webSocketSessionMap.containsKey(custId))
        {
            this.session = session;
            webSocketSessionMap.put(custId, session);
            addOnlineCount(); // 在線數(shù)加1
            log.info("有新連接[{}]接入,當(dāng)前websocket連接數(shù)為:{}", custId, getOnlineCount());
        }
        
        this.custId = custId;
        try
        {
            // 第一次建立連接，推送消息給客戶端，只會(huì)執(zhí)行一次。后續(xù)的新消息由com.kingengine.plug.redis.RedisReceiver接收到redis訂閱消息推送
            // 獲取未讀消息數(shù)
            // 由于前端傳進(jìn)來(lái)的custId是有時(shí)間后綴的，查詢時(shí)需要去掉后綴。
            String qryCustId = custId.split("_")[0];
            JSONObject unreadMsg = webSocketServer.messageService.getUnreadCount(qryCustId);
            
            // 獲取最新消息
            /*  JSONObject newMsg = webSocketServer.messageService.getNewestMsg(qryCustId);
            // 發(fā)送消息
            JSONArray msgArr = new JSONArray();
            if (newMsg!=null)
            {
                msgArr.add(newMsg);
            }*/
            JSONArray msgArr = new JSONArray();
            msgArr.add(unreadMsg);
            sendMessage(custId, msgArr.toString());
        }
        catch (Exception e)
        {
            log.error("客戶端連接websocket服務(wù)異常");
            e.printStackTrace();
        }
    }
    
    /**
     * 連接關(guān)閉調(diào)用的方法
     */
    @OnClose
    public void onClose(@PathParam("custId") String sessionKey)
    {
        if (webSocketSessionMap.containsKey(sessionKey))
        {
            try
            {
                webSocketSessionMap.get(sessionKey).close();
                webSocketSessionMap.remove(sessionKey);
            }
            catch (IOException e)
            {
                log.error("連接[{}]關(guān)閉失敗。", sessionKey);
                e.printStackTrace();
            }
            subOnlineCount();
            log.info("連接[{}]關(guān)閉，當(dāng)前websocket連接數(shù)：{}", sessionKey, onlineCount);
        }
    }
    
    /**
     * 接收客戶端發(fā)送的消息
     * @param message 客戶端發(fā)送過(guò)來(lái)的消息
     * @param session websocket會(huì)話
     */
    @OnMessage
    public void onMessage(String message, Session session)
    {
        log.info("收到來(lái)自客戶端" + custId + "的信息:" + message);
    }
    
    /**
     * 連接錯(cuò)誤時(shí)觸發(fā)
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error)
    {
        try
        {
            session.close();
        }
        catch (IOException e)
        {
            log.error("發(fā)生錯(cuò)誤，連接[{}]關(guān)閉失敗。");
            e.printStackTrace();
        }
        // log.error("websocket發(fā)生錯(cuò)誤");
        // error.printStackTrace();
    }
    
    /**
     * 給指定的客戶端推送消息，可單發(fā)和群發(fā)
     * @param sessionKeys 發(fā)送消息給目標(biāo)客戶端sessionKey，多個(gè)逗號(hào)“,”隔開1234,2345...
     * @param message
     * @throws IOException
     * @author csf
     * @date 2020年8月11日
     */
    public void sendMessage(String sessionKeys, String message)
    {
        if (StrUtil.isNotBlank(sessionKeys))
        {
            String[] sessionKeyArr = sessionKeys.split(",");
            for (String key : sessionKeyArr)
            {
                try
                {
                    // 可能存在一個(gè)賬號(hào)多點(diǎn)登錄
                    List<Session> sessionList = getLikeByMap(webSocketSessionMap, key);
                    for (Session session : sessionList)
                    {
                        session.getBasicRemote().sendText(message);
                    }
                }
                catch (IOException e)
                {
                    e.printStackTrace();
                    continue;// 某個(gè)客戶端發(fā)送異常，不影響其他客戶端發(fā)送
                }
            }
        }
        else
        {
            log.info("sessionKeys為空，沒(méi)有目標(biāo)客戶端");
        }
    }
    
    /**
     * 給當(dāng)前客戶端推送消息，首次建立連接時(shí)調(diào)用
     */
    public void sendMessage(String message)
        throws IOException
    {
        this.session.getBasicRemote().sendText(message);
    }
    
    /**
     * 檢查webSocket連接是否在線
     * @param sesstionKey webSocketMap中維護(hù)的key
     * @return 是否在線
     */
    public static boolean checkOnline(String sesstionKey)
    {
        if (webSocketSessionMap.containsKey(sesstionKey))
        {
            return true;
        }
        else
        {
            return false;
        }
    }
    
    /**
     * 獲取包含key的所有map值
     * @param map
     * @param keyLike
     * @return
     * @author csf
     * @date 2020年8月13日
     */
    private List<Session> getLikeByMap(Map<String, Session> map, String keyLike)
    {
        List<Session> list = new ArrayList<>();
        for (String key : map.keySet())
        {
            if (key.contains(keyLike))
            {
                list.add(map.get(key));
            }
        }
        return list;
    }
    
    public static synchronized int getOnlineCount()
    {
        return onlineCount;
    }
    
    public static synchronized void addOnlineCount()
    {
        WebSocketServer.onlineCount++;
    }
    
    public static synchronized void subOnlineCount()
    {
        WebSocketServer.onlineCount--;
    }
}

4.redis消息訂閱配置

import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

@Configuration
@EnableCaching
public class RedisCacheConfig
{
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter)
    {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        // 可以添加多個(gè) messageListener，配置不同的交換機(jī)
        container.addMessageListener(listenerAdapter, new PatternTopic("showNewestMsg"));// 訂閱最新消息頻道
        return container;
    }
    
    @Bean
    MessageListenerAdapter listenerAdapter(RedisReceiver receiver)
    {
        // 消息監(jiān)聽適配器
        return new MessageListenerAdapter(receiver, "onMessage");
    }
    
    @Bean
    StringRedisTemplate template(RedisConnectionFactory connectionFactory)
    {
        return new StringRedisTemplate(connectionFactory);
    }
}

5.redis配置，直接放在springboot項(xiàng)目application.properties或application.yml中

# 數(shù)據(jù)庫(kù)索引（默認(rèn)為0）
spring.redis.database=0  
spring.redis.host=192.168.1.100
spring.redis.port=6379
spring.redis.password=123456
# 連接池最大連接數(shù)（使用負(fù)值表示沒(méi)有限制）
spring.redis.pool.max-active=8  
# 連接池最大阻塞等待時(shí)間（使用負(fù)值表示沒(méi)有限制）
spring.redis.pool.max-wait=-1  
# 連接池中的最大空閑連接
spring.redis.pool.max-idle=8  
# 連接池中的最小空閑連接
spring.redis.pool.min-idle=0  
# 連接超時(shí)時(shí)間（毫秒）
spring.redis.timeout=5000

6.接收消息生產(chǎn)者發(fā)布的消息，推送給對(duì)應(yīng)的客戶端

import java.io.UnsupportedEncodingException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;

import com.kingengine.plug.websocket.WebSocketServer;

import cn.hutool.core.codec.Base64;
import cn.hutool.core.util.StrUtil;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;

/**
 * 消息監(jiān)聽對(duì)象，接收訂閱消息
 * @author csf
 * @date 2020年8月13日
 */
@Component
public class RedisReceiver implements MessageListener
{
    Logger log = LoggerFactory.getLogger(this.getClass());
    
    @Autowired
    WebSocketServer webSocketServer;
    
    /**
     * 處理接收到的訂閱消息
     */
    @Override
    public void onMessage(Message message, byte[] pattern)
    {
        String channel = new String(message.getChannel());// 訂閱的頻道名稱
        String msg = "";
        try
        {
            msg = new String(message.getBody(), "GBK");//注意與發(fā)布消息編碼一致，否則會(huì)亂碼
            if (StrUtil.isNotBlank(msg)){
                if ("showNewestMsg".endsWith(channel))// 最新消息
                {
                    JSONObject json = JSONObject.fromObject(msg);
                    webSocketServer.sendMessage(json.get("receive"),json.get("msg"));
                }else{
                    //TODO 其他訂閱的消息處理
                }
               
            }else{
                log.info("消息內(nèi)容為空，不處理。");
            }
        }
        catch (Exception e)
        {
            log.error("處理消息異常："+e.toString())
            e.printStackTrace();
        }
    }
}

7.消息發(fā)布測(cè)試

import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.Map;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import net.sf.json.JSONObject;

@RequestMapping("redis")
@RestController
public class RedisTestController
{
    @Autowired
    StringRedisTemplate template;
    
    /**
     * 發(fā)布消息測(cè)試
     *@param userid
     * @param msg
     * @return
     */
    @PostMapping("sendMessage")
    public String sendMessage(String userid,String msg)
    {
        try
        {
            String newMessge=new String(msg.getBytes("GBK"),"GBK");
            Map<String,String> map = new HashMap<String, String>();
            map.put("receive", userid);
            map.put("msg", newMessge);
            template.convertAndSend("showNewestMsg",         
          JSONObject.fromObject(map).toString());
        }
        catch (UnsupportedEncodingException e)
        {
            e.printStackTrace();
        }
        return "消息發(fā)布成功！";
    }
}

8.客戶端代碼

<!DOCTYPE html>

<html>

<head>
    <title>WebSocket測(cè)試</title>
    <meta http-equiv="Content-Type" content="text/html; charset=utf-8"/>
</head>

<body>
<div>
    來(lái)自服務(wù)端消息：
    <p id="message"></p>
</div>
</body>


<script src="http://apps.bdimg.com/libs/jquery/1.6.4/jquery.min.js"></script>

<script>
    let webSocketClient;
    if (window.WebSocket)
    {
       let custid="132456_" + Math.random();//該參數(shù)會(huì)作為websocketServer中存儲(chǔ)session的key，要保證唯一。
        webSocketClient = new WebSocket("ws://127.0.0.1:8082/bootapp/websocket/" + custid);

        //連通之后的回調(diào)事件
        webSocketClient.onopen = function () {
            webSocketClient.send("這里是地球，收到請(qǐng)回答。。。");
            //	webSocket.send('{"type":"1","data":"121"}');
        };

        //接收后臺(tái)服務(wù)端的消息
        webSocketClient.onmessage = function (evt) {
            console.log("數(shù)據(jù)已接收:" + evt.data);
            showMessage("未讀消息：" + evt.data);
        };

        //連接關(guān)閉的回調(diào)事件
        webSocketClient.onclose = function () {
            alert("連接已關(guān)閉...");
        };

    }else{
        alert("瀏覽器不支持websocket");
    }

    function showMessage(message) {
        $("#message").html(message);
    }
</script>
</html>

核心代碼至此。

到此這篇關(guān)于redis做websocket分布式消息推送服務(wù)的實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)redis websocket分布式消息推送服務(wù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: