首先：因?yàn)楣ぷ餍枰枰獙?duì)接socket.io框架對(duì)接，所以目前只能使用netty-socketio。websocket是不支持對(duì)接socket.io框架的。

netty-socketio顧名思義他是一個(gè)底層基于netty’實(shí)現(xiàn)的socket。

在springboot項(xiàng)目中的集成，請(qǐng)看下面的代碼

maven依賴(lài)

<dependency> <groupId>com.corundumstudio.socketio</groupId> <artifactId>netty-socketio</artifactId> <version>1.7.11</version></dependency>

下面就是代碼了

首先是配置參數(shù)

#socketio配置socketio: host: localhost port: 9099 # 設(shè)置最大每幀處理數(shù)據(jù)的長(zhǎng)度，防止他人利用大數(shù)據(jù)來(lái)攻擊服務(wù)器 maxFramePayloadLength: 1048576 # 設(shè)置http交互最大內(nèi)容長(zhǎng)度 maxHttpContentLength: 1048576 # socket連接數(shù)大小（如只監(jiān)聽(tīng)一個(gè)端口boss線(xiàn)程組為1即可） bossCount: 1 workCount: 100 allowCustomRequests: true # 協(xié)議升級(jí)超時(shí)時(shí)間（毫秒），默認(rèn)10秒。HTTP握手升級(jí)為ws協(xié)議超時(shí)時(shí)間 upgradeTimeout: 1000000 # Ping消息超時(shí)時(shí)間（毫秒），默認(rèn)60秒，這個(gè)時(shí)間間隔內(nèi)沒(méi)有接收到心跳消息就會(huì)發(fā)送超時(shí)事件 pingTimeout: 6000000 # Ping消息間隔（毫秒），默認(rèn)25秒。客戶(hù)端向服務(wù)器發(fā)送一條心跳消息間隔 pingInterval: 25000

上面的注釋寫(xiě)的很清楚。下面是config代碼

import com.corundumstudio.socketio.Configuration;import com.corundumstudio.socketio.SocketConfig;import com.corundumstudio.socketio.SocketIOServer;import org.springframework.beans.factory.InitializingBean;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.beans.factory.annotation.Value;import org.springframework.stereotype.Component;import javax.annotation.Resource;/** * kcm */@Componentpublic class PushServer implements InitializingBean { @Autowired private EventListenner eventListenner; @Value('${socketio.port}') private int serverPort; @Value('${socketio.host}') private String serverHost; @Value('${socketio.bossCount}') private int bossCount; @Value('${socketio.workCount}') private int workCount; @Value('${socketio.allowCustomRequests}') private boolean allowCustomRequests; @Value('${socketio.upgradeTimeout}') private int upgradeTimeout; @Value('${socketio.pingTimeout}') private int pingTimeout; @Value('${socketio.pingInterval}') private int pingInterval; @Override public void afterPropertiesSet() throws Exception { Configuration config = new Configuration(); config.setPort(serverPort); config.setHostname(serverHost); config.setBossThreads(bossCount); config.setWorkerThreads(workCount); config.setAllowCustomRequests(allowCustomRequests); config.setUpgradeTimeout(upgradeTimeout); config.setPingTimeout(pingTimeout); config.setPingInterval(pingInterval); SocketConfig socketConfig = new SocketConfig(); socketConfig.setReuseAddress(true); socketConfig.setTcpNoDelay(true); socketConfig.setSoLinger(0); config.setSocketConfig(socketConfig); SocketIOServer server = new SocketIOServer(config); server.addListeners(eventListenner); server.start(); System.out.println('啟動(dòng)正常'); }}

在就是監(jiān)聽(tīng)代碼

import com.corundumstudio.socketio.AckRequest;import com.corundumstudio.socketio.SocketIOClient;import com.corundumstudio.socketio.annotation.OnConnect;import com.corundumstudio.socketio.annotation.OnDisconnect;import com.corundumstudio.socketio.annotation.OnEvent;import org.apache.commons.lang3.StringUtils;import org.bangying.auth.JwtSupport;import org.springframework.stereotype.Component;import javax.annotation.Resource;import java.util.UUID;@Componentpublic class EventListenner { @Resource private ClientCache clientCache; @Resource private JwtSupport jwtSupport; /** * 客戶(hù)端連接 * * @param client */ @OnConnect public void onConnect(SocketIOClient client) { String userId = client.getHandshakeData().getSingleUrlParam('userId');// userId = jwtSupport.getApplicationUser().getId().toString();// userId = '8'; UUID sessionId = client.getSessionId(); clientCache.saveClient(userId, sessionId, client); System.out.println('建立連接'); } /** * 客戶(hù)端斷開(kāi) * * @param client */ @OnDisconnect public void onDisconnect(SocketIOClient client) { String userId = client.getHandshakeData().getSingleUrlParam('userId'); if (StringUtils.isNotBlank(userId)) { clientCache.deleteSessionClient(userId, client.getSessionId()); System.out.println('關(guān)閉連接'); } } //消息接收入口，當(dāng)接收到消息后，查找發(fā)送目標(biāo)客戶(hù)端，并且向該客戶(hù)端發(fā)送消息，且給自己發(fā)送消息 // 暫未使用 @OnEvent('messageevent') public void onEvent(SocketIOClient client, AckRequest request) { }}

本地緩存信息

import com.corundumstudio.socketio.SocketIOClient;import org.apache.commons.lang3.StringUtils;import org.springframework.stereotype.Component;import java.util.HashMap;import java.util.Map;import java.util.UUID;import java.util.concurrent.ConcurrentHashMap;/** * kcm */@Componentpublic class ClientCache { //本地緩存 private static Map<String, HashMap<UUID, SocketIOClient>> concurrentHashMap=new ConcurrentHashMap<>(); /** * 存入本地緩存 * @param userId 用戶(hù)ID * @param sessionId 頁(yè)面sessionID * @param socketIOClient 頁(yè)面對(duì)應(yīng)的通道連接信息 */ public void saveClient(String userId, UUID sessionId,SocketIOClient socketIOClient){ if(StringUtils.isNotBlank(userId)){ HashMap<UUID, SocketIOClient> sessionIdClientCache=concurrentHashMap.get(userId); if(sessionIdClientCache==null){sessionIdClientCache = new HashMap<>(); } sessionIdClientCache.put(sessionId,socketIOClient); concurrentHashMap.put(userId,sessionIdClientCache); } } /** * 根據(jù)用戶(hù)ID獲取所有通道信息 * @param userId * @return */ public HashMap<UUID, SocketIOClient> getUserClient(String userId){ return concurrentHashMap.get(userId); } /** * 根據(jù)用戶(hù)ID及頁(yè)面sessionID刪除頁(yè)面鏈接信息 * @param userId * @param sessionId */ public void deleteSessionClient(String userId,UUID sessionId){ concurrentHashMap.get(userId).remove(sessionId); }}

下面是存儲(chǔ)客戶(hù)端連接信息

import com.corundumstudio.socketio.SocketIOClient;import org.apache.commons.lang3.StringUtils;import org.springframework.stereotype.Component;import java.util.HashMap;import java.util.Map;import java.util.UUID;import java.util.concurrent.ConcurrentHashMap;/** * kcm */@Componentpublic class ClientCache { //本地緩存 private static Map<String, HashMap<UUID, SocketIOClient>> concurrentHashMap=new ConcurrentHashMap<>(); /** * 存入本地緩存 * @param userId 用戶(hù)ID * @param sessionId 頁(yè)面sessionID * @param socketIOClient 頁(yè)面對(duì)應(yīng)的通道連接信息 */ public void saveClient(String userId, UUID sessionId,SocketIOClient socketIOClient){ if(StringUtils.isNotBlank(userId)){ HashMap<UUID, SocketIOClient> sessionIdClientCache=concurrentHashMap.get(userId); if(sessionIdClientCache==null){sessionIdClientCache = new HashMap<>(); } sessionIdClientCache.put(sessionId,socketIOClient); concurrentHashMap.put(userId,sessionIdClientCache); } } /** * 根據(jù)用戶(hù)ID獲取所有通道信息 * @param userId * @return */ public HashMap<UUID, SocketIOClient> getUserClient(String userId){ return concurrentHashMap.get(userId); } /** * 根據(jù)用戶(hù)ID及頁(yè)面sessionID刪除頁(yè)面鏈接信息 * @param userId * @param sessionId */ public void deleteSessionClient(String userId,UUID sessionId){ concurrentHashMap.get(userId).remove(sessionId); }}

控制層推送方法

@RestController@RequestMapping('/push')public class PushController { @Resource private ClientCache clientCache; @Autowired private JwtSupport jwtSupport; @GetMapping('/message') public String pushTuUser(@Param('id') String id){ Integer userId = jwtSupport.getApplicationUser().getId(); HashMap<UUID, SocketIOClient> userClient = clientCache.getUserClient(String.valueOf(userId)); userClient.forEach((uuid, socketIOClient) -> { //向客戶(hù)端推送消息 socketIOClient.sendEvent('chatevent','服務(wù)端推送消息'); }); return 'success'; }}

到此這篇關(guān)于SpringBoot+netty-socketio實(shí)現(xiàn)服務(wù)器端消息推送的文章就介紹到這了,更多相關(guān)SpringBoot netty-socketio服務(wù)器端推送內(nèi)容請(qǐng)搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！