前言

WebSocket是一種在網(wǎng)絡(luò)應(yīng)用程序中，使客戶度端和服務(wù)器之間可以進(jìn)行雙向通信的協(xié)議镜雨。它允許數(shù)據(jù)可以在建立連接后進(jìn)行實(shí)時交換，而不必依賴傳統(tǒng)的HTTP請求-響應(yīng)模式儿捧。WebSocket使得客戶端和服務(wù)器之間的數(shù)據(jù)交換變得更加簡單荚坞，允許服務(wù)端主動向客戶端推送數(shù)據(jù)。在WebSocket API中菲盾，瀏覽器和服務(wù)器只需要完成一次握手颓影，兩者之間就直接可以創(chuàng)建持久性的連接，并進(jìn)行雙向數(shù)據(jù)傳輸懒鉴。

方案一：廣播方案

image.png

• Step1: 客戶端連接到某個Websocket Server诡挂，在該websocket Server中建立userid和session的綁定關(guān)系
• Step2: 其它服務(wù)或者客戶端通過MQ廣播消息所有Websocket Server（消息體中帶有userid）
• Step3: 所有Websocket Server 根據(jù)客戶端userid找到對應(yīng)session， 只有存在userid和session的綁定關(guān)系的Websocket Server才發(fā)送消息到客戶端

代碼演示

1.Websocket Server 建立userid和session的綁定關(guān)系

@ServerEndpoint("/websocket/{businessType}/{userId}")
@Component
public class WebSocketServer {
    /**
     * 若要實(shí)現(xiàn)服務(wù)端與單一客戶端通信的話临谱，可以使用Map來存放璃俗，其中Key可以為用戶標(biāo)識
     * 注意：allSession 只記錄當(dāng)前機(jī)器的 客戶端連接，不是所有session連接
     */

    public static ConcurrentHashMap<String, Session> allSession = new ConcurrentHashMap<>();

    @Resource
    private RedisService redisService;

    /**
     * 連接建立成功調(diào)用的方法
     *
     * @param session 可選的參數(shù)悉默。session為與某個客戶端的連接會話城豁，需要通過它來給客戶端發(fā)送數(shù)據(jù)
     */
    @OnOpen
    public void onOpen(@PathParam(value = "businessType") String businessType, @PathParam(value = "userId") String userId, Session session, EndpointConfig config) {
        if (StringUtils.isEmpty(userId)) {
            return;
        }
        /**
         * 加入到本地map
         */
        allSession.put(userId, session);
    }

    /**
     * 連接關(guān)閉調(diào)用的方法
     */
    @OnClose
    public void onClose(@PathParam(value = "userId") String userId, Session session) {
        if (StringUtils.isNotEmpty(userId)) {
            allSession.remove(userId);
        }
    }


    /**
     * 發(fā)生錯誤時調(diào)用
     *
     * @param
     * @param
     */
    @OnError
    public void onError(@PathParam(value = "userId") String userId, Session session, Throwable error) {
    }


    /**
     * 用戶id
     *
     * @param userId
     * @param message
     */
    public void sendMessageToOneUser(Integer userId, String message, String msgId) {
        if (userId == null) {
            return;
        }
        Session session = allSession.get(String.valueOf(userId));
        if (session != null) {
         //所有Websocket Server 根據(jù)客戶端userid找到對應(yīng)session， 只有存在userid和session的綁定關(guān)系的Websocket Server才發(fā)送消息到客戶端
          session.getAsyncRemote().sendText(message);
        } else {
            System.err.println("session為空");
            allSession.remove(userId + "");
        }
    }
}

2.所有Websocket Server 接收消息并處理

@Component
@RequiredArgsConstructor
public class CreateOrderConsumer implements BaseConsumer {


    private final WebSocketServer webSocketServer;


    @Override
    public Action handleMessage(Message message) {
        CreateOrderMessage createOrderMessage = JSON.parseObject(message.getBody(), LinkCreateOrderMessage.class);

        try {
           //業(yè)務(wù)校驗(yàn)省略...
           //調(diào)用WebSocketServer的sendMessageToOneUser方法抄课，里面根據(jù)客戶端userid找到對應(yīng)session唱星， 只有存在userid和session的綁定關(guān)系的Websocket Server才發(fā)送消息到客戶端
            webSocketServer.sendMessageToOneUser(createOrderMessage.getUserId(), JSON.toJSONString(linkActionRes),message.getMsgID());
        } catch (Exception e) {
            e.printStackTrace();
            return Action.ReconsumeLater;
        }
        return Action.CommitMessage;
    }
}    

方案二：目標(biāo)詢址方案（推薦）

image.png

Id標(biāo)識有兩種實(shí)現(xiàn)形式：

• 為唯一的服務(wù)名：每一個WebSocketServer生成唯一的服務(wù)名（serviceName="XXX-" + IdUtil.oneId()）并注冊到naocs服務(wù)組冊中心，uesrid與其綁定跟磨，服務(wù)適用方使用Feign 或其它RPC調(diào)用http://{serviceName}/xxx/xxx到指定WebSocketServer

• 為唯一的IP+端口:每一個WebSocketServer 獲取自己IP+端口间聊，uesrid與其綁定，服務(wù)調(diào)用方使用該IP+端口

代碼演示(唯一Id為唯一的服務(wù)名的形式)

1.綁定userid和服務(wù)名唯一Id的關(guān)系(以ApplicationName形式為例)

@SpringBootApplication
public class WsApplication  {

    public static void main(String[] args) {
        //動態(tài)服務(wù)名
        System.setProperty("myApplicationName", "WS-" + IdUtil.oneId());
        SpringApplication.run(WsApplication.class, args);
    }
}

spring:
  application:
    #隨機(jī)名字抵拘，做ws集群使用
    name: ${myApplicationName}

@ServerEndpoint("/websocket/{businessType}/{userId}")
@Component
public class WebSocketServer {
    /**
     * 若要實(shí)現(xiàn)服務(wù)端與單一客戶端通信的話哎榴，可以使用Map來存放，其中Key可以為用戶標(biāo)識
     * 注意：allSession 只記錄當(dāng)前機(jī)器的 客戶端連接，不是所有session連接
     */

    public static ConcurrentHashMap<String, Session> allSession = new ConcurrentHashMap<>();
    /**
     *
     */
    private String myApplicationName = System.getProperty("myApplicationName");
    @Resource
    private RedisService redisService;

    /**
     * 連接建立成功調(diào)用的方法
     * 關(guān)鍵代碼
     * @param session 可選的參數(shù)叹话。session為與某個客戶端的連接會話偷遗，需要通過它來給客戶端發(fā)送數(shù)據(jù)
     */
    @OnOpen
    public void onOpen(@PathParam(value = "businessType") String businessType, @PathParam(value = "userId") String userId, Session session, EndpointConfig config) {
        if (StringUtils.isEmpty(userId)) {
            return;
        }
        /**
         * 加入到本地map
         */
        allSession.put(userId, session);
        //綁定userid和服務(wù)名唯一Id的關(guān)系
        redisService.hset(WS_MAPPING, userId + "", myApplicationName);
    }

    /**
     * 連接關(guān)閉調(diào)用的方法
     */
    @OnClose
    public void onClose(@PathParam(value = "userId") String userId, Session session) {
        if (StringUtils.isNotEmpty(userId)) {
            allSession.remove(userId);
        }
    }

    /**
     * 發(fā)生錯誤時調(diào)用
     *
     * @param
     * @param
     */
    @OnError
    public void onError(@PathParam(value = "userId") String userId, Session session, Throwable error) {
    }

    /**
     * 用戶id
     *
     * @param userId
     * @param message
     */
    public void sendMessageToOneUser(Integer userId, String message) {
        if (userId == null) {
            return;
        }
        Session session = allSession.get(String.valueOf(userId));
        if (session != null) {
         //所有Websocket Server 根據(jù)客戶端userid找到對應(yīng)session， 只有存在userid和session的綁定關(guān)系的Websocket Server才發(fā)送消息到客戶端
          session.getAsyncRemote().sendText(message);
        } else {
            System.err.println("session為空");
            allSession.remove(userId + "");
        }
    }
}

2.Websocket Server提供的調(diào)用接口

@RestController
@RequestMapping("push")
public class  WebSocketPushController {


    @PostMapping("{userId}")
    public void pushMessage(@PathVariable Long userId, @RequestBody Object message) {
            webSocketServer.sendMessageToOneUser(userId, message);
    }

}

3.調(diào)用方通過nacos調(diào)用目標(biāo)Websocket Server

//業(yè)省略
MyApplicationName myApplicationName =  redisService.hget(WS_MAPPING, userId + "");

Feign：
http://${myApplicationName}/push/{userId}

參考

WebSocket集群解決方案泵喘，不用MQ