devbox
知新_RL
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Websocket Stomp+RabbitMQ实现消息推送_stomp websocket rabbitmq

作者：知新_RL | 2024-08-05 09:35:13

赞

踩

stomp websocket rabbitmq

系列导读:

1.springboot+websocket构建在线聊天室（群聊+单聊）

2.Spring Boot WebSocket：单聊(实现思路)

3.Websocket Stomp+RabbitMQ实现消息推送

目录

1、技术栈:

2、依赖:

3、修改配置文件

4、RabbitConfig 

5、消息包装类

6、利用STOMP实现前后端长连接

7、编写前端页面：

8、编写消息生产者和消费者

9、测试

1、技术栈:

后端：springboot2.0.6

前端：html  js

2、依赖:

  1. <parent>
  2. 		<groupId>org.springframework.boot</groupId>
  3. 		<artifactId>spring-boot-starter-parent</artifactId>
  4. 		<version>2.0.6.RELEASE</version>
  5. 		<relativePath/> <!-- lookup parent from repository -->
  6. 	</parent>
  7. 	<groupId>com.zj</groupId>
  8. 	<artifactId>online-test</artifactId>
  9. 	<version>0.0.1-SNAPSHOT</version>
  10. 	<name>online-test</name>
  11. 	<description>Demo project for Spring Boot</description>
  12.  
  13. 	<properties>
  14. 		<java.version>1.8</java.version>
  15. 	</properties>
  16.  
  17. 	<dependencies>
  18. 		<dependency>
  19. 			<groupId>org.springframework.boot</groupId>
  20. 			<artifactId>spring-boot-starter-web</artifactId>
  21. 		</dependency>
  22. 		<dependency>
  23. 			<groupId>mysql</groupId>
  24. 			<artifactId>mysql-connector-java</artifactId>
  25. 		</dependency>
  26. 		<dependency>
  27. 			<groupId>org.springframework.boot</groupId>
  28. 			<artifactId>spring-boot-starter-data-jpa</artifactId>
  29. 		</dependency>
  30.         <dependency>
  31.             <groupId>commons-lang</groupId>
  32.             <artifactId>commons-lang</artifactId>
  33.             <version>2.6</version>
  34.         </dependency>
  35. 		<!--websocket 相关依赖-->
  36. 		<dependency>
  37. 			<groupId>org.springframework.boot</groupId>
  38. 			<artifactId>spring-boot-starter-websocket</artifactId>
  39. 		</dependency>
  40. 		<dependency>
  41. 			<groupId>org.webjars</groupId>
  42. 			<artifactId>webjars-locator-core</artifactId>
  43. 		</dependency>
  44. 		<dependency>
  45. 			<groupId>org.webjars</groupId>
  46. 			<artifactId>sockjs-client</artifactId>
  47. 			<version>1.0.2</version>
  48. 		</dependency>
  49. 		<dependency>
  50. 			<groupId>org.webjars</groupId>
  51. 			<artifactId>stomp-websocket</artifactId>
  52. 			<version>2.3.3</version>
  53. 		</dependency>
  54.  
  55.         <!--rabbitmq-->
  56.         <dependency>
  57.             <groupId>org.springframework.boot</groupId>
  58.             <artifactId>spring-boot-starter-amqp</artifactId>
  59.         </dependency>

3、修改配置文件

#rabbitmq

spring.rabbitmq.host=192.168.XXX

spring.rabbitmq.port=5672

spring.rabbitmq.username=guest

spring.rabbitmq.password=guest

 本文省略RabbitMq的安装，这边将你安装的rabbitmq的相关信息填入即可

参考安装地址: docker 安装 RabbitMq

几分钟就搞定安装

4、RabbitConfig 

先创一个rabbitmq 的配置类，由于我们这边业务逻辑比较简单，就简单使用rabbitmq一下

这边创一个hello的消息队列。

  1. @Configuration
  2. public class RabbitConfig {
  3.  
  4.     @Bean
  5.     public Queue helloQueue() {
  6.         return new Queue("hello");
  7.     }
  8.  
  9. }

5、消息包装类

先讲一下消息包装类，前后端的消息都以这种格式来传递，大家可以根据自己的需求自定义。

  1. public class RequestMessage {
  2.  
  3.     private String room;//频道号
  4.     private String type;//消息类型('1':客户端到服务端   '2'：客户端到服务端)
  5.     private String content;//消息内容（即答案）
  6.     private String userId;//用户id
  7.     private String questionId;//题目id
  8.     private String createTime;//时间
  9.  
  10.     public RequestMessage() {
  11.     }
  12.  
  13.     public RequestMessage(String room, String type, String content, String userId, String questionId, String createTime) {
  14.         this.room = room;
  15.         this.type = type;
  16.         this.content = content;
  17.         this.userId = userId;
  18.         this.questionId = questionId;
  19.         this.createTime = createTime;
  20.     }
  21.  
  22.     public String getRoom() {
  23.         return room;
  24.     }
  25.  
  26.     public String getType() {
  27.         return type;
  28.     }
  29.  
  30.     public String getContent() {
  31.         return content;
  32.     }
  33.  
  34.     public void setRoom(String room) {
  35.         this.room = room;
  36.     }
  37.  
  38.     public String getUserId() {
  39.         return userId;
  40.     }
  41.  
  42.     public void setUserId(String userId) {
  43.         this.userId = userId;
  44.     }
  45.  
  46.     public String getQuestionId() {
  47.         return questionId;
  48.     }
  49.  
  50.     public void setQuestionId(String questionId) {
  51.         this.questionId = questionId;
  52.     }
  53.  
  54.     public String getCreateTime() {
  55.         return createTime;
  56.     }
  57.  
  58.     public void setCreateTime(String createTime) {
  59.         this.createTime = createTime;
  60.     }
  61.  
  62.  
  63.     public void setType(String type) {
  64.         this.type = type;
  65.     }
  66.  
  67.     public void setContent(String content) {
  68.         this.content = content;
  69.     }
  70. }

6、利用STOMP实现前后端长连接

这部分与我之前的文章类似：SpringBoot+STOMP 实现聊天室（单聊+多聊）

本文这边 省略了消息群发

首先编写websocket配置类:

  1. @Configuration
  2. @EnableWebSocketMessageBroker
  3. public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
  4.  
  5.     @Override
  6.     public void configureMessageBroker(MessageBrokerRegistry config) {
  7.         /**
  8.          * 订阅来自"/topic"和"/user"的消息，
  9.          * /topic 单聊
  10.          * /all 群聊
  11.          */
  12.         config.enableSimpleBroker("/topic","/all");
  13.  
  14.         /**
  15.          * 客户端发送过来的消息，需要以"/app"为前缀，再经过Broker转发给响应的Controller
  16.          */
  17.         config.setApplicationDestinationPrefixes("/app");
  18.     }
  19.  
  20.     @Override
  21.     public void registerStompEndpoints(StompEndpointRegistry registry) {
  22.  
  23.         /**
  24.          * 路径"/websocket"被注册为STOMP端点，对外暴露，客户端通过该路径接入WebSocket服务
  25.          */
  26.         registry.addEndpoint("/websocket").setAllowedOrigins("*").withSockJS();
  27.     }
  28.  
  29.  
  30. }

然后编写消息转发器（前端的消息通过这边，进行转发）

  1. @RestController
  2. public class WebSocketTestController {
  3.  
  4.     @Autowired
  5.     private SimpMessagingTemplate messagingTemplate;
  6.  
  7.     @Autowired
  8.     Sender senderMQ;
  9.  
  10.     /**聊天室（单聊+多聊）&&消息转发
  11.      * @param requestMessage
  12.      * @throws Exception
  13.      */
  14.     @CrossOrigin
  15.     @MessageMapping("/chat")
  16.     public void messageHandling(RequestMessage requestMessage) throws Exception {
  17.         String destination = "/topic/" + HtmlUtils.htmlEscape(requestMessage.getRoom());
  18.  
  19.         String room = HtmlUtils.htmlEscape(requestMessage.getRoom());//htmlEscape  转换为HTML转义字符表示
  20.         String type = HtmlUtils.htmlEscape(requestMessage.getType());
  21.         String content = HtmlUtils.htmlEscape(requestMessage.getContent());
  22.         String userId = HtmlUtils.htmlEscape(requestMessage.getUserId());
  23.         String questionId = HtmlUtils.htmlEscape(requestMessage.getQuestionId());
  24.         String createTime = HtmlUtils.htmlEscape(requestMessage.getCreateTime());
  25.  
  26.         System.out.println( requestMessage.getRoom() );
  27.         System.out.println( content );
  28.  
  29.  
  30.       
  31.  
  32.         messagingTemplate.convertAndSend(destination, requestMessage);
  33.     }
  34.  
  35. }

注意：

1、使用＠MessageMapping注解来标识所有发送到“/chat”这个destination的消息，都会被路由到这个方法进行处理

2、使用@SendTo注解来标识这个方法返回的结果，都会被发送到它指定的destination，“/topic” 

3、传入的参数RequestMessage requestMessage为客户端发送过来的消息，是自动绑定的。

将前端传过来的消息解析，然后再通过messagingTemplate.convertAndSend（“目的地”，“消息内容”）转发出去。

7、编写前端页面：

这边的前端页面仅仅用了html+js

  1. <!DOCTYPE HTML>
  2. <html>
  3. <head>
  4.     <meta charset="UTF-8">
  5.     <title>My WebSocket</title>
  6.  
  7.     <script src="js/sockjs.min.js"></script>
  8.     <script src="js/jquery.min.js"></script>
  9.     <script src="js/stomp.min.js"></script>
  10.     <!--<script type="text/javascript"></script>-->
  11.  
  12.     <style>
  13.         #message22{
  14.             margin-top:40px;
  15.             border:1px solid gray;
  16.             padding:20px;
  17.         }
  18.     </style>
  19.  
  20.     <style>
  21.         #message{
  22.             margin-top:40px;
  23.             border:1px solid gray;
  24.             padding:20px;
  25.         }
  26.     </style>
  27.  
  28. </head>
  29.  
  30. <body>
  31.  
  32. 频道号：<input id="room" type="text"/>
  33. <button onclick="conectWebSocket()">连接WebSocket</button>
  34. <button onclick="disconnect()">断开连接</button>
  35. <hr />
  36.  
  37. <div id="message22"></div>
  38.  
  39.  
  40. <br />
  41. 做题区：<input id="text" type="text" />
  42. <!-- 频道号：<input id="toUser" type="text" /> -->
  43. <button onclick="sendMessage()">发送消息</button>
  44. <div id="message"></div>
  45. </body>
  46.  
  47. <script type="text/javascript">
  48.     var stompClient;
  49.  
  50.     var serverUrl = "http://localhost:8080/websocket";
  51.  
  52.     var room;//频道号
  53.  
  54.  
  55.     var websocket = null;
  56.  
  57.     //websocket连接
  58.     function conectWebSocket(){
  59.  
  60.         this.room = document.getElementById('room').value;//频道号
  61.  
  62.         console.log(this.room);
  63.  
  64.         console.log(this.serverUrl);
  65.  
  66.         var socket = new SockJS(this.serverUrl);
  67.         this.stompClient = Stomp.over(socket);
  68.         var that = this;
  69.           this.stompClient.connect({}, function (frame) {
  70.               that.stompClient.subscribe('/topic/'+that.room ,function(txt) {
  71.                   // console.log("websocket连接成功");
  72.                   // console.log(txt);
  73.                   document.getElementById('message').innerHTML += JSON.parse(txt.body)['content']+ '<br/>';
  74.                   // const sender = JSON.parse(message.body)['sender'];
  75.                 const language = JSON.parse(message.body)['language'];
  76.                 const content = JSON.parse(message.body)['content'];
  77.                 const type = JSON.parse(message.body)['type'];
  78.               });
  79.           });
  80.  
  81.  
  82.      }
  83.  
  84.      //发送消息
  85.     function sendMessage() {
  86.         //获取输入的文本信息进行发送
  87.         var message = document.getElementById('text').value;
  88.         // var room = document.getElementById('toUser').value;
  89.         var socketMsg = {msg:message,toUser:room};
  90.         var that = this
  91.         this.stompClient.send(
  92.             '/app/chat',
  93.             {},
  94.             JSON.stringify({
  95.                 'room': that.room,
  96.                 'type': "1",//1,2
  97.                 'content': message,
  98.                 'userId':"566072523",//小明
  99.                 'questionId':"222299023",//题目1
  100.                 'createTime':"",
  101.                  })
  102.         );
  103.  
  104.     }
  105.  
  106.  
  107.     function disconnect() {
  108.  
  109.          //断开连接的方法
  110.         if (this.stompClient !== undefined) {
  111.  
  112.             this.stompClient.disconnect();
  113.             alert("Disconnected");
  114.         }else{
  115.             alert("当前没有连接websocket")
  116.         }
  117.         this.stompClient = undefined;
  118.         
  119.     }
  120.  
  121. </script>
  122. </html>

页面都很简单，就不细讲了，但要引入sockjs.min.js   jquery.min.js   stomp.min.js这3个js

这样就前后端长连接就完成了，赶紧来试试~

 可以看出长连接成功

  下面我们要来讲讲 消息推送如何实现，当我们前端发一个http请求，在这个请求结束前，后端会新开一个线程。当原本的请求结束时，后端新开的那个线程还在运行，当它结束时，后端应该要返回消息给前端，但之前的请求已经结束，http握手已经早结束了，消息怎么传递。所以我们需要用到rabbitmq(用来消息的解耦)  websocket（用来保证前后端通信通道不关闭），来让后端主动推送的消息，能到前端页面显示。

8、编写消息生产者和消费者

  现在我们讲讲 rabbitmq 消息2个重要的组成部分：消息生产者和消息消费者

sender:

  1. @Component
  2. public class Sender {
  3.  
  4.     @Autowired
  5.     private AmqpTemplate rabbitTemplate;
  6.  
  7.     public void send(String context) { //注意因为是AmqpTemplate，所有这里只接受String,byte[],Seriz..
  8.         System.out.println("Sender : " + context);
  9.         this.rabbitTemplate.convertAndSend("hello", context);
  10.     }
  11.  
  12. }

这里注意一下，发送消息采用了AmqpTemplate 模板 。AmqpTemplate接口已经定义了发送和接收消息的基本操作。我们直接使用即可，但要注意的是 必须符合它的类型，这里只支持   String,byte[],Seriz..类型的消息

convertAndSend("hello", context);      hello就是我们前面创的消息队列，context是消息的内容

但我们有时候消息特别复杂，一般用对象来储存消息。这个怎么办呢？，这边我们先不讲，到我们下面再讲。我们继续来讲消息消费者

receiver:

  1. @Component
  2. @RabbitListener(queues = "hello")
  3. public class Receiver {
  4.  
  5.  
  6.     @Autowired
  7.     private SimpMessagingTemplate messagingTemplate;
  8.  
  9.  
  10.     @RabbitHandler
  11.     public void process(String context) throws IOException {
  12.         System.out.println("Receiver : " + context);
  13.  
  14.         RequestMessage mqTask = new RequestMessage(  );
  15.         BeanUtils.copyProperties( JsonUtils.jsonToObject( context,RequestMessage.class ),mqTask );
  16.  
  17.         if (Objects.equals( mqTask.getType(), "2" )) {
  18.             String destination = "/topic/" +mqTask.getRoom();
  19.  
  20.             messagingTemplate.convertAndSend( destination, mqTask);
  21.         }
  22.  
  23.     }
  24.  
  25. }

这里注意：我们前面说过，由于消息一般比较复杂，所以发送过来的消息一般是对象类型，但AmqpTemplate不支持，所以我们需要在消息生产者这边把它先转化成String型，传到消息消费者这边，再转化成对象。然后对消息进行处理。

最后通过            

messagingTemplate.convertAndSend( destination, mqTask);

将消息发送到前端页面上。

我们这边用到的工具类：

  1. public class JsonUtils {
  2.     private static final ObjectMapper MAPPER = new ObjectMapper();
  3.  
  4.     /**
  5.      * 对象-->Json字符串
  6.      * @version 创建时间：2018年4月17日 下午3:39:35
  7.      */
  8.     public static String objectToJson(Object data) {
  9.         try {
  10.             return MAPPER.writeValueAsString(data);
  11.         } catch (JsonProcessingException e) {
  12.             e.printStackTrace();
  13.         }
  14.         return null;
  15.     }
  16.  
  17.     /**
  18.      * Json字符串-->对象
  19.      * @version 创建时间：2018年4月17日 下午3:39:45
  20.      */
  21.     public static <T> T jsonToObject(String jsonData, Class<T> beanType) {
  22.         try {
  23.             return MAPPER.readValue(jsonData, beanType);
  24.         } catch (Exception e) {
  25.             e.printStackTrace();
  26.         }
  27.         return null;
  28.     }
  29.  
  30.     /**
  31.      * Json字符串--> List<对象>
  32.      * @version 创建时间：2018年4月17日 下午3:40:09
  33.      */
  34.     public static <T> List<T> jsonToList(String jsonData, Class<T> beanType) {
  35.         JavaType javaType = MAPPER.getTypeFactory().constructParametricType(List.class, beanType);
  36.         try {
  37.             return MAPPER.readValue(jsonData, javaType);
  38.         } catch (Exception e) {
  39.             e.printStackTrace();
  40.         }
  41.  
  42.         return null;
  43.     }
  44.  
  45.     public static <T> Map<String, T> jsonToMap(String jsonData) {
  46.         ObjectMapper mapper = new ObjectMapper();
  47.         try {
  48.             return mapper.readValue(jsonData, Map.class);
  49.         } catch (IOException e) {
  50.             e.printStackTrace();
  51.         }
  52.         return null;
  53.     }
  54.  
  55.     public static String mapToJson(Map map) {
  56.         try {
  57.             return MAPPER.writeValueAsString(map);
  58.         } catch (JsonProcessingException e) {
  59.             e.printStackTrace();
  60.         }
  61.         return null;
  62.     }
  63.  
  64.     public static <T> Set<T> jsonToSet(String jsonData) {
  65.         ObjectMapper mapper = new ObjectMapper();
  66.         try {
  67.             return mapper.readValue(jsonData, Set.class);
  68.         } catch (IOException e) {
  69.             e.printStackTrace();
  70.         }
  71.         return null;
  72.     }
  73.  
  74.     public static String setToJson(Set set) {
  75.         try {
  76.             return MAPPER.writeValueAsString(set);
  77.         } catch (JsonProcessingException e) {
  78.             e.printStackTrace();
  79.         }
  80.         return null;
  81.     }
  82. }

9、测试

我们先定义一个sevice方法，来实现异步。

  1. public interface CommonService {
  2.  
  3.     void testAsync();
  4.  
  5. }
  1. @Service
  2. public class CommonServiceImpl implements CommonService {
  3.  
  4.  
  5.     @Autowired
  6.     Sender sender;
  7.  
  8.  
  9.     @Async
  10.     @Override
  11.     public void testAsync() {
  12.  
  13.         RequestMessage mqTask = new RequestMessage(  );
  14.         for(int i=0;i<6;i++) {
  15.             mqTask.setRoom( "123");
  16.             mqTask.setUserId("000");
  17.             mqTask.setType( "2" );
  18.             mqTask.setQuestionId( "0000");
  19.             mqTask.setCreateTime( "0000");
  20.             mqTask.setContent("this:"+i);
  21.  
  22.             sender.send( JsonUtils.objectToJson( mqTask ) );
  23.             try {
  24.                 Thread.sleep( 1000 );
  25.             } catch (InterruptedException e) {
  26.                 e.printStackTrace();
  27.             }
  28.         }
  29.     }
  30.     
  31.  
  32.  
  33. }

 这里实现异步，为了简单就只用了@Async

再写一个测试controller

  1. @RestController
  2. public class TestController {
  3.  
  4.     @Autowired
  5.     CommonService commonService;
  6.  
  7.    
  8.     @GetMapping("/test/testAsync")
  9.     public String testAsync() {
  10.         commonService.testAsync();
  11.         return "http请求已结束";
  12.     }
  13.  
  14. }

 打开浏览器，先连接websocket,然后再开一个窗口，输入请求

localhost:8080/test/testAsync

发现如下：

 表明成功，本文暂且就到这了 = - =

声明：本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

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

  

闽ICP备14008679号