返利APP排行榜數據實時更新：基于 WebSocket 與 Redis 的高并發數據推送技術

大家好，我是阿可，微賺淘客系統及省賺客APP創始人，是個冬天不穿秋褲，天冷也要風度的程序猿！

在返利APP運營中，用戶對排行榜數據的實時性要求極高——無論是“今日收益TOP10”還是“熱門商品銷量榜”，延遲超過1秒就可能影響用戶體驗。傳統的輪詢方案不僅會造成服務器資源浪費，還無法滿足高并發場景下的實時推送需求。本文將基于WebSocket的全雙工通信能力與Redis的高性能緩存特性，提供一套可落地的高并發數據推送方案，全程附帶完整Java代碼實現。

一、技術選型核心依據

1.1 WebSocket 替代輪詢的必然性

傳統輪詢通過客戶端定時發送HTTP請求獲取數據，存在兩大問題：一是空請求占比高（當數據無更新時，請求仍會占用帶寬與服務器資源）；二是實時性差（輪詢間隔決定了數據延遲）。而WebSocket通過一次TCP握手建立持久連接，服務器可主動向客戶端推送數據，推送延遲可控制在100ms內，且單個連接的資源占用僅為HTTP輪詢的1/10。

1.2 Redis 高并發支撐能力

排行榜數據需頻繁更新（如用戶返利金額變動）與查詢，Redis的Sorted Set（ZSet） 結構天然適合排序場景，支持O(logN)復雜度的插入、刪除與排序操作。同時，Redis的發布訂閱（Pub/Sub）功能可實現數據更新后的實時通知，配合WebSocket完成端到端推送。

二、核心技術實現（Java代碼）

2.1 Redis 排行榜操作封裝（cn.juwatech.redis.RankRedisService）

package cn.juwatech.redis;import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Service;import javax.annotation.Resource;
import java.util.Set;@Service
public class RankRedisService {// 返利排行榜Redis Key前綴private static final String RANK_KEY_PREFIX = "rebate:rank:";// 排行榜數據更新訂閱頻道public static final String RANK_UPDATE_CHANNEL = "rebate:rank:update";@Resourceprivate RedisTemplate<String, Object> redisTemplate;@Resourceprivate ZSetOperations<String, Object> zSetOperations;/*** 更新用戶排行榜數據* @param rankType 排行榜類型（如"daily_income"：日收益，"month_sales"：月銷量）* @param userId 用戶ID* @param score 排序分數（如收益金額、銷量）*/public void updateRankScore(String rankType, String userId, double score) {String rankKey = getRankKey(rankType);// 存儲用戶ID與對應分數到ZSet，分數相同按插入順序排序zSetOperations.add(rankKey, userId, score);// 限制排行榜長度（僅保留前100名，避免數據膨脹）zSetOperations.removeRange(rankKey, 0, -101);// 發布數據更新通知redisTemplate.convertAndSend(RANK_UPDATE_CHANNEL, rankType);}/*** 獲取排行榜前N名數據* @param rankType 排行榜類型* @param topSize 前N名* @return 有序集合（score從高到低）*/public Set<ZSetOperations.TypedTuple<Object>> getTopRank(String rankType, int topSize) {String rankKey = getRankKey(rankType);// ZSet默認按score升序，reverseRange按降序獲取前topSize條return zSetOperations.reverseRangeWithScores(rankKey, 0, topSize - 1);}/*** 構建完整Redis Key*/private String getRankKey(String rankType) {return RANK_KEY_PREFIX + rankType;}
}

2.2 WebSocket 連接管理與數據推送（cn.juwatech.websocket.RankWebSocketServer）

基于Spring WebSocket實現，支持用戶建立連接時訂閱指定排行榜，數據更新時定向推送。

package cn.juwatech.websocket;import cn.juwatech.redis.RankRedisService;
import cn.juwatech.redis.RankRedisService;
import com.alibaba.fastjson.JSONObject;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;import javax.annotation.Resource;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;@Component
public class RankWebSocketServer extends TextWebSocketHandler implements MessageListener {// 存儲用戶會話與訂閱的排行榜類型（線程安全）private final Map<WebSocketSession, String> sessionRankMap = new ConcurrentHashMap<>();// 排行榜Redis服務@Resourceprivate RankRedisService rankRedisService;/*** 客戶端建立WebSocket連接時觸發*/@Overridepublic void afterConnectionEstablished(WebSocketSession session) throws Exception {// 從連接參數中獲取用戶訂閱的排行榜類型（如?rankType=daily_income）String rankType = session.getUri().getQuery().split("=")[1];sessionRankMap.put(session, rankType);// 首次連接時推送當前排行榜數據pushRankData(session, rankType);}/*** 接收Redis數據更新通知，推送排行榜數據*/@Overridepublic void onMessage(Message message, byte[] pattern) {// 解析Redis訂閱消息（排行榜類型）String rankType = new String(message.getBody());// 遍歷訂閱該排行榜的所有會話，推送數據sessionRankMap.forEach((session, type) -> {if (rankType.equals(type) && session.isOpen()) {pushRankData(session, rankType);}});}/*** 推送排行榜數據到客戶端*/private void pushRankData(WebSocketSession session, String rankType) {try {// 從Redis獲取前10名排行榜數據var topRank = rankRedisService.getTopRank(rankType, 10);// 構建JSON格式響應（包含排行榜類型、數據列表）JSONObject response = new JSONObject();response.put("rankType", rankType);response.put("data", topRank.stream().map(tuple -> {JSONObject item = new JSONObject();item.put("userId", tuple.getValue());item.put("score", tuple.getScore());return item;}).toList());// 發送文本消息session.sendMessage(new TextMessage(response.toJSONString()));} catch (IOException e) {// 記錄推送失敗日志（實際項目中需結合監控告警）e.printStackTrace();}}/*** 客戶端關閉連接時清理會話*/@Overridepublic void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {sessionRankMap.remove(session);}
}

2.3 WebSocket 配置類（cn.juwatech.config.WebSocketConfig）

package cn.juwatech.config;import cn.juwatech.redis.RankRedisService;
import cn.juwatech.websocket.RankWebSocketServer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;import javax.annotation.Resource;@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {@Resourceprivate RankWebSocketServer rankWebSocketServer;/*** 注冊WebSocket處理器，配置訪問路徑*/@Overridepublic void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {// 允許跨域訪問，配置WebSocket訪問路徑為/ws/rebate-rankregistry.addHandler(rankWebSocketServer, "/ws/rebate-rank").setAllowedOrigins("*");}/*** 配置Redis消息監聽器，訂閱排行榜更新頻道*/@Beanpublic RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory,MessageListenerAdapter listenerAdapter) {RedisMessageListenerContainer container = new RedisMessageListenerContainer();container.setConnectionFactory(connectionFactory);// 訂閱排行榜更新頻道container.addMessageListener(listenerAdapter, new PatternTopic(RankRedisService.RANK_UPDATE_CHANNEL));return container;}/*** 綁定Redis消息監聽器與WebSocket處理器*/@Beanpublic MessageListenerAdapter listenerAdapter(RankWebSocketServer rankWebSocketServer) {return new MessageListenerAdapter(rankWebSocketServer);}
}

三、高并發場景優化策略

3.1 連接數承載優化

單個WebSocket服務器的并發連接數受限于操作系統文件句柄數（默認1024），生產環境需通過以下配置提升承載能力：

1. 調整Linux系統參數：echo "net.core.somaxconn=65535" >> /etc/sysctl.conf（最大監聽隊列數）；
2. 采用Nginx反向代理實現WebSocket集群負載均衡，配置示例：

http {upstream websocket_servers {server 192.168.1.100:8080 weight=1;server 192.168.1.101:8080 weight=1;}server {listen 80;location /ws/rebate-rank {# 啟用WebSocket代理proxy_pass http://websocket_servers;proxy_http_version 1.1;proxy_set_header Upgrade $http_upgrade;proxy_set_header Connection "upgrade";proxy_set_header Host $host;}}
}

3.2 Redis 性能優化

1. 開啟Redis持久化（AOF+RDB），避免排行榜數據丟失；
2. 對排行榜Key設置過期時間（如日收益榜24小時過期），減少內存占用；
3. 使用Redis集群（3主3從），提升讀吞吐量與可用性。

四、客戶端接入示例（JavaScript）

// 建立WebSocket連接（根據環境替換域名）
const rankType = "daily_income"; // 訂閱日收益排行榜
const ws = new WebSocket(`ws://your-domain.com/ws/rebate-rank?rankType=${rankType}`);// 接收服務器推送的排行榜數據
ws.onmessage = function(event) {const rankData = JSON.parse(event.data);console.log("實時排行榜更新：", rankData);// 渲染排行榜到頁面（示例：更新表格）renderRankTable(rankData.data);
};// 連接關閉處理
ws.onclose = function() {console.log("WebSocket連接關閉，嘗試重連...");// 重連邏輯（避免頻繁重連，添加延遲）setTimeout(() => window.location.reload(), 3000);
};// 渲染排行榜表格
function renderRankTable(data) {const table = document.getElementById("rankTable").getElementsByTagName("tbody")[0];table.innerHTML = "";data.forEach((item, index) => {const row = table.insertRow();row.insertCell(0).textContent = index + 1; // 排名row.insertCell(1).textContent = item.userId; // 用戶IDrow.insertCell(2).textContent = (item.score).toFixed(2); // 收益金額（保留2位小數）});
}

本文著作權歸聚娃科技省賺客app開發者團隊，轉載請注明出處！