SpringBoot+netty-socketio實現服務器端消息推送

Posted on 2021-03-17 by WalkonNet

首先：因為工作需要，需要對接socket.io框架對接，所以目前隻能使用netty-socketio。websocket是不支持對接socket.io框架的。

netty-socketio顧名思義他是一個底層基於netty’實現的socket。

在springboot項目中的集成，請看下面的代碼

maven依賴

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

下面就是代碼瞭

首先是配置參數

#socketio配置
socketio:
 host: localhost
 port: 9099
 # 設置最大每幀處理數據的長度，防止他人利用大數據來攻擊服務器
 maxFramePayloadLength: 1048576
 # 設置http交互最大內容長度
 maxHttpContentLength: 1048576
 # socket連接數大小（如隻監聽一個端口boss線程組為1即可）
 bossCount: 1
 workCount: 100
 allowCustomRequests: true
 # 協議升級超時時間（毫秒），默認10秒。HTTP握手升級為ws協議超時時間
 upgradeTimeout: 1000000
 # Ping消息超時時間（毫秒），默認60秒，這個時間間隔內沒有接收到心跳消息就會發送超時事件
 pingTimeout: 6000000
 # Ping消息間隔（毫秒），默認25秒。客戶端向服務器發送一條心跳消息間隔
 pingInterval: 25000

上面的註釋寫的很清楚。下面是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
 */
@Component
public 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("啟動正常");
  }
}

在就是監聽代碼

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;

@Component
public class EventListenner {
  @Resource
  private ClientCache clientCache;

  @Resource
  private JwtSupport jwtSupport;

  /**
   * 客戶端連接
   *
   * @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("建立連接");
  }

  /**
   * 客戶端斷開
   *
   * @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("關閉連接");
    }
  }

  //消息接收入口，當接收到消息後，查找發送目標客戶端，並且向該客戶端發送消息，且給自己發送消息
  // 暫未使用
  @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
 */
@Component
public class ClientCache {

  //本地緩存
  private static Map<String, HashMap<UUID, SocketIOClient>> concurrentHashMap=new ConcurrentHashMap<>();
  /**
   * 存入本地緩存
   * @param userId 用戶ID
   * @param sessionId 頁面sessionID
   * @param socketIOClient 頁面對應的通道連接信息
   */
  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);
    }
  }
  /**
   * 根據用戶ID獲取所有通道信息
   * @param userId
   * @return
   */
  public HashMap<UUID, SocketIOClient> getUserClient(String userId){
    return concurrentHashMap.get(userId);
  }
  /**
   * 根據用戶ID及頁面sessionID刪除頁面鏈接信息
   * @param userId
   * @param sessionId
   */
  public void deleteSessionClient(String userId,UUID sessionId){
    concurrentHashMap.get(userId).remove(sessionId);
  }
}

下面是存儲客戶端連接信息

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
 */
@Component
public class ClientCache {

  //本地緩存
  private static Map<String, HashMap<UUID, SocketIOClient>> concurrentHashMap=new ConcurrentHashMap<>();
  /**
   * 存入本地緩存
   * @param userId 用戶ID
   * @param sessionId 頁面sessionID
   * @param socketIOClient 頁面對應的通道連接信息
   */
  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);
    }
  }
  /**
   * 根據用戶ID獲取所有通道信息
   * @param userId
   * @return
   */
  public HashMap<UUID, SocketIOClient> getUserClient(String userId){
    return concurrentHashMap.get(userId);
  }
  /**
   * 根據用戶ID及頁面sessionID刪除頁面鏈接信息
   * @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) -> {
      //向客戶端推送消息
      socketIOClient.sendEvent("chatevent","服務端推送消息");
    });
    return "success";
  }
}

到此這篇關於SpringBoot+netty-socketio實現服務器端消息推送的文章就介紹到這瞭,更多相關SpringBoot netty-socketio服務器端推送內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

SpringBoot+netty-socketio實現服務器端消息推送

推薦閱讀：

發佈留言取消回覆

近期文章

推薦閱讀：

發佈留言 取消回覆

近期文章

標籤

發佈留言取消回覆