springboot websocket叢集（stomp協議）連線時候傳遞引數
阿新 • • 發佈：2019-07-05
最近在公司專案中接到個需求。就是後臺跟前端瀏覽器要保持長連線，後臺主動往前臺推資料。
網上查了下，websocket stomp協議處理這個很簡單。尤其是跟springboot 整合。
但是由於開始是單機玩的，很順利。
但是後面部署到生產搞叢集的話，就會出問題了。
假如叢集兩個節點，瀏覽器A與節點A建立連線，A節點發的訊息瀏覽器A節點肯定能收到。但是B節點由於沒有跟瀏覽器A建立連線。B節點發的訊息瀏覽器就收不到了。
網上也查了好多，但是沒有一個說的很清楚的，也很多都是理論層面的。
還有很多思路都是通過session獲取資訊的。但是這都不是我需要的。我需要的是從前臺傳遞引數，連線的時候每個節點儲存下。然後通過SimpleUserRegistry.getUser獲取。
話不多說，直接上程式碼。
<script type="text/javascript" src="${request.contextPath}/scripts/sockjs.min.js"></script>
<script type="text/javascript" src="${request.contextPath}/scripts/stomp.min.js"></script>
var WEB_SOCKET = {
topic : "",
url : "",
stompClient : null,
connect : function(url, topic, callback,userid) {
this.url = url;
this.topic = topic;
var socket = new SockJS(url); //連線SockJS的endpoint名稱為"endpointOyzc"
WEB_SOCKET.stompClient = Stomp.over(socket);//使用STMOP子協議的WebSocket客戶端
WEB_SOCKET.stompClient.connect({userid:userid},function(frame){//連線WebSocket服務端
// console.log('Connected:' + frame);
//通過stompClient.subscribe訂閱/topic/getResponse 目標(destination)傳送的訊息
WEB_SOCKET.stompClient.subscribe(topic, callback);
});
}
};這是響應的前端程式碼。只需要引入兩個js。呼叫new SockJS(url) 就代表跟伺服器建立連線了。
@Configuration
//註解開啟使用STOMP協議來傳輸基於代理(message broker)的訊息,這時控制器支援使用@MessageMapping,就像使用@RequestMapping一樣
@EnableWebSocketMessageBroker
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer {
@Autowired
private GetHeaderParamInterceptor getHeaderParamInterceptor;
@Override
//註冊STOMP協議的節點(endpoint),並對映指定的url
public void registerStompEndpoints(StompEndpointRegistry registry) {
//註冊一個STOMP的endpoint,並指定使用SockJS協議
registry.addEndpoint("/endpointOyzc")
.setAllowedOrigins("*")
.withSockJS();
/* registry.addEndpoint("/endpointOyzc")
.setAllowedOrigins("*")
.setHandshakeHandler(xlHandshakeHandler)
.withSockJS();*/
}
@Override
//配置訊息代理(Message Broker)
public void configureMessageBroker(MessageBrokerRegistry registry) {
//點對點應配置一個/user訊息代理，廣播式應配置一個/topic訊息代理
registry.enableSimpleBroker("/topic", "/user");
// 全域性使用的訊息字首（客戶端訂閱路徑上會體現出來）
//registry.setApplicationDestinationPrefixes("/app");
//點對點使用的訂閱字首（客戶端訂閱路徑上會體現出來），不設定的話，預設也是/user/
registry.setUserDestinationPrefix("/user");
}
/**
* 採用自定義攔截器，獲取connect時候傳遞的引數
*
* @param registration
*/
@Override
public void configureClientInboundChannel(ChannelRegistration registration) {
registration.interceptors(getHeaderParamInterceptor);
}
}注：上面的endpointOyzc就是前端的url。後面註冊端點，前臺連結。
然後注意下configureClientInboundChannel這個方法，這個方法裡面注入攔截器就是為了連結時候接收引數的。
/**
* @author : hao
* @description : websocket建立連結的時候獲取headeri裡認證的引數攔截器。
* @time : 2019/7/3 20:42
*/
@Component
public class GetHeaderParamInterceptor extends ChannelInterceptorAdapter {
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
if (StompCommand.CONNECT.equals(accessor.getCommand())) {
Object raw = message.getHeaders().get(SimpMessageHeaderAccessor.NATIVE_HEADERS);
if (raw instanceof Map) {
Object name = ((Map) raw).get("userid");
if (name instanceof LinkedList) {
// 設定當前訪問的認證使用者
accessor.setUser(new JqxxPrincipal(((LinkedList) name).get(0).toString()));
}
}
}
return message;
}
}
/**
* @author : hao
* @description : 自定義的java.security.Principal
* @time : 2019/7/3 20:42
*/
public class JqxxPrincipal implements Principal {
private String loginName;
public JqxxPrincipal(String loginName) {
this.loginName = loginName;
}
@Override
public String getName() {
return loginName;
}
}
這樣就存入的前臺傳的引數。
後臺發訊息的時候怎麼發呢？
/**
* @author : hao
* @description : websocket傳送代理，負責傳送訊息
* @time : 2019/7/4 11:01
*/
@Component
@Slf4j
public class WebsocketSendProxy<T> {
@Autowired
private SimpMessagingTemplate template;
@Autowired
private SimpUserRegistry userRegistry;
@Resource(name = "redisServiceImpl")
private RedisService redisService;
@Value("spring.redis.message.topic-name")
private String topicName;
public void sendMsg(RedisWebsocketMsg<T> redisWebsocketMsg) {
SimpUser simpUser = userRegistry.getUser(redisWebsocketMsg.getReceiver());
log.info("傳送訊息前獲取接收方為{},根據Registry獲取本節點上這個使用者{}", redisWebsocketMsg.getReceiver(), simpUser);
if (simpUser != null && StringUtils.isNotBlank(simpUser.getName())) {
//2. 獲取WebSocket客戶端的訂閱地址
WebSocketChannelEnum channelEnum = WebSocketChannelEnum.fromCode(redisWebsocketMsg.getChannelCode());
if (channelEnum != null) {
//3. 給WebSocket客戶端傳送訊息
template.convertAndSendToUser(redisWebsocketMsg.getReceiver(), channelEnum.getSubscribeUrl(), redisWebsocketMsg.getContent());
}
} else {
//給其他訂閱了主題的節點發訊息,因為本節點沒有
redisService.convertAndSend(topicName, redisWebsocketMsg);
}
}
}
可以發現上面程式碼利用了redis監聽模型，也就是redis模型的訊息佇列
/**
* @author : hao
* @description : redis訊息監聽實現類，接收處理類
* @time : 2019/7/3 14:00
*/
@Component
@Slf4j
public class MessageReceiver {
@Autowired
private SimpMessagingTemplate messagingTemplate;
@Autowired
private SimpUserRegistry userRegistry;
/**
* 處理WebSocket訊息
*/
public void receiveMessage(RedisWebsocketMsg redisWebsocketMsg) {
log.info(MessageFormat.format("Received Message: {0}", redisWebsocketMsg));
//1. 取出使用者名稱並判斷是否連線到當前應用節點的WebSocket
SimpUser simpUser = userRegistry.getUser(redisWebsocketMsg.getReceiver());
if (simpUser != null && StringUtils.isNotBlank(simpUser.getName())) {
//2. 獲取WebSocket客戶端的訂閱地址
WebSocketChannelEnum channelEnum = WebSocketChannelEnum.fromCode(redisWebsocketMsg.getChannelCode());
if (channelEnum != null) {
//3. 給WebSocket客戶端傳送訊息
messagingTemplate.convertAndSendToUser(redisWebsocketMsg.getReceiver(), channelEnum.getSubscribeUrl(), redisWebsocketMsg.getContent());
}
}
}
}
redis訊息模型只貼部分程式碼就好了
/**
* 訊息監聽器
*/
@Bean
MessageListenerAdapter messageListenerAdapter(MessageReceiver messageReceiver, Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer){
//訊息接收者以及對應的預設處理方法
MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(messageReceiver, "receiveMessage");
//訊息的反序列化方式
messageListenerAdapter.setSerializer(jackson2JsonRedisSerializer);
return messageListenerAdapter;
}
/**
* message listener container
*/
@Bean
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory
, MessageListenerAdapter messageListenerAdapter){
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
//新增訊息監聽器
container.addMessageListener(messageListenerAdapter, new PatternTopic(topicName));
return container;
}
上面的思路大體如下：客戶端簡歷連結時候，傳過來userid儲存起來。發訊息的時候 通過userRegistry獲取，能獲取到就證明是跟本節點建立的連結，直接用本節點發訊息就好了。
如果不是就利用redis訊息佇列，把訊息推出去。每個節點去判斷獲取看下是不是本節點的userid。這樣就實現了叢集的部