webflux的websocket连接与生命周期

于 2019-04-23 10:43:46 发布 2622 收藏 6
文章标签： webflux
版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://blog.csdn.net/sinat_39291367/article/details/89467555
版权

1、配置入口：

  1. import com.mti.handler.MessageHandler;
  2. import org.springframework.beans.factory.annotation.Autowired;
  3. import org.springframework.context.annotation.Bean;
  4. import org.springframework.context.annotation.Configuration;
  5. import org.springframework.core.Ordered;
  6. import org.springframework.web.reactive.HandlerMapping;
  7. import org.springframework.web.reactive.handler.SimpleUrlHandlerMapping;
  8. import org.springframework.web.reactive.socket.WebSocketHandler;
  9. import org.springframework.web.reactive.socket.server.support.WebSocketHandlerAdapter;
  10.  
  11. import java.util.HashMap;
  12. import java.util.Map;
  13. /**
  14.  * ThreadConfig class
  15.  *
  16.  * @author zhaoyj
  17.  * @date 2019/3/12
  18.  */
  19. @Configuration
  20. public class WebSocketConfiguration {
  21.     @Autowired
  22.     @Bean
  23.     public HandlerMapping webSocketMapping(final MessageHandler echoHandler) {
  24.         final Map<String, WebSocketHandler> map = new HashMap<>();
  25.         map.put("/echo", echoHandler);
  26.         final SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
  27.         mapping.setOrder(Ordered.HIGHEST_PRECEDENCE);
  28.         mapping.setUrlMap(map);
  29.         return mapping;
  30.     }
  31.  
  32.     @Bean
  33.     public WebSocketHandlerAdapter handlerAdapter() {
  34.         return new WebSocketHandlerAdapter();
  35.     }
  36. }

2、配置Handler

  1. import com.alibaba.fastjson.JSONObject;
  2. import com.google.protobuf.InvalidProtocolBufferException;
  3. import com.mti.configuration.Systemconfig;
  4. import com.mti.enums.ReferenceMsgType;
  5. import com.mti.exception.BusinessException;
  6. import com.mti.handler.up.StreamReferenceReq;
  7. import com.mti.proto.Linkproto;
  8. import com.mti.vo.Message;
  9. import com.mti.websocket.SocketSessionRegistry;
  10. import lombok.extern.slf4j.Slf4j;
  11. import org.springframework.beans.factory.annotation.Autowired;
  12. import org.springframework.core.io.buffer.DataBufferFactory;
  13. import org.springframework.core.task.TaskExecutor;
  14. import org.springframework.stereotype.Component;
  15. import org.springframework.web.reactive.socket.WebSocketHandler;
  16. import org.springframework.web.reactive.socket.WebSocketMessage;
  17. import org.springframework.web.reactive.socket.WebSocketSession;
  18. import reactor.core.publisher.Flux;
  19. import reactor.core.publisher.Mono;
  20.  
  21. import java.util.Optional;
  22. import java.util.concurrent.ScheduledThreadPoolExecutor;
  23. import java.util.concurrent.TimeUnit;
  24.  
  25. /**
  26.  * MessageHandler class
  27.  *
  28.  * @author zhaoyj
  29.  * @date 2019/3/12
  30.  */
  31. @Component
  32. @Slf4j
  33. public class MessageHandler implements WebSocketHandler {
  34.  
  35.     @Autowired
  36.     private SocketSessionRegistry sessionRegistry;
  37.     @Autowired
  38.     private ScheduledThreadPoolExecutor executor;
  39.     @Autowired
  40.     private Systemconfig systemconfig;
  41.  
  42.     @Autowired
  43.     DispatchFactory dispatchFactory;
  44.  
  45.     @Autowired
  46.     TaskExecutor taskExecutor;
  47.  
  48.     @Override
  49.     public Mono<Void> handle(WebSocketSession session) {
  50.  
  51.         return session.receive().doOnSubscribe(s -> {
  52.             log.info("发起连接:{}",s);
  53.             /**
  54.              * 你有10秒时间登陆，不登陆就关掉连接;并且不给任何错误信息
  55.              */
  56.             if(systemconfig.getLoginInterval() != 0){
  57.                 executor.schedule(() -> sessionRegistry.checkAndRemove(session),systemconfig.getLoginInterval(),TimeUnit.SECONDS);
  58.             }
  59.             if(systemconfig.getPingInterval() != 0){
  60.                 executor.schedule(() ->  session.send(Flux.just(session.pingMessage(DataBufferFactory::allocateBuffer))).toProcessor(),systemconfig.getPingInterval(), TimeUnit.SECONDS);
  61.             }
  62.         }).doOnTerminate(() -> {
  63.            sessionRegistry.unregisterSession(session);
  64.             StreamReferenceReq req = (StreamReferenceReq) dispatchFactory.getCommand(ReferenceMsgType.SEND_VALUE);
  65.             taskExecutor.execute(() -> Optional.ofNullable(req.removeSession(session)).ifPresent(list -> list.forEach(req::sendStopStreamConfig)));
  66.            log.info("doOnTerminate");
  67.         }).doOnComplete(() -> {
  68.             log.info("doOnComplete");
  69.         }).doOnCancel(() -> {
  70.             log.info("doOnCancel");
  71.         }).doOnNext(message -> {
  72.             if(message.getType().equals(WebSocketMessage.Type.BINARY)){
  73.                 log.info("收到二进制消息");
  74.                 Linkproto.LinkCmd linkCmd = null;
  75.                 try {
  76.                     linkCmd = Optional.ofNullable(Linkproto.LinkCmd.parseFrom(message.getPayload().asByteBuffer())).orElseThrow(() -> new BusinessException(500,"解析出错了"));
  77.                     BaseDispatch<Linkproto.LinkCmd> dispatch = dispatchFactory.getCommand(linkCmd.getTypeValue());
  78.                     log.info("处理session,{},消息实体,{},类型，{},dispatch:{}",session,linkCmd,linkCmd.getTypeValue(),dispatch);
  79.                     dispatch.excuted(session, linkCmd);
  80.                 } catch (InvalidProtocolBufferException e) {
  81.                     e.printStackTrace();
  82.                 }
  83.             }else if(message.getType().equals(WebSocketMessage.Type.TEXT)){
  84.                 String content = message.getPayloadAsText();
  85.                 log.info("收到文本消息:{}",content);
  86.                 Message msg = null;
  87.                 try{
  88.                     msg = JSONObject.parseObject(content, Message.class);
  89.                 }catch (Exception e){
  90.                     JSONObject obj = new JSONObject();
  91.                     obj.put("content","无法理解你发过来的消息内容，不予处理:"+content);
  92.                     obj.put("msgType",Linkproto.LinkCmdType.LINK_CMD_ZERO_VALUE);
  93.                     session.send(Flux.just(session.textMessage(obj.toJSONString()))).then().toProcessor();
  94.                     log.error("解析消息内容出错");
  95.                     return;
  96.                 }
  97.                 BaseDispatch<Linkproto.LinkCmd> dispatch = dispatchFactory.getCommand(msg.getMsgType());
  98.                 if(dispatch != null){
  99.                     dispatch.executeMsg(session, msg);
  100.                 }
  101.             }else if(message.getType().equals(WebSocketMessage.Type.PING)){
  102.                 session.send(Flux.just(session.pongMessage(s -> s.wrap(new byte[256]))));
  103.                 log.info("收到ping消息");
  104.             }else if(message.getType().equals(WebSocketMessage.Type.PONG)){
  105.                 log.info("收到pong消息");
  106.                 if(systemconfig.getPingInterval() != 0){
  107.                     executor.schedule(() ->  session.send(Flux.just(session.pingMessage(DataBufferFactory::allocateBuffer))).toProcessor(),systemconfig.getPingInterval(), TimeUnit.SECONDS);
  108.                 }
  109.             }
  110.         }).doOnError(e -> {
  111.             e.printStackTrace();
  112.             log.error("doOnError");
  113.         }).doOnRequest(r -> {
  114.             log.info("doOnRequest");
  115.         }).then();
  116.     }

这边显示的是整个从连接建立到连接断开的生命周期，可以区区分二进制消息还是文本消息，发送消息时，一定要加上toProcessor()，不然不会发送。

如果要发送消息到其它客户端，需要在后台将连接过来的session保存起来，根据用户名或者其它方式保存之后，获取到session进行发送：如下面这个SocketSessionRegistry类

  1. import com.mti.enums.SocketCloseStatus;
  2. import lombok.extern.slf4j.Slf4j;
  3. import org.springframework.stereotype.Service;
  4. import org.springframework.util.CollectionUtils;
  5. import org.springframework.util.StringUtils;
  6. import org.springframework.web.reactive.socket.WebSocketSession;
  7.  
  8. import java.util.*;
  9. import java.util.concurrent.ConcurrentHashMap;
  10. import java.util.concurrent.ConcurrentMap;
  11. import java.util.concurrent.CopyOnWriteArraySet;
  12. import java.util.concurrent.CountDownLatch;
  13.  
  14. /**
  15.  *用户session记录类
  16.  *
  17.  * @author zhaoyj
  18.  * @date 2019/3/12
  19.  */
  20. @Service
  21. @Slf4j
  22. public class SocketSessionRegistry {
  23.  
  24.     /**
  25.      * 这个集合存储session
  26.      */
  27.     private final ConcurrentMap<String, Set<String>> userSessionIds = new ConcurrentHashMap<>();
  28.  
  29.     private final ConcurrentMap<String, WebSocketSession> clientInfoSessionIds = new ConcurrentHashMap<>();
  30.  
  31.     private final ConcurrentMap<String,String> sessionIdUser = new ConcurrentHashMap<>();
  32.     private ConcurrentMap<String, CountDownLatch> cacheTimestamp = new ConcurrentHashMap<>();
  33.     private final Object lock = new Object();
  34.  
  35.  
  36.     /**
  37.      * 获取sessionId
  38.      *
  39.      * @param user
  40.      * @return
  41.      */
  42.     private Set<String> getSessionIds(String user) {
  43.         Set<String> set = this.userSessionIds.get(user);
  44.         return set != null ? set : Collections.emptySet();
  45.     }
  46.  
  47.     /**
  48.      * 获取用户session
  49.      * @param user
  50.      * @return
  51.      */
  52.     public Collection<WebSocketSession> getSessionByUser(String user){
  53.         Set<String> sessionIds = Optional.ofNullable(getSessionIds(user)).orElse(new CopyOnWriteArraySet<>());
  54.         List<WebSocketSession> sessions = new ArrayList<>();
  55.         for (String sessionId : sessionIds) {
  56.             sessions.add(clientInfoSessionIds.get(sessionId));
  57.         }
  58.         return sessions;
  59.     }
  60.  
  61.     /**
  62.      * 获取用户session
  63.      * @param users
  64.      * @return
  65.      */
  66.     public Collection<WebSocketSession> getSessionByUsers(Collection<String> users){
  67.         List<WebSocketSession> sessions = new ArrayList<>();
  68.         if(!CollectionUtils.isEmpty(users)){
  69.             for (String user : users) {
  70.                 sessions.addAll(getSessionByUser(user));
  71.             }
  72.         }
  73.         return sessions;
  74.     }
  75.  
  76.     /**
  77.      * 获取所有session
  78.      * @return  Collection<WebSocketSession>
  79.      */
  80.     public Collection<WebSocketSession> getAllSessions(){
  81.         return clientInfoSessionIds.values();
  82.     }
  83.  
  84.     /**
  85.      * 获取所有session
  86.      *
  87.      * @return
  88.      */
  89.     public ConcurrentMap<String, Set<String>> getAllSessionIds() {
  90.         return this.userSessionIds;
  91.     }
  92.     /**
  93.      * 获取所有session
  94.      *
  95.      * @return
  96.      */
  97.     public ConcurrentMap<String, WebSocketSession> getAllSessionWebSocketInfos() {
  98.         return this.clientInfoSessionIds;
  99.     }
  100.     /**
  101.      * register session
  102.      *
  103.      * @param user
  104.      * @param sessionId
  105.      */
  106.     private void registerSessionId(String user, String sessionId) {
  107.  
  108.         synchronized (this.lock) {
  109.             Set<String> set = this.userSessionIds.get(user);
  110.             if (set == null) {
  111.                 set = new CopyOnWriteArraySet<>();
  112.                 this.userSessionIds.put(user, set);
  113.             }
  114.             set.add(sessionId);
  115.         }
  116.     }
  117.  
  118.     /**
  119.      * 保存session
  120.      * @param session WebSocketSession
  121.      */
  122.     public  void registerSession(WebSocketSession session,String user){
  123.         if(StringUtils.isEmpty(user)){
  124.             user = parseUserByURI(session).get("user");
  125.         }
  126.         if(!StringUtils.isEmpty(user)){
  127.             String sessionId = session.getId();
  128.             registerSessionId(user,sessionId);
  129.             registerSessionId(session);
  130.             sessionIdUser.putIfAbsent(sessionId,user);
  131.         }
  132.     }
  133.     /**
  134.      * 从session里面解析参数
  135.      * @param session
  136.      * @return
  137.      */
  138.     private Map<String, String> parseUserByURI(WebSocketSession session){
  139.         Map<String, String> map = new HashMap<>();
  140.         String[] params = Optional.ofNullable(session.getHandshakeInfo().getUri().getQuery()).orElse("").split("&");
  141.         for (String param : params) {
  142.             String[] temp = param.split("=");
  143.             if(temp.length == 2){
  144.                 map.put(temp[0],temp[1]);
  145.             }
  146.         }
  147.         return map;
  148.     }
  149.     public WebSocketSession getSessionBySessionId(String sessionId){
  150.         return this.clientInfoSessionIds.get(sessionId);
  151.     }
  152.     private void registerSessionId(WebSocketSession websocketInfo) {
  153.         String sessionId = websocketInfo.getId();
  154.         CountDownLatch signal = cacheTimestamp.putIfAbsent(sessionId, new CountDownLatch(1));
  155.         if (signal == null) {
  156.             signal = cacheTimestamp.get(sessionId);
  157.             try {
  158.                 if (!clientInfoSessionIds.containsKey(sessionId)) {
  159.                     WebSocketSession set = this.clientInfoSessionIds.get(sessionId);
  160.                     if (set == null) {
  161.                         clientInfoSessionIds.putIfAbsent(sessionId, websocketInfo);
  162.                     }
  163.                 }
  164.             } finally {
  165.                 signal.countDown();
  166.                 cacheTimestamp.remove(sessionId);
  167.             }
  168.         } else {
  169.             try {
  170.                 signal.await();
  171.             } catch (InterruptedException e) {
  172.                 e.printStackTrace();
  173.             }
  174.         }
  175.     }
  176.  
  177.     private void unregisterSessionId(String userName, String sessionId) {
  178.  
  179.         synchronized (this.lock) {
  180.             Set set = this.userSessionIds.get(userName);
  181.             if (set != null && set.remove(sessionId) && set.isEmpty()) {
  182.                 this.userSessionIds.remove(userName);
  183.             }
  184.         }
  185.     }
  186.     private void unregisterSessionId(String sessionId) {
  187.  
  188.         synchronized (this.lock) {
  189.             WebSocketSession set = this.clientInfoSessionIds.get(sessionId);
  190.             if (set != null) {
  191.                 this.clientInfoSessionIds.remove(sessionId);
  192.             }
  193.         }
  194.     }
  195.  
  196.     public void unregisterSession(WebSocketSession session){
  197.         String sessionId = session.getId();
  198.         String user = sessionIdUser.get(sessionId);
  199.         if(!StringUtils.isEmpty(user)){
  200.             unregisterSessionId(sessionId);
  201.             unregisterSessionId(user,sessionId);
  202.             sessionIdUser.remove(sessionId);
  203.         }
  204.     }
  205.  
  206.     public void checkAndRemove(WebSocketSession session){
  207.         String sessionId = session.getId();
  208.         if(!this.clientInfoSessionIds.containsKey(sessionId)){
  209.             log.info("sessionId:{} 10秒内没有登陆,关掉它",sessionId);
  210.             session.close(SocketCloseStatus.UN_LOGIN.getCloseStatus()).toProcessor();
  211.         }else{
  212.             log.info("sessinId:{}已经登陆，是合法的",sessionId);
  213.         }
  214.     }
  215. }
userSessionIds是保存用记所属的sessionId列表的，因为同一个用户可能会在不同地方登陆，会有多个session
clientInfoSessionIds这个是保存session的，可以根据sessionId对应到用户。

这几周慢慢摸索出来的结果，网上资料很少，官网上的也不是很全，可能有不对的地方，在此做个记录！