一、序言

上節(jié)我們在 WebSocket的那些事（4-Spring中的STOMP支持詳解） 中詳細(xì)說明了通過Spring內(nèi)置消息代理結(jié)合STOMP子協(xié)議進(jìn)行Websocket通信，以及相關(guān)注解的使用及原理。

但是Spring內(nèi)置消息代理會(huì)有一些限制，比如只支持STOMP協(xié)議的一部分命令，像acks、receipts命令都是不支持的，還有由于內(nèi)置消息代理把消息存儲在內(nèi)存，當(dāng)應(yīng)用不可用時(shí)，客戶端也就訂閱不到到后臺推送的消息。

這節(jié)我們將會(huì)使用支持STOMP協(xié)議的外部消息代理（RabbitMQ）進(jìn)行Websocket通信。

---

二、開啟RabbitMQ外部消息代理

服務(wù)端路由發(fā)送消息以及客戶端訂閱消息都要通過STOMP協(xié)議與RabbitMQ進(jìn)行交互，由于RabbitMQ默認(rèn)沒有啟動(dòng)STOMP插件，因此我們需要先啟用該插件。

rabbitmq-plugins enable rabbitmq_stomp

啟動(dòng)該插件后，RabbitMQ中STOMP適配器默認(rèn)會(huì)監(jiān)聽61613端口，如果是云服務(wù)器，需要把該端口在安全組中放開。

關(guān)于該插件說明請參考：RabbitMQ中STOMP插件說明。

---

三、代碼示例

我們在 WebSocket的那些事（4-Spring中的STOMP支持詳解）中寫了一個(gè)簡單的聊天Demo示例，下面我們對該聊天Demo示例進(jìn)行改造，將Spring內(nèi)置消息代理替換成RabbitMQ外部消息代理。

1、Maven依賴項(xiàng)

服務(wù)端和客戶端與外部消息代理都是通過TCP進(jìn)行通信，Spring底層默認(rèn)使用的是Netty和Reactor，因此需要引入相關(guān)依賴項(xiàng)。

<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、相關(guān)實(shí)體

(1) 請求消息參數(shù)

@Data
public class WebSocketMsgDTO {private String name;private String content;
}

(2) 響應(yīng)消息內(nèi)容

@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class WebSocketMsgVO {private String content;
}

(3) 自定義認(rèn)證用戶信息

@Data
@AllArgsConstructor
@NoArgsConstructor
public class StompAuthenticatedUser implements Principal {/*** 用戶唯一ID*/private String userId;/*** 用戶昵稱*/private String nickName;/*** 用于指定用戶消息推送的標(biāo)識* @return*/@Overridepublic String getName() {return this.userId;}}

3、自定義用戶認(rèn)證攔截器

@Slf4j
public class UserAuthenticationChannelInterceptor implements ChannelInterceptor {private static final String USER_ID = "User-ID";private static final String USER_NAME = "User-Name";@Overridepublic Message<?> preSend(Message<?> message, MessageChannel channel) {StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);// 如果是連接請求，記錄userIdif (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，它會(huì)維護(hù)一個(gè)系統(tǒng)共享的單一TCP連接和外部消息代理進(jìn)行通信，該TCP連接僅僅適用于服務(wù)端，用來發(fā)送消息，而不是接收消息，通過Broker Relay的systemLogin和systemPasscode屬性可以設(shè)置該連接的認(rèn)證信息。

Broker Relay也會(huì)為每個(gè)連接的Websocket客戶端創(chuàng)建一個(gè)TCP連接，該連接用來接收消息，通過clientLogin和clientPasscode屬性可以設(shè)置連接的認(rèn)證信息。

/*** Websocket連接外部消息代理配置* @author Nick Liu* @date 2023/9/6*/
@Configuration
@EnableWebSocketMessageBroker
public class WebsocketExternalMessageBrokerConfig implements WebSocketMessageBrokerConfigurer {@Overridepublic void configureClientInboundChannel(ChannelRegistration registration) {// 攔截器配置registration.interceptors(new UserAuthenticationChannelInterceptor());}@Overridepublic void registerStompEndpoints(StompEndpointRegistry registry) {registry.addEndpoint("/websocket") // WebSocket握手端口.addInterceptors(new HttpSessionHandshakeInterceptor()).setAllowedOriginPatterns("*") // 設(shè)置跨域.withSockJS(); // 開啟SockJS回退機(jī)制}@Overridepublic void configureMessageBroker(MessageBrokerRegistry registry) {registry.setApplicationDestinationPrefixes("/app") // 發(fā)送到服務(wù)端目的地前綴.enableStompBrokerRelay("/topic") // 開啟外部消息代理，指定消息訂閱前綴.setRelayHost("localhost") // 外部消息代理Host.setRelayPort(61613) // 外部消息代理STOMP端口.setSystemLogin("admin")  // 共享系統(tǒng)連接用戶名，該連接主要用來發(fā)送消息.setSystemPasscode("admin") // 共享系統(tǒng)連接密碼，該連接主要用來發(fā)送消息.setClientLogin("admin") // 客戶端連接用戶名，該連接主要用來接收消息.setClientPasscode("admin") // 客戶端連接密碼，該連接主要用來接收消息.setVirtualHost("/stomp"); // RabbitMQ虛擬主機(jī)}
}

備注：我們可以為服務(wù)端與客戶端的連接設(shè)置不同的用戶，針對客戶端連接用戶進(jìn)行權(quán)限管控，保證系統(tǒng)的安全性，在這里為了方便測試我們統(tǒng)一用一個(gè)用戶。

5、ChatController

STOMP協(xié)議并沒有規(guī)定消息代理必須支持哪種類型的Destinations(目的地)，但是RabbitMQ STOMP適配器只支持一些指定的目的地類型，如下圖：

• /exchange：指定交換機(jī)和路由key，發(fā)送和訂閱來自隊(duì)列的消息。
• /queue：發(fā)送和訂閱受STOMP網(wǎng)關(guān)管理的隊(duì)列的消息，最多只有一個(gè)訂閱者能到消息。
• /amq/queue：發(fā)送和訂閱不受STOMP網(wǎng)關(guān)管理的隊(duì)列的消息。
• /topic：發(fā)送和訂閱來自臨時(shí)或者持久Topic的消息，多個(gè)訂閱者都能接收到消息。
• /temp-queue/：發(fā)送和訂閱來自臨時(shí)隊(duì)列的消息。

參考文檔見：RabbitMQ中STOMP插件說明。

在下面的示例中，我們選用了/topic的開頭的消息發(fā)送和訂閱前綴，目的地格式只能為/topic/{routing-key}，routing-key不能有斜杠，否則會(huì)報(bào)錯(cuò)。

@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"，如果不加該注解則會(huì)自動(dòng)發(fā)送到"/topic" + "/chat/group"* @param webSocketMsgDTO 請求參數(shù)，消息處理器會(huì)自動(dòng)將JSON字符串轉(zhuǎn)換為對象* @return 消息內(nèi)容，方法返回值將會(huì)廣播給所有訂閱"/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"，發(fā)送目的地默認(rèn)會(huì)拼接上"/user/"前綴* 實(shí)際發(fā)送目的地為"/user/topic/chat/private"* @param webSocketMsgDTO 請求參數(shù)，消息處理器會(huì)自動(dòng)將JSON字符串轉(zhuǎn)換為對象* @return 消息內(nèi)容，方法返回值將會(huì)基于SessionID單播給指定用戶*/@MessageMapping("/chat/private")@SendToUser("/topic/chat-private")public WebSocketMsgVO privateChat(WebSocketMsgDTO webSocketMsgDTO) {log.info("Private chat message received: {}", FastJsonUtils.toJsonString(webSocketMsgDTO));String content = "私聊消息回復(fù)：" + webSocketMsgDTO.getContent();return WebSocketMsgVO.builder().content(content).build();}/*** 定時(shí)消息推送，這里我們會(huì)列舉所有在線的用戶，然后單播給指定用戶。* 通過SimpMessagingTemplate實(shí)例可以在任何地方推送消息。*/@Scheduled(fixedRate = 10 * 1000)public void pushMessageAtFixedRate() {log.info("當(dāng)前在線人數(shù): {}", simpUserRegistry.getUserCount());if (simpUserRegistry.getUserCount() <= 0) {return;}// 這里的Principal為StompAuthenticatedUser實(shí)例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("定時(shí)推送的私聊消息, 接收人: %s, 時(shí)間: %s", nickName, LocalDateTime.now()));log.info("開始推送消息給指定用戶, userId: {}, 消息內(nèi)容:{}", 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">發(fā)送的消息:</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()">發(fā)送群聊消息</button><button onclick="sendPrivateMessage()">發(fā)送私聊消息</button><button onclick="disconnect()">斷開連接</button></div>
</div>
<script type="text/javascript">$(() => {$('#msgSent').val('');$("#groupMsgReceived").val('');$("#privateMsgReceived").val('');});let stompClient = null;// 連接服務(wù)器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`);});// 訂閱定時(shí)推送的單播消息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');});}// 發(fā)送群聊消息const sendGroupMessage = () => {const msg = {name: $('#username').val(), content: $('#msgSent').val()};stompClient.send('/app/chat/group', {}, JSON.stringify(msg));}// 發(fā)送私聊消息const sendPrivateMessage = () => {const msg = {name: $('#username').val(), content: $('#msgSent').val()};stompClient.send('/app/chat/private', {}, JSON.stringify(msg));}
</script>
</body>
</html>

---

四、測試示例

1、群聊、私聊、后臺定時(shí)推送測試

啟動(dòng)應(yīng)用程序，日志打印顯示系統(tǒng)連接建立成功，如下：

打開瀏覽器訪問http://localhost:8080/page/chat可進(jìn)入聊天頁，同時(shí)打開兩個(gè)窗口訪問。

---

2、登錄RabbitMQ控制臺查看隊(duì)列信息

可以看到所有消息都發(fā)送到了amq.topic交換機(jī)上（Topic類型）， RabbitMQ會(huì)為每個(gè)連接的客戶端創(chuàng)建3個(gè)隊(duì)列。

因?yàn)槲覀冊?code>ChatController中定義了三個(gè)目的地，Routing Key分別是/topic/chat-group、/topic/chat-private、/topic/chat-push。群聊消息目的地/topic/chat-group綁定了兩個(gè)隊(duì)列，用于實(shí)現(xiàn)廣播訂閱，其它兩個(gè)Routing Key分別綁定到了不同的隊(duì)列上，實(shí)現(xiàn)唯一訂閱。

---

五、結(jié)語

下一節(jié)我們將會(huì)詳細(xì)說明RabbitMQ STOMP適配器支持的各種消息目的地類型的區(qū)別以及適用場景。