devbox
小小林熬夜学编程
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

实现Websocket集群及通信的第二种方式（含拦截器）_websockethandler

作者：小小林熬夜学编程 | 2024-03-15 14:53:08

赞

踩

websockethandler

一、第一种方式的缺点

        为了防止恶意占用网络连接资源，需要在websockt连接加入拦截器，但是在查找了大量网络资源后，根据注解@ServerEndpoint进行websocket连接的方式进行拦截我没有找到，其中有一篇博文是在@ServerEndPoint中加入自定义的配置器。

附：文章出处

去实现ServerEndpointConfig.Configurator内部类中的modifyHandShake方法进行拦截，我尝试了一下后，没有第二种方式简单，而且第二种方式具有通用性，较第一种方式要好一点，可以根据自己的情况进行选择。

二、第二种方式实现Websocket集群及通信

        集群只需要加入SpringCloud依赖加入注册中心，再使用网关进行同一转发、负载均衡即可搭建集群，同上一篇博文一致，在此篇不做展示。        

效果展示：

        如果说第一种方式是一个websocket连接一个解决方案，那么第二种方式就是websocket集体注册，共享解决方案。具体代码如下

1、目录结构

  

2、pom依赖

  1.     <parent>
  2.         <artifactId>spring-boot-starter-parent</artifactId>
  3.         <groupId>org.springframework.boot</groupId>
  4.         <version>2.1.6.RELEASE</version>
  5.     </parent>
  6.  
  7.     <properties>
  8.         <maven.compiler.source>8</maven.compiler.source>
  9.         <maven.compiler.target>8</maven.compiler.target>
  10.     </properties>
  11.     <dependencies>
  12.         <dependency>
  13.             <groupId>org.springframework</groupId>
  14.             <artifactId>spring-context</artifactId>
  15.         </dependency>
  16.         <dependency>
  17.             <groupId>org.springframework</groupId>
  18.             <artifactId>spring-beans</artifactId>
  19.         </dependency>
  20.         <dependency>
  21.             <groupId>org.springframework.boot</groupId>
  22.             <artifactId>spring-boot-starter-web</artifactId>
  23.         </dependency>
  24.         <dependency>
  25.             <groupId>org.springframework</groupId>
  26.             <artifactId>spring-aspects</artifactId>
  27.         </dependency>
  28.         <dependency>
  29.             <groupId>org.springframework.boot</groupId>
  30.             <artifactId>spring-boot-starter-redis</artifactId>
  31.             <version>1.4.1.RELEASE</version>
  32.         </dependency>
  33.         <dependency>
  34.             <groupId>org.springframework.cloud</groupId>
  35.             <artifactId>spring-cloud-starter</artifactId>
  36.             <version>2.1.0.RELEASE</version>
  37.         </dependency>
  38.         <dependency>
  39.             <groupId>org.springframework.boot</groupId>
  40.             <artifactId>spring-boot-starter-websocket</artifactId>
  41.         </dependency>
  42.         <dependency>
  43.             <groupId>org.projectlombok</groupId>
  44.             <artifactId>lombok</artifactId>
  45.         </dependency>
  46.         <dependency>
  47.             <groupId>com.alibaba</groupId>
  48.             <artifactId>fastjson</artifactId>
  49.             <version>1.2.9</version>
  50.         </dependency>
  51.         <dependency>
  52.             <groupId>org.springframework.boot</groupId>
  53.             <artifactId>spring-boot-starter-amqp</artifactId>
  54.         </dependency>
  55.     </dependencies>

3、代码根据结构从上至下

①SpringWebSocketConfig

  1. import org.springframework.context.annotation.Bean;
  2. import org.springframework.context.annotation.Configuration;
  3. import org.springframework.web.servlet.config.annotation.EnableWebMvc;
  4. import org.springframework.web.servlet.config.annotation.WebMvcConfigurerAdapter;
  5. import org.springframework.web.socket.config.annotation.EnableWebSocket;
  6. import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
  7. import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
  8. import org.springframework.web.socket.handler.TextWebSocketHandler;
  9.  
  10.  
  11. @Configuration
  12. @EnableWebMvc
  13. @EnableWebSocket
  14. public class SpringWebSocketConfig extends WebMvcConfigurerAdapter implements WebSocketConfigurer {
  15.     public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
  16.         //.setAllowedOrigins("*") 允许跨域访问
  17.         registry.addHandler(webSocketHandler(),"/webSocket").addInterceptors(new SpringWebSocketHandlerInterceptor()).setAllowedOrigins("*");
  18.         registry.addHandler(webSocketHandler(), "/sockjs/socketServer.do").addInterceptors(new SpringWebSocketHandlerInterceptor()).setAllowedOrigins("*");
  19.     }
  20.  
  21.     @Bean
  22.     public TextWebSocketHandler webSocketHandler(){
  23.         return new SpringWebSocketHandler();
  24.     }
  25.  
  26. }

②Controller层没有实际意义，不写也可以

③SpringWebSocketHandler

  1. import com.alibaba.fastjson.JSONObject;
  2. import org.slf4j.Logger;
  3. import org.slf4j.LoggerFactory;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.stereotype.Component;
  6. import org.springframework.web.socket.CloseStatus;
  7. import org.springframework.web.socket.TextMessage;
  8. import org.springframework.web.socket.WebSocketMessage;
  9. import org.springframework.web.socket.WebSocketSession;
  10. import org.springframework.web.socket.handler.TextWebSocketHandler;
  11. import java.io.IOException;
  12. import java.util.Map;
  13. import java.util.Set;
  14. import java.util.concurrent.ConcurrentHashMap;
  15. import java.util.concurrent.atomic.AtomicInteger;
  16.  
  17. /**
  18.  * @author secondj
  19.  * @Date 2021/11/15 10:26
  20.  */
  21. @Component
  22. public class SpringWebSocketHandler extends TextWebSocketHandler {
  23.     @Autowired
  24.     FanoutSender fanoutSender;
  25.  
  26.     private static final AtomicInteger ati = new AtomicInteger();
  27.  
  28.     public static final ConcurrentHashMap<String,WebSocketSession> map = new ConcurrentHashMap<>();
  29.  
  30.     private static Logger logger = LoggerFactory.getLogger(SpringWebSocketHandler.class);
  31.  
  32.     public SpringWebSocketHandler() {
  33.         // TODO Auto-generated constructor stub
  34.     }
  35.  
  36.     /**
  37.      * 连接成功后给前端发送的消息，会触发页面上onopen方法
  38.      */
  39.     public void afterConnectionEstablished(WebSocketSession session) throws Exception {
  40.         // TODO Auto-generated method stub
  41.         Object sid = session.getAttributes().get("tdt_sid");
  42.         map.put(sid.toString(),session);
  43.         int num = ati.incrementAndGet();
  44.         logger.info("connect to the websocket success......当前数量:{}",num);
  45.         //这块会实现自己业务，比如，当用户登录后，会把离线消息推送给用户
  46.         TextMessage returnMessage = new TextMessage("连接成功");
  47.         session.sendMessage(returnMessage);
  48.     }
  49.  
  50.     /**
  51.      * 关闭连接时触发
  52.      * 调用session.close也会触发
  53.      */
  54.     public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception {
  55.         logger.debug("websocket connection closed......");
  56.         String sid= (String) session.getAttributes().get("tdt_sid");
  57.         logger.info("用户"+sid+"已退出！");
  58.         if(map.containsKey(sid)){
  59.             map.remove(sid);
  60.         }
  61.         int num = ati.decrementAndGet();
  62.         logger.info("剩余在线用户:{}"+num);
  63.     }
  64.  
  65.     /**
  66.      * js调用websocket.send时候，会调用该方法
  67.      */
  68.     @Override
  69.     protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
  70.         WebSocketMessage message2self = new TextMessage("该用户没有上线".getBytes());
  71.         String payload = message.getPayload();
  72.         WebSocketVO webSocketVO = JSONObject.parseObject(payload, WebSocketVO.class);
  73.         String toSid = webSocketVO.getToUserId();
  74.         if(map.containsKey(toSid)){
  75.             map.get(toSid).sendMessage(message);
  76.         }else {
  77.             ConcurrentHashMap<String,String> sendMap = new ConcurrentHashMap();
  78.             sendMap.put(toSid,message.getPayload());
  79.             logger.info("getPayLoad():{}",message.getPayload());
  80.             logger.info("messge:{}",message.toString());
  81.             logger.info("map数据:{}",sendMap.toString());
  82.             fanoutSender.sendMessage(sendMap.toString());
  83.         }
  84.     }
  85.  
  86.     public void sendMessage(TextMessage message) throws Exception {
  87.         WebSocketMessage message2self = new TextMessage("该用户没有上线".getBytes());
  88.         String payload = message.getPayload();
  89.         WebSocketVO webSocketVO = JSONObject.parseObject(payload, WebSocketVO.class);
  90.         String toSid = webSocketVO.getToUserId();
  91.         if(map.containsKey(toSid)){
  92.             map.get(toSid).sendMessage(message);
  93.         }
  94.     }
  95.  
  96.     /**
  97.      * 代理发生异常时执行
  98.      */
  99.     public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
  100.         if(session.isOpen()){session.close();}
  101.         logger.debug("websocket connection closed......");
  102.         Map<String, Object> attributes = session.getAttributes();
  103.         Object sid = attributes.get("sid");
  104.         map.remove(sid);
  105.         WebSocketMessage webSocketMessage = new TextMessage("发生异常错误".getBytes());
  106.         session.sendMessage(webSocketMessage);
  107.     }
  108.  
  109.     public boolean supportsPartialMessages() {
  110.         return false;
  111.     }
  112.  
  113.     /**
  114.      * 给某个用户发送消息
  115.      *
  116.      * @param sid
  117.      * @param message
  118.      */
  119.     public void sendMessageToUser(String sid, TextMessage message) {
  120.         Set<Map.Entry<String, WebSocketSession>> entries = map.entrySet();
  121.         for (Map.Entry<String, WebSocketSession> entry : entries) {
  122.             if (entry.getValue().getAttributes().get("tdt_sid").equals(sid)) {
  123.                 try {
  124.                     if (entry.getValue().isOpen()) {
  125.                         entry.getValue().sendMessage(message);
  126.                     }
  127.                 } catch (IOException e) {
  128.                     e.printStackTrace();
  129.                 }
  130.                 break;
  131.             }
  132.         }
  133.     }
  134.  
  135.     public void sendMessageToUser(String sid, String message) {
  136.         TextMessage messages = new TextMessage(message.getBytes());
  137.         this.sendMessageToUser(sid,messages);
  138.     }
  139.  
  140.     /**
  141.      * 给所有在线用户发送消息
  142.      *
  143.      * @param message
  144.      */
  145.     public void sendMessageToUsers(TextMessage message) throws IOException {
  146.         Set<Map.Entry<String, WebSocketSession>> entries = map.entrySet();
  147.         for (Map.Entry<String, WebSocketSession> entry : entries) {
  148.             WebSocketSession session = entry.getValue();
  149.             if(session.isOpen()){
  150.                 session.sendMessage(message);
  151.             }
  152.         }
  153.     }
  154. }

④SpringWebSocketHandlerInterceptor

  1. import org.springframework.http.server.ServerHttpRequest;
  2. import org.springframework.http.server.ServerHttpResponse;
  3. import org.springframework.http.server.ServletServerHttpRequest;
  4. import org.springframework.stereotype.Component;
  5. import org.springframework.web.socket.WebSocketHandler;
  6. import org.springframework.web.socket.server.support.HttpSessionHandshakeInterceptor;
  7.  
  8. import javax.servlet.http.HttpSession;
  9. import java.util.Map;
  10.  
  11. /**
  12.  * @author secondj
  13.  * @Date 2021/11/11 15:52
  14.  */
  15. @Component
  16. public class SpringWebSocketHandlerInterceptor extends HttpSessionHandshakeInterceptor {
  17.  
  18.     @Override
  19.     public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler,
  20.                                    Map<String, Object> attributes) throws Exception {
  21.         //TODO 获取url传递的参数，通过attributes在Interceptor处理结束后传递给WebSocketHandler
  22.         //TODO WebSocketHandler可以通过WebSocketSession的getAttributes()方法获取参数
  23.         //设置session值这里根据自己需求设置即可
  24.         if (request instanceof ServletServerHttpRequest) {
  25.             ServletServerHttpRequest servletRequest = (ServletServerHttpRequest) request;
  26.             HttpSession session = servletRequest.getServletRequest().getSession(true);
  27.             String sid = servletRequest.getServletRequest().getParameter("userId");
  28.             String token = servletRequest.getServletRequest().getParameter("token");
  29.             if(!token.equals("EXAM_PERMISSION")){
  30.                 return false;
  31.             }
  32.             if (session != null) {
  33.                 String userName = (String) session.getAttribute("exam_sid");
  34.                 if (userName == null) {
  35.                     userName = sid;
  36.                 }
  37.                 attributes.put("exam_sid",userName);
  38.                 return true;
  39.             }
  40.         }
  41.         return false;
  42.     }
  43.  
  44.     @Override
  45.     public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler,
  46.                                Exception ex) {
  47.         // TODO Auto-generated method stub
  48.         super.afterHandshake(request, response, wsHandler, ex);
  49.     }
  50. }

⑤RetryCache该类是用来缓存队列消息和消息重发的，属于消息安全

  1. import lombok.AllArgsConstructor;
  2. import lombok.Data;
  3. import lombok.NoArgsConstructor;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.springframework.stereotype.Component;
  6.  
  7. import java.util.Map;
  8. import java.util.concurrent.ConcurrentHashMap;
  9. import java.util.concurrent.atomic.AtomicInteger;
  10.  
  11. /**
  12.  * @author secondj
  13.  * @Date 2021/11/15 16:28
  14.  */
  15. @Slf4j
  16. @Component
  17. public class RetryCache {
  18.     private SendMessage sendMessage;
  19.  
  20.     private boolean stop = false;
  21.  
  22.     private Map<String,MessageWithTime> map = new ConcurrentHashMap<>();
  23.  
  24.     private AtomicInteger id = new AtomicInteger();
  25.  
  26.     @Data
  27.     @AllArgsConstructor
  28.     @NoArgsConstructor
  29.     private static class MessageWithTime{
  30.         long time;
  31.         Object message;
  32.     }
  33.  
  34.     public void sender(SendMessage sendMessage){
  35.         this.sendMessage = sendMessage;
  36.         startRetry();
  37.     }
  38.  
  39.     public String generaterId(){
  40.         return ""+id.incrementAndGet();
  41.     }
  42.  
  43.     public void add(String id,Object message){
  44.         map.put(id,new MessageWithTime(System.currentTimeMillis(),message));
  45.     }
  46.  
  47.     public void del(String id){
  48.         map.remove(id);
  49.     }
  50.  
  51.     //多线程发送消息
  52.     private void startRetry(){
  53.         new Thread(()->{
  54.             while(!stop){
  55.                 try {
  56.                     Thread.sleep(System.currentTimeMillis());
  57.                 } catch (InterruptedException e) {
  58.                     e.printStackTrace();
  59.                 }
  60.                 long now = System.currentTimeMillis();
  61.                 for (String key : map.keySet()) {
  62.                     MessageWithTime messageWithTime = map.get(key);
  63.                     if(null != messageWithTime){
  64.                         if(messageWithTime.getTime()+ 3 * Constant.VALID_TIME < now){
  65.                             log.info("send message failed after 3 min " + messageWithTime);
  66.                             del(key);
  67.                         }else if (messageWithTime.getTime() + Constant.VALID_TIME< now) {
  68.                             DetailResult detailRes = sendMessage.send(messageWithTime.getMessage());
  69.                             if (detailRes.isSuccess()) {
  70.                                 del(key);
  71.                             }
  72.                         }
  73.                     }
  74.                 }
  75.             }
  76.         }).start();
  77.     }
  78. }

⑥RabbitMQConfig

  1. import lombok.extern.slf4j.Slf4j;
  2. import org.springframework.amqp.core.*;
  3. import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
  4. import org.springframework.amqp.rabbit.connection.ConnectionFactory;
  5. import org.springframework.amqp.rabbit.connection.CorrelationData;
  6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.beans.factory.annotation.Value;
  9. import org.springframework.context.annotation.Bean;
  10. import org.springframework.context.annotation.Configuration;
  11.  
  12. /**
  13.  * @author secondj
  14.  * @Date 2021/11/15 16:20
  15.  */
  16. @Configuration
  17. @Slf4j
  18. public class RabbitMQConfig {
  19.     @Value("${spring.rabbitmq.addresses}")
  20.     public String addresses;
  21.  
  22.     @Value("${spring.rabbitmq.port}")
  23.     public String port;
  24.  
  25.     @Value("${spring.rabbitmq.username}")
  26.     private String username;
  27.  
  28.     @Value("${spring.rabbitmq.password}")
  29.     private String password;
  30.  
  31.     @Value("${spring.rabbitmq.virtual-host}")
  32.     private String virtualHost;
  33.  
  34.     @Value("${spring.rabbitmq.publisher-confirms}")
  35.     private boolean publisherConfirms;
  36.  
  37.     @Value("${tdt.queue}")
  38.     public String queue;
  39.  
  40.     @Value("${tdt.exchange}")
  41.     public String exchange;
  42.  
  43.     @Autowired
  44.     RetryCache retryCache;
  45.  
  46.     /**
  47.      * 创建连接工厂
  48.      * @return
  49.      */
  50.     @Bean
  51.     public ConnectionFactory connectionFactory(){
  52.         CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
  53.         connectionFactory.setHost("127.0.0.1");
  54.         connectionFactory.setPort(Integer.valueOf(port));
  55.         connectionFactory.setUsername(username);
  56.         connectionFactory.setPassword(password);
  57.         connectionFactory.setVirtualHost(virtualHost);
  58.         connectionFactory.setPublisherConfirms(publisherConfirms);
  59.         return connectionFactory;
  60.     }
  61.  
  62.     @Bean
  63.     public Queue queueTdt(){
  64.         log.info("创建队列成功：{}",queue);
  65.         return new Queue(queue);
  66.     }
  67.  
  68.     @Bean
  69.     public FanoutExchange fanoutExchangeTdt(){
  70.         log.info("创建交换机成功:{}",exchange);
  71.         return new FanoutExchange(exchange);
  72.     }
  73.  
  74.     @Bean
  75.     public Binding bindingTdt(){
  76.         Binding bind = BindingBuilder.bind(queueTdt()).to(fanoutExchangeTdt());
  77.         log.info("交换机队列绑定成功");
  78.         return bind;
  79.     }
  80.  
  81.     @Bean
  82.     public RabbitTemplate rabbitTemplate(){
  83.         RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
  84.         //TODO 失败通知
  85.         rabbitTemplate.setMandatory(true);
  86.         //TODO 失败回调
  87.         rabbitTemplate.setReturnCallback(returnCallback());
  88.         //TODO 发送方确认
  89.         rabbitTemplate.setConfirmCallback(confirmCallback());
  90.         return rabbitTemplate;
  91.     }
  92.  
  93.     //===============发送方确认===============
  94.     public RabbitTemplate.ConfirmCallback confirmCallback(){
  95.         return new RabbitTemplate.ConfirmCallback(){
  96.             @Override
  97.             public void confirm(CorrelationData correlationData,
  98.                                 boolean ack, String cause) {
  99.                 if (ack) {
  100.                     log.info("发送者确认发送给mq成功");
  101.                 } else {
  102.                     //处理失败的消息
  103.                     log.info("发送者发送给mq失败,考虑重发:"+cause);
  104.                 }
  105.             }
  106.         };
  107.     }
  108.  
  109.     //===============失败通知===============
  110.     public RabbitTemplate.ReturnCallback returnCallback(){
  111.         return new RabbitTemplate.ReturnCallback(){
  112.             @Override
  113.             public void returnedMessage(Message message,
  114.                                         int replyCode,
  115.                                         String replyText,
  116.                                         String exchange,
  117.                                         String routingKey) {
  118.                 log.info("无法路由的消息，需要考虑另外处理。");
  119.                 log.info("Returned replyText："+replyText);
  120.                 log.info("Returned exchange："+exchange);
  121.                 log.info("Returned routingKey："+routingKey);
  122.                 String msgJson = new String(message.getBody());
  123.                 log.info("Returned Message："+msgJson);
  124.             }
  125.         };
  126.     }
  127. }

⑦Constant

  1. /**
  2.  * @author secondj
  3.  * @Date 2021/11/15 16:31
  4.  */
  5. public class Constant {
  6.     public static final long VALID_TIME = 3600l;
  7. }

⑧ReceiveMessage和SendMessage

  1. /**
  2.  * @author secondj
  3.  * @Date 2021/11/15 16:45
  4.  */
  5. public interface ReceiveMessage {
  6.     DetailResult receive(Object obj);
  7. }
  1. /**
  2.  * @author secondj
  3.  * @Date 2021/11/15 16:29
  4.  */
  5. public interface SendMessage {
  6.     DetailResult send(Object obj);
  7. }

⑨FanoutReceiver

  1. import com.alibaba.fastjson.JSONException;
  2. import com.alibaba.fastjson.JSONObject;
  3. import com.rabbitmq.client.Channel;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.slf4j.Logger;
  6. import org.slf4j.LoggerFactory;
  7. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  8. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  9. import org.springframework.amqp.support.AmqpHeaders;
  10. import org.springframework.beans.factory.annotation.Autowired;
  11. import org.springframework.messaging.handler.annotation.Header;
  12. import org.springframework.stereotype.Component;
  13. import java.io.IOException;
  14. import java.util.Map;
  15. import java.util.Set;
  16.  
  17. /**
  18.  * @author secondj
  19.  * @Date 2021/11/15 16:22
  20.  */
  21. @Component
  22. @Slf4j
  23. public class FanoutReceiver {
  24.     private static Logger logger = LoggerFactory.getLogger(FanoutReceiver.class);
  25.     @Autowired
  26.     SpringWebSocketHandler handler;
  27.  
  28.     @RabbitHandler
  29.     @RabbitListener(queues = "queue_mqsocket")//动态绑定
  30.     public void receiveMessage(String jsonObject, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
  31.         //返回字符串
  32.         try{
  33.             log.info("队列接收到消息：{}",jsonObject);
  34.             jsonObject = jsonObject.replace("=",":");
  35.             Map<String,String> mapstr = JSONObject.parseObject(jsonObject, Map.class);
  36.             Set<Map.Entry<String, String>> entries = mapstr.entrySet();
  37.             for (Map.Entry<String, String> entry : entries) {
  38.                 Object obj = entry.getKey();
  39.                 String sid = obj.toString();
  40.  
  41.                 Object objs = entry.getValue();
  42.                 String message = objs.toString();
  43.  
  44.                 if(SpringWebSocketHandler.map.containsKey(sid)){
  45.                     handler.sendMessageToUser(sid,message);
  46.                 }
  47.             }
  48.         }catch (JSONException e){
  49.             e.printStackTrace();
  50.             return;
  51.         }
  52.         try {
  53.             channel.basicAck(tag,false);
  54.         } catch (IOException e) {
  55.             e.printStackTrace();
  56.         }
  57.     }
  58. }

⑩DetailResult

  1. import lombok.AllArgsConstructor;
  2. import lombok.Data;
  3. import lombok.NoArgsConstructor;
  4.  
  5. /**
  6.  * @author secondj
  7.  * @Date 2021/11/15 16:30
  8.  */
  9. @Data
  10. @AllArgsConstructor
  11. @NoArgsConstructor
  12. public class DetailResult {
  13.     private boolean flag;
  14.     public Object message;
  15.  
  16.     public boolean isSuccess() {
  17.         return flag == true;
  18.     }
  19. }

⑩①FanoutSender

  1. import cn.tdt.rabbitmq.cache.RetryCache;
  2. import cn.tdt.rabbitmq.function.SendMessage;
  3. import cn.tdt.rabbitmq.result.DetailResult;
  4. import com.alibaba.fastjson.JSONObject;
  5. import lombok.extern.slf4j.Slf4j;
  6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.stereotype.Component;
  9.  
  10. /**
  11.  * @author secondj
  12.  * @Date 2021/11/15 16:22
  13.  */
  14. @Component
  15. @Slf4j
  16. public class FanoutSender implements SendMessage {
  17.     RetryCache retryCache = new RetryCache();
  18.     @Autowired
  19.     RabbitTemplate rabbitTemplate;
  20.  
  21.     //发送消息
  22.     public void sendMessage(Object obj) {
  23.         rabbitTemplate.setExchange("exchange_mqsocket");
  24.         log.info("【消息发送者】发送消息到fanout交换机"+ JSONObject.toJSONString(obj));
  25.         try{
  26.             send(obj);
  27.         }catch (RuntimeException ex){
  28.             ex.printStackTrace();
  29.             log.info("send failed"+ex);
  30.             try{
  31.                 send(obj);
  32.             }catch (RuntimeException e){
  33.                 e.printStackTrace();
  34.                 log.info("retry send failed"+e);
  35.             }
  36.         }
  37.     }
  38.  
  39.     //客户端发送消息前，先在本地进行缓存
  40.     @Override
  41.     public DetailResult send(Object message) {
  42.         try{
  43.             String id = retryCache.generaterId();
  44.             retryCache.add(id,message);
  45.             rabbitTemplate.convertAndSend("exchange_mqsocket","",message);
  46. //            rabbitTemplate.correlationConvertAndSend(message,new CorrelationData(id));
  47. //            rabbitTemplate.correlationConvertAndSend(message,new CorrelationData(id));
  48.         }catch (Exception e){
  49.             return new DetailResult(false,"");
  50.         }
  51.         return new DetailResult(true,"");
  52.     }
  53. }

⑩②WebSocketVO

  1. import lombok.Data;
  2.  
  3. /**
  4.  * @author secondj
  5.  * @Date 2021/11/15 14:53
  6.  */
  7. @Data
  8. public class WebSocketVO {
  9.     private String toUserId;
  10.     private String msgType;
  11.     private String msgInfo;
  12. }

⑩③启动类

  1. import org.springframework.boot.SpringApplication;
  2. import org.springframework.boot.autoconfigure.SpringBootApplication;
  3.  
  4. /**
  5.  * @author secondj
  6.  * @Date 2021/11/11 15:37
  7.  */
  8. @SpringBootApplication
  9. public class SSOApplicationRun {
  10.     public static void main(String[] args) {
  11.         SpringApplication.run(SSOApplicationRun.class,args);
  12.     }
  13. }

前端代码

  1. <!DOCTYPE html>
  2. <html lang="en">
  3. <head>
  4.     <meta charset="UTF-8">
  5.     <title>WebSocket</title>
  6.  
  7. </head>
  8. <body>
  9. <h3>hello socket</h3>
  10. <p>【userId】：<div><input id="userId" name="userId" type="text"></div>
  11. <p>【toUserId】：<div><input id="toUserId" name="toUserId" type="text"></div>
  12. <p>【msgType】：<div><input id="msgType" name="msgType" type="text"></div>
  13. <p>【msgInfo】：<div><input id="msgInfo" name="msgInfo" type="text"></div>
  14. <p>【操作】:<div><button onclick="openSocket()">开启socket</button></div>
  15. <p>【操作】：<div><button onclick="sendMessage()">发送消息</button></div>
  16. </body>
  17. <script>
  18.  
  19.  
  20.     var socket;
  21.     function openSocket() {
  22.         if(typeof(WebSocket) == "undefined") {
  23.             console.log("您的浏览器不支持WebSocket");
  24.         }else{
  25.             console.log("您的浏览器支持WebSocket");
  26.             //实现化WebSocket对象，指定要连接的服务器地址与端口  建立连接
  27.             var userId = document.getElementById('userId').value;
  28.             //这里采用get请求的方式发送请求，拦截器可以根据getParameter方式获取参数值
  29.             var socketUrl="ws://localhost:8080/ctl/webSocket?token=TDT_PERMISSION&userId="+userId;
  30.             console.log(socketUrl);
  31.             if(socket != null){
  32.                 socket.close();
  33.                 socket=null;
  34.             }
  35.             socket = new WebSocket(socketUrl);
  36.             //打开事件
  37.             socket.onopen = function() {
  38.                 console.log("websocket已打开");
  39.                 //socket.send("这是来自客户端的消息" + location.href + new Date());
  40.             };
  41.             //获得消息事件
  42.             socket.onmessage = function(msg) {
  43.                 var serverMsg = "收到服务端信息：" + msg.data;
  44.                 console.log(serverMsg);
  45.                 //发现消息进入    开始处理前端触发逻辑
  46.             };
  47.             //关闭事件
  48.             socket.onclose = function() {
  49.                 console.log("websocket已关闭");
  50.             };
  51.             //发生了错误事件
  52.             socket.onerror = function() {
  53.                 console.log("websocket发生了错误");
  54.             }
  55.         }
  56.     }
  57.     function sendMessage() {
  58.         if(typeof(WebSocket) == "undefined") {
  59.             console.log("您的浏览器不支持WebSocket");
  60.         }else {
  61.             // console.log("您的浏览器支持WebSocket");
  62.             var toUserId = document.getElementById('toUserId').value;
  63.             var msgInfo = document.getElementById('msgInfo').value;
  64.             var msgType = document.getElementById('msgType').value;
  65.             var msg = '{"toUserId":"'+toUserId+'","msgInfo":"'+msgInfo+'","msgType":"'+msgType+'"}';
  66.             console.log(msg);
  67.             socket.send(msg);
  68.         }
  69.     }
  70.  
  71. </script>
  72. </html>

遇到的问题：

        在往消息队列中发送消息时，如果发送的消息是对象，会在接收消息时自动添加Properties字段属性，影响json转换，解决方案为把发送的消息对象转换为字符串对象或者json串。

待优化的问题：

        在单点登录系统中，拦截websocket请求后，未登录用户进行重定向。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小小林熬夜学编程/article/detail/241983
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号