Spring Boot中使用WebSocket總結（二）：向指定使用者傳送WebSocket訊息並處理對方不線上的情況

WebSocket SpringBoot · 發表 2018-11-21 08:27:19

摘要：在上一篇文章（ www.zifangsky.cn/1355.html ）中我介紹了在Spring專案中使用WebSocket的幾種實現方式。但是，上篇文章中只介紹了服務端採用廣播模式給所有客戶端傳送訊息，然而我們有時需要服務端給指定使用者的客戶端傳送訊息（比如：傳送Web通知、實時列印使用者...

在上一篇文章（ ofollow,noindex">www.zifangsky.cn/1355.html ）中我介紹了在Spring專案中使用Socket/">WebSocket的幾種實現方式。但是，上篇文章中只介紹了服務端採用廣播模式給所有客戶端傳送訊息，然而我們有時需要服務端給指定使用者的客戶端傳送訊息（比如：傳送Web通知、實時列印使用者任務的日誌、兩個使用者點對點聊天等）。

關於服務端如何給指定使用者的客戶端傳送訊息，一般可以通過以下三種方案來實現：

方案一 ：WebSocket使用“Java提供的@ServerEndpoint註解”實現或者使用“Spring低層級API”實現，在建立連線時從 HttpSession 中獲取使用者登入後的使用者名稱，然後把“ 使用者名稱+該WebSocket連線 ”儲存到 ConcurrentHashMap 。給指定使用者傳送訊息，只需要根據接收者的使用者名稱獲取對方已經建立的WebSocket連線，接著給他傳送訊息即可。
方案二 ：在頁面的監聽路徑前面動態添加當前登入的“ 使用者ID/使用者名稱 ”，這樣給指定使用者傳送訊息，只需要傳送廣播訊息到監聽了前面那個路徑的客戶端即可。
方案三 ：這種方案類似於方案一。使用Spring的高階API實現WebSocket，然後自定義 HandshakeHandler 類並重寫 determineUser 方法，其目的是為了在建立連線時使用使用者登入後的使用者名稱作為此次WebSocket的憑證，最後我們就可以使用 messagingTemplate.convertAndSendToUser 方法給指定使用者傳送訊息了。

注：本篇文章的完整原始碼可以參考： github.com/zifangsky/W…

使用SimpMessagingTemplate傳送訊息

使用 org.springframework.messaging.simp.SimpMessagingTemplate 類可以在服務端的任意地方給客戶端傳送訊息。此外，在我們配置Spring支援STOMP後 SimpMessagingTemplate 類就會被自動裝配到Spring的上下文中，因此我們只需要在想要使用的地方使用 @Autowired 註解注入SimpMessagingTemplate即可使用。

需要說明的是， SimpMessagingTemplate 類有兩個重要的方法，它們分別是：

public void convertAndSend(D destination, Object payload) ：給監聽了路徑 destination 的所有客戶端傳送訊息 payload
public void convertAndSendToUser(String user, String destination, Object payload) ：給監聽了路徑 destination 的使用者 user 傳送訊息 payload

一個簡單示例：

package cn.zifangsky.stompwebsocket.controller;

import cn.zifangsky.stompwebsocket.model.websocket.Greeting;
import cn.zifangsky.stompwebsocket.model.websocket.HelloMessage;
import cn.zifangsky.stompwebsocket.service.RedisService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.messaging.simp.user.SimpUserRegistry;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import javax.annotation.Resource;

/**
 * 測試{@link org.springframework.messaging.simp.SimpMessagingTemplate}類的基本用法
 * @author zifangsky
 * @date 2018/10/10
 * @since 1.0.0
 */
@Controller
@RequestMapping(("/wsTemplate"))
public class MessageTemplateController {
private final Logger logger = LoggerFactory.getLogger(getClass());

@Autowired
private SimpMessagingTemplate messagingTemplate;

@Autowired
private SimpUserRegistry userRegistry;

@Resource(name = "redisServiceImpl")
private RedisService redisService;

/**
* 簡單測試SimpMessagingTemplate的用法
*/
@PostMapping("/greeting")
@ResponseBody
public String greeting(@RequestBody Greeting greeting) {
this.messagingTemplate.convertAndSend("/topic/greeting", new HelloMessage("Hello," + greeting.getName() + "!"));

return "ok";
}

}
複製程式碼

很顯然，這裡傳送的地址是上篇文章中最後那個示例監聽的地址，在客戶端頁面建立連線後，我們使用 Postman 請求一下上面這個方法，效果如下：

然後我們可以發現頁面中也收到訊息了：

向指定使用者傳送WebSocket訊息並處理對方不線上的情況

給指定使用者傳送訊息：

如果接收者線上，則直接傳送訊息；
否則將訊息儲存到redis，等使用者上線後主動拉取未讀訊息。

（1）自定義HandshakeInterceptor，用於禁止未登入使用者連線WebSocket：

package cn.zifangsky.stompwebsocket.interceptor.websocket;

import cn.zifangsky.stompwebsocket.common.Constants;
import cn.zifangsky.stompwebsocket.common.SpringContextUtils;
import cn.zifangsky.stompwebsocket.model.User;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.server.ServerHttpRequest;
import org.springframework.http.server.ServerHttpResponse;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.server.HandshakeInterceptor;

import javax.servlet.http.HttpSession;
import java.text.MessageFormat;
import java.util.Map;

/**
 * 自定義{@link org.springframework.web.socket.server.HandshakeInterceptor}，實現“需要登入才允許連線WebSocket”
 *
 * @author zifangsky
 * @date 2018/10/11
 * @since 1.0.0
 */
@Component
public class AuthHandshakeInterceptor implements HandshakeInterceptor {
private final Logger logger = LoggerFactory.getLogger(getClass());

@Override
public boolean beforeHandshake(ServerHttpRequest serverHttpRequest, ServerHttpResponse serverHttpResponse, WebSocketHandler webSocketHandler, Map<String, Object> map) throws Exception {
HttpSession session = SpringContextUtils.getSession();
User loginUser = (User) session.getAttribute(Constants.SESSION_USER);

if(loginUser != null){
logger.debug(MessageFormat.format("使用者{0}請求建立WebSocket連線", loginUser.getUsername()));
return true;
}else{
logger.error("未登入系統，禁止連線WebSocket");
return false;
}

}

@Override
public void afterHandshake(ServerHttpRequest serverHttpRequest, ServerHttpResponse serverHttpResponse, WebSocketHandler webSocketHandler, Exception e) {

}

}
複製程式碼

（2）自定義HandshakeHandler，用於在建立WebSocket的時候使用自定義的Principal：

package cn.zifangsky.stompwebsocket.interceptor.websocket;

import cn.zifangsky.stompwebsocket.common.Constants;
import cn.zifangsky.stompwebsocket.common.SpringContextUtils;
import cn.zifangsky.stompwebsocket.model.User;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.server.ServerHttpRequest;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.server.support.DefaultHandshakeHandler;

import javax.servlet.http.HttpSession;
import java.security.Principal;
import java.text.MessageFormat;
import java.util.Map;

/**
 * 自定義{@link org.springframework.web.socket.server.support.DefaultHandshakeHandler}，實現“生成自定義的{@link java.security.Principal}”
 *
 * @author zifangsky
 * @date 2018/10/11
 * @since 1.0.0
 */
@Component
public class MyHandshakeHandler extends DefaultHandshakeHandler{
private final Logger logger = LoggerFactory.getLogger(getClass());

@Override
protected Principal determineUser(ServerHttpRequest request, WebSocketHandler wsHandler, Map<String, Object> attributes) {
HttpSession session = SpringContextUtils.getSession();
User loginUser = (User) session.getAttribute(Constants.SESSION_USER);

if(loginUser != null){
logger.debug(MessageFormat.format("WebSocket連線開始建立Principal，使用者：{0}", loginUser.getUsername()));
return new MyPrincipal(loginUser.getUsername());
}else{
logger.error("未登入系統，禁止連線WebSocket");
return null;
}
}

}
複製程式碼

相應地，這裡的 MyPrincipal 繼承了 java.security.Principal 類：

package cn.zifangsky.stompwebsocket.interceptor.websocket;

import java.security.Principal;

/**
 * 自定義{@link java.security.Principal}
 *
 * @author zifangsky
 * @date 2018/10/11
 * @since 1.0.0
 */
public class MyPrincipal implements Principal {
private String loginName;

public MyPrincipal(String loginName) {
this.loginName = loginName;
}

@Override
public String getName() {
return loginName;
}
}
複製程式碼

（3）自定義ChannelInterceptor，用於在使用者斷開連線的時候記錄日誌：

package cn.zifangsky.stompwebsocket.interceptor.websocket;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.stereotype.Component;

import java.security.Principal;
import java.text.MessageFormat;

/**
 * 自定義{@link org.springframework.messaging.support.ChannelInterceptor}，實現斷開連線的處理
 *
 * @author zifangsky
 * @date 2018/10/10
 * @since 1.0.0
 */
@Component
public class MyChannelInterceptor implements ChannelInterceptor{
private final Logger logger = LoggerFactory.getLogger(getClass());

@Override
public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex) {
StompHeaderAccessor accessor = StompHeaderAccessor.wrap(message);
StompCommand command = accessor.getCommand();

//使用者已經斷開連線
if(StompCommand.DISCONNECT.equals(command)){
String user = "";
Principal principal = accessor.getUser();
if(principal != null && StringUtils.isNoneBlank(principal.getName())){
user = principal.getName();
}else{
user = accessor.getSessionId();
}

logger.debug(MessageFormat.format("使用者{0}的WebSocket連線已經斷開", user));
}
}

}
複製程式碼

（4）WebSocket相關的完整配置：

package cn.zifangsky.stompwebsocket.config;

import cn.zifangsky.stompwebsocket.interceptor.websocket.AuthHandshakeInterceptor;
import cn.zifangsky.stompwebsocket.interceptor.websocket.MyChannelInterceptor;
import cn.zifangsky.stompwebsocket.interceptor.websocket.MyHandshakeHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.config.ChannelRegistration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;

/**
 * WebSocket相關配置
 *
 * @author zifangsky
 * @date 2018/9/30
 * @since 1.0.0
 */
@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer{
@Autowired
private AuthHandshakeInterceptor authHandshakeInterceptor;

@Autowired
private MyHandshakeHandler myHandshakeHandler;

@Autowired
private MyChannelInterceptor myChannelInterceptor;

@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
registry.addEndpoint("/stomp-websocket").withSockJS();

registry.addEndpoint("/chat-websocket")
.addInterceptors(authHandshakeInterceptor)
.setHandshakeHandler(myHandshakeHandler)
.withSockJS();
}

@Override
public void configureMessageBroker(MessageBrokerRegistry registry) {
//客戶端需要把訊息傳送到/message/xxx地址
registry.setApplicationDestinationPrefixes("/message");
//服務端廣播訊息的路徑字首，客戶端需要相應訂閱/topic/yyy這個地址的訊息
registry.enableSimpleBroker("/topic");
}

@Override
public void configureClientInboundChannel(ChannelRegistration registration) {
registration.interceptors(myChannelInterceptor);
}

}
複製程式碼

（5）Controller中的訊息處理如下：

package cn.zifangsky.stompwebsocket.controller;

import cn.zifangsky.stompwebsocket.common.Constants;
import cn.zifangsky.stompwebsocket.common.SpringContextUtils;
import cn.zifangsky.stompwebsocket.enums.ExpireEnum;
import cn.zifangsky.stompwebsocket.model.User;
import cn.zifangsky.stompwebsocket.model.websocket.Greeting;
import cn.zifangsky.stompwebsocket.model.websocket.HelloMessage;
import cn.zifangsky.stompwebsocket.service.RedisService;
import cn.zifangsky.stompwebsocket.utils.JsonUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.messaging.simp.user.SimpUser;
import org.springframework.messaging.simp.user.SimpUserRegistry;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import javax.annotation.Resource;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpSession;
import java.text.MessageFormat;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 測試{@link org.springframework.messaging.simp.SimpMessagingTemplate}類的基本用法
 * @author zifangsky
 * @date 2018/10/10
 * @since 1.0.0
 */
@Controller
@RequestMapping(("/wsTemplate"))
public class MessageTemplateController {
private final Logger logger = LoggerFactory.getLogger(getClass());

@Autowired
private SimpMessagingTemplate messagingTemplate;

@Autowired
private SimpUserRegistry userRegistry;

@Resource(name = "redisServiceImpl")
private RedisService redisService;

/**
* 簡單測試SimpMessagingTemplate的用法
*/
@PostMapping("/greeting")
@ResponseBody
public String greeting(@RequestBody Greeting greeting) {
this.messagingTemplate.convertAndSend("/topic/greeting", new HelloMessage("Hello," + greeting.getName() + "!"));

return "ok";
}

/**
* 給指定使用者傳送WebSocket訊息
*/
@PostMapping("/sendToUser")
@ResponseBody
public String chat(HttpServletRequest request) {
//訊息接收者
String receiver = request.getParameter("receiver");
//訊息內容
String msg = request.getParameter("msg");
HttpSession session = SpringContextUtils.getSession();
User loginUser = (User) session.getAttribute(Constants.SESSION_USER);

HelloMessage resultData = new HelloMessage(MessageFormat.format("{0} say: {1}", loginUser.getUsername(), msg));
this.sendToUser(loginUser.getUsername(), receiver, "/topic/reply", JsonUtils.toJson(resultData));

return "ok";
}

/**
* 給指定使用者傳送訊息，並處理接收者不線上的情況
* @param sender 訊息傳送者
* @param receiver 訊息接收者
* @param destination 目的地
* @param payload 訊息正文
*/
private void sendToUser(String sender, String receiver, String destination, String payload){
SimpUser simpUser = userRegistry.getUser(receiver);

//如果接收者存在，則傳送訊息
if(simpUser != null && StringUtils.isNoneBlank(simpUser.getName())){
this.messagingTemplate.convertAndSendToUser(receiver, destination, payload);
}
//否則將訊息儲存到redis，等使用者上線後主動拉取未讀訊息
else{
//儲存訊息的Redis列表名
String listKey = Constants.REDIS_UNREAD_MSG_PREFIX + receiver + ":" + destination;
logger.info(MessageFormat.format("訊息接收者{0}還未建立WebSocket連線，{1}傳送的訊息【{2}】將被儲存到Redis的【{3}】列表中", receiver, sender, payload, listKey));

//儲存訊息到Redis中
redisService.addToListRight(listKey, ExpireEnum.UNREAD_MSG, payload);
}

}


/**
* 拉取指定監聽路徑的未讀的WebSocket訊息
* @param destination 指定監聽路徑
* @return java.util.Map<java.lang.String,java.lang.Object>
*/
@PostMapping("/pullUnreadMessage")
@ResponseBody
public Map<String, Object> pullUnreadMessage(String destination){
Map<String, Object> result = new HashMap<>();
try {
HttpSession session = SpringContextUtils.getSession();
//當前登入使用者
User loginUser = (User) session.getAttribute(Constants.SESSION_USER);

//儲存訊息的Redis列表名
String listKey = Constants.REDIS_UNREAD_MSG_PREFIX + loginUser.getUsername() + ":" + destination;
//從Redis中拉取所有未讀訊息
List<Object> messageList = redisService.rangeList(listKey, 0, -1);

result.put("code", "200");
if(messageList !=null && messageList.size() > 0){
//刪除Redis中的這個未讀訊息列表
redisService.delete(listKey);
//將資料新增到返回集，供前臺頁面展示
result.put("result", messageList);
}
}catch (Exception e){
result.put("code", "500");
result.put("msg", e.getMessage());
}

return result;
}

}
複製程式碼

注：這裡對應的幾個Redis操作的方法如下：

@Override
public boolean delete(String key) {
return redisTemplate.delete(key);
}

@Override
public void addToListLeft(String listKey, ExpireEnum expireEnum, Object... values) {
//繫結操作
BoundListOperations<String, Object> boundValueOperations = redisTemplate.boundListOps(listKey);
//插入資料
boundValueOperations.leftPushAll(values);
//設定過期時間
boundValueOperations.expire(expireEnum.getTime(),expireEnum.getTimeUnit());
}

@Override
public void addToListRight(String listKey, ExpireEnum expireEnum, Object... values) {
//繫結操作
BoundListOperations<String, Object> boundValueOperations = redisTemplate.boundListOps(listKey);
//插入資料
boundValueOperations.rightPushAll(values);
//設定過期時間
boundValueOperations.expire(expireEnum.getTime(),expireEnum.getTimeUnit());
}

@Override
public List<Object> rangeList(String listKey, long start, long end) {
//繫結操作
BoundListOperations<String, Object> boundValueOperations = redisTemplate.boundListOps(listKey);
//查詢資料
return boundValueOperations.range(start, end);
}
複製程式碼

（6）示例頁面：

<html xmlns:th="http://www.thymeleaf.org">
<head>
<meta content="text/html;charset=UTF-8"/>
<meta http-equiv="Content-Type" content="text/html; charset=utf-8"/>
<meta http-equiv="X-UA-Compatible" content="IE=edge"/>
<meta name="viewport" content="width=device-width, initial-scale=1"/>
<title>Chat With STOMP Message</title>
<script src="https://cdnjs.cloudflare.com/ajax/libs/jquery/3.3.1/jquery.min.js"></script>
<script src="https://cdnjs.cloudflare.com/ajax/libs/sockjs-client/1.1.4/sockjs.min.js"></script>
<script src="https://cdnjs.cloudflare.com/ajax/libs/stomp.js/2.3.3/stomp.min.js"></script>
<script th:src="@{/layui/layui.js}"></script>
<script th:src="@{/layui/lay/modules/layer.js}"></script>
<link th:href="@{/layui/css/layui.css}" rel="stylesheet">
<link th:href="@{/layui/css/modules/layer/default/layer.css}" rel="stylesheet">
<link th:href="@{/css/style.css}" rel="stylesheet">
<style type="text/css">
#connect-container {
margin: 0 auto;
width: 400px;
}

#connect-container div {
padding: 5px;
margin: 0 7px 10px 0;
}

.message input {
padding: 5px;
margin: 0 7px 10px 0;
}

.layui-btn {
display: inline-block;
}
</style>
<script type="text/javascript">
var stompClient = null;

$(function () {
var target = $("#target");
if (window.location.protocol === 'http:') {
target.val('http://' + window.location.host + target.val());
} else {
target.val('https://' + window.location.host + target.val());
}
});

function setConnected(connected) {
var connect = $("#connect");
var disconnect = $("#disconnect");
var echo = $("#echo");

if (connected) {
connect.addClass("layui-btn-disabled");
disconnect.removeClass("layui-btn-disabled");
echo.removeClass("layui-btn-disabled");
} else {
connect.removeClass("layui-btn-disabled");
disconnect.addClass("layui-btn-disabled");
echo.addClass("layui-btn-disabled");
}

connect.attr("disabled", connected);
disconnect.attr("disabled", !connected);
echo.attr("disabled", !connected);
}

//連線
function connect() {
var target = $("#target").val();

var ws = new SockJS(target);
stompClient = Stomp.over(ws);

stompClient.connect({}, function () {
setConnected(true);
log('Info: STOMP connection opened.');

//連線成功後，主動拉取未讀訊息
pullUnreadMessage("/topic/reply");

//訂閱服務端的/topic/reply地址
stompClient.subscribe("/user/topic/reply", function (response) {
log(JSON.parse(response.body).content);
})
},function () {
//斷開處理
setConnected(false);
log('Info: STOMP connection closed.');
});
}

//斷開連線
function disconnect() {
if (stompClient != null) {
stompClient.disconnect();
stompClient = null;
}
setConnected(false);
log('Info: STOMP connection closed.');
}

//向指定使用者傳送訊息
function sendMessage() {
if (stompClient != null) {
var receiver = $("#receiver").val();
var msg = $("#message").val();
log('Sent: ' + JSON.stringify({'receiver': receiver, 'msg':msg}));

$.ajax({
url: "/wsTemplate/sendToUser",
type: "POST",
dataType: "json",
async: true,
data: {
"receiver": receiver,
"msg": msg
},
success: function (data) {

}
});
} else {
layer.msg('STOMP connection not established, please connect.', {
offset: 'auto'
,icon: 2
});
}
}

//從伺服器拉取未讀訊息
function pullUnreadMessage(destination) {
$.ajax({
url: "/wsTemplate/pullUnreadMessage",
type: "POST",
dataType: "json",
async: true,
data: {
"destination": destination
},
success: function (data) {
if (data.result != null) {
$.each(data.result, function (i, item) {
log(JSON.parse(item).content);
})
} else if (data.code !=null && data.code == "500") {
layer.msg(data.msg, {
offset: 'auto'
,icon: 2
});
}
}
});
}

//日誌輸出
function log(message) {
console.debug(message);
}
</script>
</head>
<body>
<noscript><h2 style="color: #ff0000">Seems your browser doesn't support Javascript! Websockets rely on Javascript being
enabled. Please enable
Javascript and reload this page!</h2></noscript>
<div>
<div id="connect-container" class="layui-elem-field">
<legend>Chat With STOMP Message</legend>
<div>
<input id="target" type="text" class="layui-input" size="40" style="width: 350px" value="/chat-websocket"/>
</div>
<div>
<button id="connect" class="layui-btn layui-btn-normal" onclick="connect();">Connect</button>
<button id="disconnect" class="layui-btn layui-btn-normal layui-btn-disabled" disabled="disabled"
onclick="disconnect();">Disconnect
</button>

</div>
<div class="message">
<input id="receiver" type="text" class="layui-input" size="40" style="width: 350px" placeholder="接收者姓名" value=""/>
<input id="message" type="text" class="layui-input" size="40" style="width: 350px" placeholder="訊息內容" value=""/>
</div>
<div>
<button id="echo" class="layui-btn layui-btn-normal layui-btn-disabled" disabled="disabled"
onclick="sendMessage();">Send Message
</button>
</div>
</div>
</div>
</body>
</html>
複製程式碼

啟動專案後，分別在兩個瀏覽器中使用不同的賬號登入，接著互相給對方傳送訊息，效果如下：

介面一：

介面二：