WebSocket的那些事(5-Spring STOMP支持之连接外部消息代理)

这篇具有很好参考价值的文章主要介绍了WebSocket的那些事(5-Spring STOMP支持之连接外部消息代理)。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

一、序言

上节我们在 WebSocket的那些事(4-Spring中的STOMP支持详解) 中详细说明了通过Spring内置消息代理结合STOMP子协议进行Websocket通信,以及相关注解的使用及原理。

但是Spring内置消息代理会有一些限制,比如只支持STOMP协议的一部分命令,像acksreceipts命令都是不支持的,还有由于内置消息代理把消息存储在内存,当应用不可用时,客户端也就订阅不到到后台推送的消息。

这节我们将会使用支持STOMP协议的外部消息代理(RabbitMQ)进行Websocket通信。


二、开启RabbitMQ外部消息代理

服务端路由发送消息以及客户端订阅消息都要通过STOMP协议与RabbitMQ进行交互,由于RabbitMQ默认没有启动STOMP插件,因此我们需要先启用该插件。

rabbitmq-plugins enable rabbitmq_stomp

启动该插件后,RabbitMQ中STOMP适配器默认会监听61613端口,如果是云服务器,需要把该端口在安全组中放开。

关于该插件说明请参考:RabbitMQ中STOMP插件说明。


三、代码示例

我们在 WebSocket的那些事(4-Spring中的STOMP支持详解)中写了一个简单的聊天Demo示例,下面我们对该聊天Demo示例进行改造,将Spring内置消息代理替换成RabbitMQ外部消息代理。

1、Maven依赖项

服务端和客户端与外部消息代理都是通过TCP进行通信,Spring底层默认使用的是NettyReactor,因此需要引入相关依赖项。

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-reactor-netty</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-thymeleaf</artifactId>
</dependency>

2、相关实体

(1) 请求消息参数

@Data
public class WebSocketMsgDTO {

	private String name;

	private String content;
}


(2) 响应消息内容

@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class WebSocketMsgVO {

	private String content;
}

(3) 自定义认证用户信息

@Data
@AllArgsConstructor
@NoArgsConstructor
public class StompAuthenticatedUser implements Principal {

	/**
	 * 用户唯一ID
	 */
	private String userId;

	/**
	 * 用户昵称
	 */
	private String nickName;

	/**
	 * 用于指定用户消息推送的标识
	 * @return
	 */
	@Override
	public String getName() {
		return this.userId;
	}

}

3、自定义用户认证拦截器

@Slf4j
public class UserAuthenticationChannelInterceptor implements ChannelInterceptor {

	private static final String USER_ID = "User-ID";
	private static final String USER_NAME = "User-Name";

	@Override
	public Message<?> preSend(Message<?> message, MessageChannel channel) {
		StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
		// 如果是连接请求,记录userId
		if (StompCommand.CONNECT.equals(accessor.getCommand())) {
			String userID = accessor.getFirstNativeHeader(USER_ID);
			String username = accessor.getFirstNativeHeader(USER_NAME);

			log.info("Stomp User-Related headers found, userID: {}, username:{}", userID, username);
			accessor.setUser(new StompAuthenticatedUser(userID, username));
		}

		return message;
	}

}

4、Websocket外部消息代理配置

Spring中与外部消息代理通信的中间方被称之为Broker Relay,它会维护一个系统共享的单一TCP连接和外部消息代理进行通信,该TCP连接仅仅适用于服务端,用来发送消息,而不是接收消息,通过Broker RelaysystemLoginsystemPasscode属性可以设置该连接的认证信息。

Broker Relay也会为每个连接的Websocket客户端创建一个TCP连接,该连接用来接收消息,通过clientLoginclientPasscode属性可以设置连接的认证信息。

/**
 * Websocket连接外部消息代理配置
 * @author Nick Liu
 * @date 2023/9/6
 */
@Configuration
@EnableWebSocketMessageBroker
public class WebsocketExternalMessageBrokerConfig implements WebSocketMessageBrokerConfigurer {

	@Override
	public void configureClientInboundChannel(ChannelRegistration registration) {
		// 拦截器配置
		registration.interceptors(new UserAuthenticationChannelInterceptor());
	}

	@Override
	public void registerStompEndpoints(StompEndpointRegistry registry) {
		registry.addEndpoint("/websocket") // WebSocket握手端口
			.addInterceptors(new HttpSessionHandshakeInterceptor())
			.setAllowedOriginPatterns("*") // 设置跨域
			.withSockJS(); // 开启SockJS回退机制
	}

	@Override
	public void configureMessageBroker(MessageBrokerRegistry registry) {
		registry.setApplicationDestinationPrefixes("/app") // 发送到服务端目的地前缀
			.enableStompBrokerRelay("/topic") // 开启外部消息代理,指定消息订阅前缀
			.setRelayHost("localhost") // 外部消息代理Host
			.setRelayPort(61613) // 外部消息代理STOMP端口
			.setSystemLogin("admin")  // 共享系统连接用户名,该连接主要用来发送消息
			.setSystemPasscode("admin") // 共享系统连接密码,该连接主要用来发送消息
			.setClientLogin("admin") // 客户端连接用户名,该连接主要用来接收消息
			.setClientPasscode("admin") // 客户端连接密码,该连接主要用来接收消息
			.setVirtualHost("/stomp"); // RabbitMQ虚拟主机
	}
}

备注:我们可以为服务端与客户端的连接设置不同的用户,针对客户端连接用户进行权限管控,保证系统的安全性,在这里为了方便测试我们统一用一个用户。

5、ChatController

STOMP协议并没有规定消息代理必须支持哪种类型的Destinations(目的地),但是RabbitMQ STOMP适配器只支持一些指定的目的地类型,如下图:
WebSocket的那些事(5-Spring STOMP支持之连接外部消息代理),WebSocket,websocket,spring,网络协议

  • /exchange:指定交换机和路由key,发送和订阅来自队列的消息。
  • /queue:发送和订阅受STOMP网关管理的队列的消息,最多只有一个订阅者能到消息。
  • /amq/queue:发送和订阅不受STOMP网关管理的队列的消息。
  • /topic:发送和订阅来自临时或者持久Topic的消息,多个订阅者都能接收到消息。
  • /temp-queue/:发送和订阅来自临时队列的消息。

参考文档见:RabbitMQ中STOMP插件说明。

在下面的示例中,我们选用了/topic的开头的消息发送和订阅前缀,目的地格式只能为/topic/{routing-key}routing-key不能有斜杠,否则会报错。

@Slf4j
@Controller
@RequiredArgsConstructor
public class ChatController {

	private final SimpUserRegistry simpUserRegistry;
	private final SimpMessagingTemplate simpMessagingTemplate;

	/**
	 * 模板引擎为Thymeleaf,需要加上spring-boot-starter-thymeleaf依赖,
	 * @return
	 */
	@GetMapping("/page/chat")
	public ModelAndView turnToChatPage() {
		return new ModelAndView("chat");
	}

	/**
	 * 群聊消息处理
	 * 这里我们通过@SendTo注解指定消息目的地为"/topic/chat/group",如果不加该注解则会自动发送到"/topic" + "/chat/group"
	 * @param webSocketMsgDTO 请求参数,消息处理器会自动将JSON字符串转换为对象
	 * @return 消息内容,方法返回值将会广播给所有订阅"/topic/chat/group"的客户端
	 */
	@MessageMapping("/chat/group")
	@SendTo("/topic/chat-group")
	public WebSocketMsgVO groupChat(WebSocketMsgDTO webSocketMsgDTO) {
		log.info("Group chat message received: {}", FastJsonUtils.toJsonString(webSocketMsgDTO));
		String content = String.format("来自[%s]的群聊消息: %s", webSocketMsgDTO.getName(), webSocketMsgDTO.getContent());
		return WebSocketMsgVO.builder().content(content).build();
	}

	/**
	 * 私聊消息处理
	 * 这里我们通过@SendToUser注解指定消息目的地为"/topic/chat/private",发送目的地默认会拼接上"/user/"前缀
	 * 实际发送目的地为"/user/topic/chat/private"
	 * @param webSocketMsgDTO 请求参数,消息处理器会自动将JSON字符串转换为对象
	 * @return 消息内容,方法返回值将会基于SessionID单播给指定用户
	 */
	@MessageMapping("/chat/private")
	@SendToUser("/topic/chat-private")
	public WebSocketMsgVO privateChat(WebSocketMsgDTO webSocketMsgDTO) {
		log.info("Private chat message received: {}", FastJsonUtils.toJsonString(webSocketMsgDTO));
		String content = "私聊消息回复:" + webSocketMsgDTO.getContent();
		return WebSocketMsgVO.builder().content(content).build();
	}

	/**
	 * 定时消息推送,这里我们会列举所有在线的用户,然后单播给指定用户。
	 * 通过SimpMessagingTemplate实例可以在任何地方推送消息。
	 */
	@Scheduled(fixedRate = 10 * 1000)
	public void pushMessageAtFixedRate() {
		log.info("当前在线人数: {}", simpUserRegistry.getUserCount());
		if (simpUserRegistry.getUserCount() <= 0) {
			return;
		}

		// 这里的Principal为StompAuthenticatedUser实例
		Set<StompAuthenticatedUser> users = simpUserRegistry.getUsers().stream()
			.map(simpUser -> StompAuthenticatedUser.class.cast(simpUser.getPrincipal()))
			.collect(Collectors.toSet());

		users.forEach(authenticatedUser -> {
			String userId = authenticatedUser.getUserId();
			String nickName = authenticatedUser.getNickName();
			WebSocketMsgVO webSocketMsgVO = new WebSocketMsgVO();
			webSocketMsgVO.setContent(String.format("定时推送的私聊消息, 接收人: %s, 时间: %s", nickName, LocalDateTime.now()));

			log.info("开始推送消息给指定用户, userId: {}, 消息内容:{}", userId, FastJsonUtils.toJsonString(webSocketMsgVO));
			simpMessagingTemplate.convertAndSendToUser(userId, "/topic/chat-push", webSocketMsgVO);
		});
	}

}

6、前端页面chat.html

<!DOCTYPE html>
<html lang="en" xmlns:th="http://www.thymeleaf.org">
<head>
    <meta charset="UTF-8">
    <title>chat</title>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/sockjs-client/1.6.1/sockjs.min.js"></script>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/stomp.js/2.3.3/stomp.min.js"></script>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/jquery/3.6.4/jquery.min.js"></script>
    <style>
        #mainWrapper {
            width: 600px;
            margin: auto;
        }
    </style>
</head>
<body>
<div id="mainWrapper">
    <div>
        <label for="username" style="margin-right: 5px">姓名:</label><input id="username" type="text"/>
    </div>
    <div id="msgWrapper">
        <p style="vertical-align: top">发送的消息:</p>
        <textarea id="msgSent" style="width: 600px;height: 100px"></textarea>
        <p style="vertical-align: top">收到的群聊消息:</p>
        <textarea id="groupMsgReceived" style="width: 600px;height: 100px"></textarea>
        <p style="vertical-align: top">收到的私聊消息:</p>
        <textarea id="privateMsgReceived" style="width: 600px;height: 200px"></textarea>
    </div>
    <div style="margin-top: 5px;">
        <button onclick="connect()">连接</button>
        <button onclick="sendGroupMessage()">发送群聊消息</button>
        <button onclick="sendPrivateMessage()">发送私聊消息</button>
        <button onclick="disconnect()">断开连接</button>
    </div>
</div>
<script type="text/javascript">
    $(() => {
        $('#msgSent').val('');
        $("#groupMsgReceived").val('');
        $("#privateMsgReceived").val('');
    });

    let stompClient = null;


    // 连接服务器
    const connect = () => {
        const header = {"User-ID": new Date().getTime().toString(), "User-Name": $('#username').val()};
        const ws = new SockJS('http://localhost:8080/websocket');
        stompClient = Stomp.over(ws);
        stompClient.connect(header, () => subscribeTopic());
    }

    // 订阅主题
    const subscribeTopic = () => {
        alert("连接成功!");

        // 订阅广播消息
        stompClient.subscribe('/topic/chat-group', function (message) {
                console.log(`Group message received : ${message.body}`);
                const resp = JSON.parse(message.body);
                const previousMsg = $("#groupMsgReceived").val();
                $("#groupMsgReceived").val(`${previousMsg}${resp.content}\n`);
            }
        );
        // 订阅单播消息
        stompClient.subscribe('/user/topic/chat-private', message => {
                console.log(`Private message received : ${message.body}`);
                const resp = JSON.parse(message.body);
                const previousMsg = $("#privateMsgReceived").val();
                $("#privateMsgReceived").val(`${previousMsg}${resp.content}\n`);
            }
        );
        // 订阅定时推送的单播消息
        stompClient.subscribe(`/user/topic/chat-push`, message => {
                console.log(`Private message received : ${message.body}`);
                const resp = JSON.parse(message.body);
                const previousMsg = $("#privateMsgReceived").val();
                $("#privateMsgReceived").val(`${previousMsg}${resp.content}\n`);
            }
        );
    };

    // 断连
    const disconnect = () => {
        stompClient.disconnect(() => {
            $("#msgReceived").val('Disconnected from WebSocket server');
        });
    }

    // 发送群聊消息
    const sendGroupMessage = () => {
        const msg = {name: $('#username').val(), content: $('#msgSent').val()};
        stompClient.send('/app/chat/group', {}, JSON.stringify(msg));
    }

    // 发送私聊消息
    const sendPrivateMessage = () => {
        const msg = {name: $('#username').val(), content: $('#msgSent').val()};
        stompClient.send('/app/chat/private', {}, JSON.stringify(msg));
    }
</script>
</body>
</html>

四、测试示例

1、群聊、私聊、后台定时推送测试

启动应用程序,日志打印显示系统连接建立成功,如下:

WebSocket的那些事(5-Spring STOMP支持之连接外部消息代理),WebSocket,websocket,spring,网络协议

打开浏览器访问http://localhost:8080/page/chat可进入聊天页,同时打开两个窗口访问。
WebSocket的那些事(5-Spring STOMP支持之连接外部消息代理),WebSocket,websocket,spring,网络协议


WebSocket的那些事(5-Spring STOMP支持之连接外部消息代理),WebSocket,websocket,spring,网络协议

2、登录RabbitMQ控制台查看队列信息

WebSocket的那些事(5-Spring STOMP支持之连接外部消息代理),WebSocket,websocket,spring,网络协议
可以看到所有消息都发送到了amq.topic交换机上(Topic类型), RabbitMQ会为每个连接的客户端创建3个队列。

因为我们在ChatController中定义了三个目的地,Routing Key分别是/topic/chat-group/topic/chat-private/topic/chat-push。群聊消息目的地/topic/chat-group绑定了两个队列,用于实现广播订阅,其它两个Routing Key分别绑定到了不同的队列上,实现唯一订阅。


五、结语

下一节我们将会详细说明RabbitMQ STOMP适配器支持的各种消息目的地类型的区别以及适用场景。

WebSocket的那些事(5-Spring STOMP支持之连接外部消息代理),WebSocket,websocket,spring,网络协议文章来源地址https://www.toymoban.com/news/detail-705408.html

到了这里,关于WebSocket的那些事(5-Spring STOMP支持之连接外部消息代理)的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • WebSocket的那些事(3-STOMP实操篇)

    上节中我们在 WebSocket的那些事(2-实操篇)中简单介绍了Spring中对于 WebSocket 的封装,并实现一个简单的服务端,这节我们将会结合 STOMP 子协议实现 WebSocket 通信。 WebSocket协议定义了两种消息类型(文本类型和二进制类型),但是消息内容却是未定义的,下面我们介绍一下

    2024年02月07日
    浏览(34)
  • WebSocket的那些事(6- RabbitMQ STOMP目的地详解)

    在上节 WebSocket的那些事(5-Spring STOMP支持之连接外部消息代理)中我们已经简单介绍了各种目的地类型,如下图: 这一节我们来详细探讨一下各种目的地类型的区别。 在 MESSAGE 报文中请求头 destination 的格式如下: 发布到默认交换机的消息目的地格式为: /queue/queue_name 。 发

    2024年02月08日
    浏览(40)
  • SpringBoot + WebSocket+STOMP指定推送消息

    前些天发现了一个巨牛的人工智能学习网站,通俗易懂,风趣幽默,忍不住分享一下给大家。点击跳转到网站。 本文将简单的描述SpringBoot + WebSocket+STOMP指定推送消息场景,不包含信息安全加密等,请勿用在生产环境。 JDK:11+ Maven: 3.5+ SpringBoot: 2.6+ stompjs@7.0.0 STOMP 是面向简

    2024年02月14日
    浏览(47)
  • uni-app + SpringBoot +stomp 支持websocket 打包app

    websocket 协议是在http 协议的基础上的升级,通过一次http 请求建立长连接,转而变为TCP 的全双工通信;而http 协议是一问一答的请求方式方式。 websocket-uni.js

    2024年02月11日
    浏览(49)
  • Nginx代理nginx.conf配置——反向代理(对WebSocket支持)

    基于Nginx代理nginx.conf配置——反向代理,如果要添加websocket支持,需要进行如下配置 在http中添加一下配置,添加对websocket支持 配置后重新运行nginx后,websocket即可成功代理。 注意防火墙端口是否开放、nginx如果使用docker,其对应的端口是否映射出来。

    2024年02月13日
    浏览(38)
  • Spring RabbitMQ那些事(3-消息可靠传输和订阅)

    在有些业务场景中,消息是不能丢的,比如 分布式事务资金动账 ,出账方扣款,那么入账方就一定要收款。以前写了一篇分布式事务的文章,里面的跨地区转账就是一个实际案例。 消息是有可能丢的,比如生产者在发送消息时 broker 服务挂了,消息没有来得及落盘,这时消

    2024年01月21日
    浏览(39)
  • Spring RabbitMQ那些事(2-两种方式实现延时消息订阅)

    业务开发中有很多延时操作的场景,比如最常见的 超时订单自动关闭 、 延时异步处理 ,我们常用的实现方式有: 定时任务轮询 (有延时)。 借助Redission的延时队列 。 Redis的key过期事件通知机制 (需开启key过期事件通知,对Redis有性能损耗)。 RocketMQ中定时消息推送 (支

    2024年02月04日
    浏览(39)
  • Spring RabbitMQ那些事(1-交换机配置&消息发送订阅实操)

    在上一节 RabbitMQ中的核心概念和交换机类型 中我们介绍了RabbitMQ中的一些核心概念,尤其是各种交换机的类型,接下来我们将具体讲解各种交换机的配置和消息订阅实操。 我们先上应用启动配置文件 application.yml ,如下: 备注:这里我们指定了 RabbitListenerContainerFactory 的类型

    2024年01月17日
    浏览(44)
  • WebSocket—STOMP详解(官方原版)

    WebSocket协议定义了两种类型的消息(文本和二进制),但其内容未作定义。该协议定义了一种机制,供客户端和服务器协商在WebSocket之上使用的子协议(即更高级别的消息传递协议),以定义各自可以发送何种消息、格式是什么、每个消息的内容等等。子协议的使用是可选的

    2024年02月04日
    浏览(38)
  • websocket + stomp + sockjs学习

    Spring WebSocket整合Stomp源码详解 PDF版本 Spring SpringBoot官方文档资料 spring5.1.9官方文档关于websocket的介绍 spring5.3.29官方文档关于websocket的介绍 WebSocket入门教程示例代码,代码地址已fork至本地gitee,原github代码地址,源老外的代码地址 [WebSocket入门]手把手搭建WebSocket多人在线聊天

    2024年02月12日
    浏览(32)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包