Spring WebSocket 结合 RabbitMq

文章目录

Spring WebSocket 结合 RabbitMq
- - - 参考链接
  - 前言
  - RabbitMQ安装
  - - Windows10安装
    - - 步骤
    - RabbitMq安装 Stomp 插件
  - RbbitMq Stomp 合法前缀约束
  - - 官方解释
    - 个人实验理解
    - - /exchange 前缀
      - /queue 前缀
      - /topic 前缀
      - /amq/queue 前缀
      - /temp-queue/
    - Stomp 控制队列属性
  - 代码实现
  - - 依赖注入
    - 基于 Stomp 目的路径形式
    - 直接使用 RabbitTemplate
  - 集群 / 负载均衡简单看法
  - 项目代码

参考链接

Spring 官文： https://docs.spring.io/spring-framework/docs/6.0.0-SNAPSHOT/reference/html/web.html#websocket-stomp-benefits

前端页面： https://github.com/callicoder/spring-boot-websocket-chat-demo

前言

阅读该篇文章请先浏览另一篇，该篇文章是基于其上的改进：
Springboot整合WebSocket（基于Stomp）

此外可以看下另外相关文章：

Springboot 整合 WebSocket 简单实现

Springboot整合RabbitMQ详解

RabbitMQ安装 Windows10安装步骤

到erlang官网下载win10版安装包。下载完成后傻瓜式安装。
配置erlang环境变量

cmd输入erl验证安装是否成功，如下成功；ctrl+c退出

傻瓜式安装RabbitMQ服务。
在RabbitMQ的gitHub项目中，下载window版本的服务端安装包
进入安装目录，sbin目录下，执行：rabbitmq-plugins enable rabbitmq_management 命令安装管理页面的插件

双击rabbitmq-server.bat启动脚本或者在sbin目录下执行：rabbitmq-server start，然后打开服务管理可以看到RabbitMQ正在运行。
打开浏览器输入http://localhost:15672，账号密码默认是：guest/guest

同理配置RabbitMq环境变量，方便我们直接cmd输入命令

RabbitMq安装 Stomp 插件

cmd 进入 RabittMq 安装位置 /sbin 目录下，执行以下两个命令：

rabbitmq-plugins enable rabbitmq_stomp

rabbitmq-plugins enable rabbitmq_web_stomp

重启 RabbitMq 服务

cmd 进入 RabittMq 安装位置 /bin 目录下，执行以下命令

# 停止 rabbitmq 服务
rabbitmq-service stop
# 启动 rabbitmq 服务
rabbitmq-service start

进入 RabbitMq 首页可以看到插件已安装

RbbitMq Stomp 合法前缀约束官方解释

https://www.rabbitmq.com/stomp.html

个人实验理解 /exchange 前缀

格式：/exchange/< exchangeName>/ < pattern>

解释：该前缀需要手动创建一个 exchangeName 的交换机，否则会出现错误。
/exchange 不适合使用现有队列中的消息。为每个订阅者创建一个新队列，并使用提供的路由键绑定到指定的交换。要使用现有队列，请使用 /amq/queue 。默认队列会在无人订阅时自动删除。

/queue 前缀

格式：/queue /< pattern>

解释：该前缀会发送到默认 amq.topic 交换机，创建的队列名由 queueName 指定。
如果未指定队列参数，则默认队列是持久的、非独占的、非自动删除的。
此外若有多个客户端订阅该队列，当队列存在消息时会被订阅该消息的客户端轮询消费，而不是所有客户端都消费该消息。

/topic 前缀

格式：/topic /< pattern>

解释：该前缀会使用默认 amq.topic 交换机，若没有符合 pattern 的队列，则创建一个随机队列名，队列的 routeKey/bindKey 即为pattern，默认队列会在无人订阅时自动删除。
可以使用 stomp.default_topic_exchange 指定交换机

/amq/queue 前缀

格式：/amq/queue/< name>

解释：该前缀会使用默认交换机，但需要我们自己创建 name 队列，否则会报错。
如果未指定队列参数，则默认队列是持久的、非独占的、非自动删除的。

/temp-queue/

创建一个临时队列（具有生成的名称），该队列专用于会话并自动订阅该队列。

Stomp 控制队列属性

同样见官文：https://www.rabbitmq.com/stomp.html

代码实现依赖注入



    
    
        org.springframework.boot
        spring-boot-starter-websocket
    

    
    
        org.springframework.boot
        spring-boot-starter-amqp
    

    
    
        org.springframework.boot
        spring-boot-starter-reactor-netty
    


    
        org.springframework.boot
        spring-boot-starter-test
        test
    

    
    
        org.projectlombok
        lombok
        compile
    

    
        com.alibaba
        fastjson
        1.2.47

基于 Stomp 目的路径形式

服务端 Controller

@RestController
public class ChatController {

    @Autowired
    private SimpMessagingTemplate messagingTemplate;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    
    @MessageMapping("/chat/sendMessage")
    public void sendMessage(@Payload ChatMessage chatMessage) {
        // 发送到指定fanoutQueue队列
        messagingTemplate.convertAndSend("/amq/queue/fanoutQueue",chatMessage);
    }

    
    @PostMapping("/chat/single")
    public void sendSingleMessage(@RequestBody ChatMessage chatMessage) {       										messagingTemplate.convertAndSend("/topic/single"+chatMessage.getToUser(),chatMessage);
    }

    
    @MessageMapping("/chat/addUser")
    public void addUser(@Payload ChatMessage chatMessage,
                               SimpMessageHeaderAccessor headerAccessor) {
        // Add username in web socket session
        headerAccessor.getSessionAttributes().put("username", chatMessage.getSender());
        messagingTemplate.convertAndSend("/amq/queue/fanoutQueue",chatMessage);
    }

}

前端 js 订阅

function onConnected() {
    // 订阅群发主题，特定队列 fanoutQueue
    stompClient.subscribe('/amq/queue/fanoutQueue', onMessageReceived);
    
    // 订阅一对一主题，即通过用户名等唯一性标识拼接到订阅主题地址
    stompClient.subscribe('/exchange/WebSocket/single.' + username, onMessageReceived);

    stompClient.send("/app/chat/addUser",
        {},
        JSON.stringify({sender: username, type: 'JOIN'})
    )

    connectingElement.classList.add('hidden');
}

直接使用 RabbitTemplate

服务端发送消息改写

@RestController
public class ChatController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    
    @PostMapping("/chat/sendMessage")
    public void sendMsg(@RequestBody ChatMessage chatMessage) {
        String str = JSON.toJSONString(chatMessage);
        rabbitTemplate.convertAndSend("WebSocket","*",str);
    }

    
    @PostMapping("/chat/single")
    public void sendSingleMessage(@RequestBody ChatMessage chatMessage) {
        String str = JSON.toJSONString(chatMessage);
        rabbitTemplate.convertAndSend("WebSocket","single." + chatMessage.getToUser(),str);

    }

    
    @MessageMapping("/chat/addUser")
    public void addUser(@Payload ChatMessage chatMessage,
                               SimpMessageHeaderAccessor headerAccessor) {
        // Add username in web socket session
        headerAccessor.getSessionAttributes().put("username", chatMessage.getSender());
        messagingTemplate.convertAndSend("/topic/public",chatMessage);
    }

}

前端按照目标路径自行修改即可

function onConnected() {
    // 订阅群发主题
    stompClient.subscribe('/exchange/WebSocket/*',onMessageReceived)

    // 订阅一对一主题，即通过用户名等唯一性标识拼接到订阅主题地址
    stompClient.subscribe('/topic/single' + username,onMessageReceived)

    stompClient.send("/app/chat/addUser",
        {},
        JSON.stringify({sender: username, type: 'JOIN'})
    )

    connectingElement.classList.add('hidden');
}

集群 / 负载均衡简单看法

对于集群和负载均衡，我们无法通过简单的基于本机内存的 Stomp 实现，这时通过 RabbitMq 这种消息中间件就能很好的实现。当我们的 websocket 服务横向扩展时，因为我们是通过代理订阅 Mq 的主题，这样只要 Mq 主题上出现消息，订阅这个主题的客户端就会自然的收到消息推送。我们的代码也不需要做太大的变动，可以通过服务加入到注册中心，这样网关就能根据服务名负载均衡的去与 websocket 服务建立连接。

项目代码

最后附上所有的代码地址：https://gitee.com/ahang-gitee/websocket

若文章有误，欢迎评论区指正。

Spring WebSocket 结合 RabbitMq

Java相关栏目本月热门文章