分布式WebSocket - 4SpringBoot集成STOMP协议,RabbitMQ为消息代理
Posted 叁滴水
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了分布式WebSocket - 4SpringBoot集成STOMP协议,RabbitMQ为消息代理相关的知识,希望对你有一定的参考价值。
一、为什么引入RabbitMQ为消息代理?
spring基于内存为代理时,扩容不太方便,一旦服务宕机,内存的消息将全部丢失。因此需要有方便扩容的解决方案。
方案1
- 通过负载策略,用户1、2连接推送服务A。用户3、4连接服务器B。redis存储每个用户和服务器的对应关系,当然这里也不一定是redis,zookeeper也可以。
- 业务系统在发送消息时,去redis查看用户在哪个服务器上,然后通过RPC调用对应服务器的接口进行发现消息,这个逻辑需要自行实现。
方案2
- 通过负载策略,用户1、2连接推送服务A。用户3、4连接服务器B。他们分别自己订阅自己想要了解的主题。
- 业务服务器想要推送消息时,不管是群里还是资料,可以连接任意服务器,只要推送给STOMP代理即可。STOMP代理会自行推送消息。
- 这样如果推送服务压力大时,可以任意扩容。
因此,在一些需要快速开发的项目中,后端使用了springboot,而且使用了STOMP协议,那么这个方案是一个不错的选择。
二、为什么使用RabbitMQ?
支持STOMP协议的中间件还有activemq
,但是rabbitmq相对来说更加强大,使用的人更多。
三、架构图
这个架构图和内存的broker类似,区别在于,左下角有一个专门存储消息的容器。这个容器就是RabbitMQ。当然,由这个架构图看来,在原有的功能上配置RabbitMQ只需要修改部分代码即可。
四、安装RabbitMQ
网上安装RabbitMQ的资料很多,这里我就不在陈述。如果安装Windows版本的RabbitMQ可以参考。我在测试时,用的是相对新一点的版本。
安装成功之后启用插件,注:这是两个命令,一个一个执行。
rabbitmq-plugins enable rabbitmq_stomp
rabbitmq-plugins enable rabbitmq_web_stomp
正常的RabbitMQ启用stomp协议之后,看下图
五、代码配置RabbitMQ
5.1、服务器端
配置类WebSocketConfig.java
@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer
{
// 启用一个简单的基于内存的消息代理
@Override
public void configureMessageBroker(MessageBrokerRegistry config) {
//通过/topic 开头的主题可以进行订阅
config.enableSimpleBroker("/topic");
//send命令时需要带上/app前缀
config.setApplicationDestinationPrefixes("/app");
//配置RabbitMQ代理
// 配置支持的topic
config.enableStompBrokerRelay("/topic/","/queue/","exchange")
.setRelayHost("localhost") //地址
.setRelayPort(61613) //端口
.setClientLogin("guest") // 账号密码
.setClientPasscode("guest")
.setVirtualHost("/");
}
@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
//连接前缀
registry.addEndpoint("/gs-guide-websocket")
.setAllowedOrigins("*") // 跨域处理
.withSockJS(); //支持socketJs
}
}
控制器类
@Slf4j
@RestController
public class TestController
{
@Autowired
private SimpMessagingTemplate simpMessagingTemplate;
@GetMapping ("/queue")
public void queue(HelloMessage message) throws Exception {
simpMessagingTemplate.convertAndSend ("/queue/user"+message.getUserId (),
new Greeting("Hello, " + htmlUtils.htmlEscape(message.getName()) + "!"));
}
@GetMapping ("/topic")
public void topic(HelloMessage message) throws Exception {
simpMessagingTemplate.convertAndSend ("/topic/user"+message.getUserId (),
new Greeting("Hello, " + HtmlUtils.htmlEscape(message.getName()) + "!"));
}
}
5.2、H5端
var stompClient = null;
var userId = null;
function setConnected(connected) {
$("#connect").prop("disabled", connected);
$("#disconnect").prop("disabled", !connected);
if (connected) {
$("#conversation").show();
}
else {
$("#conversation").hide();
}
$("#greetings").html("");
}
function connect() {
var socket = new SockJS('/gs-guide-websocket');
stompClient = Stomp.over(socket);
stompClient.connect({}, function (frame) {
setConnected(true);
console.log('Connected: ' + frame);
stompClient.subscribe('/queue/user'+userId, function (greeting) {
showGreeting(JSON.parse(greeting.body).content);
});
stompClient.subscribe('/topic/user'+userId, function (greeting) {
showGreeting(JSON.parse(greeting.body).content);
});
});
}
function disconnect() {
if (stompClient !== null) {
stompClient.disconnect();
}
setConnected(false);
console.log("Disconnected");
}
function showGreeting(message) {
$("#greetings").append("<tr><td>" + message + "</td></tr>");
}
function GetQueryString(name) {
var reg = new RegExp("(^|&)" + name + "=([^&]*)(&|$)", "i");
var r = window.location.search.substr(1).match(reg); //获取url中"?"符后的字符串并正则匹配
var context = "";
if (r != null)
context = r[2];
reg = null;
r = null;
return context == null || context == "" || context == "undefined" ? "" : context;
}
$(function () {
userId = GetQueryString("userId");
$("form").on('submit', function (e) {
e.preventDefault();
});
$( "#connect" ).click(function() { connect(); });
$( "#disconnect" ).click(function() { disconnect(); });
});
html
<!DOCTYPE html>
<html>
<head>
<title>Hello WebSocket</title>
<link href="/webjars/bootstrap/css/bootstrap.min.css" rel="stylesheet">
<link href="/main.css" rel="stylesheet">
<script type="text/javascript" src="https://cdn.bootcss.com/jquery/3.2.1/jquery.min.js"></script>
<script src="/webjars/sockjs-client/sockjs.min.js"></script>
<script src="/webjars/stomp-websocket/stomp.min.js"></script>
<script src="/app.js"></script>
</head>
<body>
<noscript><h2 style="color: #ff0000">Seems your browser doesn't support Javascript! Websocket relies on Javascript being
enabled. Please enable
Javascript and reload this page!</h2></noscript>
<div id="main-content" class="container">
<div class="row">
<div class="col-md-6">
<form class="form-inline">
<div class="form-group">
<label for="connect">WebSocket connection:</label>
<button id="connect" class="btn btn-default" type="submit">Connect</button>
<button id="disconnect" class="btn btn-default" type="submit" disabled="disabled">Disconnect
</button>
</div>
</form>
</div>
<div class="col-md-6">
<!-- -->
</div>
</div>
<div class="row">
<div class="col-md-12">
<table id="conversation" class="table table-striped">
<thead>
<tr>
<th>Greetings</th>
</tr>
</thead>
<tbody id="greetings">
</tbody>
</table>
</div>
</div>
</div>
</body>
</html>
浏览器进入http://localhost:8080/?userId=1
,点击connect进行连接。然后调用controller的queue方法,向用户1发送信息,http://localhost:8080/queue?userId=1&name=ni'hao
发送信息。
RabbitMQ中队列信息如下,如此,RabbitMQ代理配置完成。
以上是关于分布式WebSocket - 4SpringBoot集成STOMP协议,RabbitMQ为消息代理的主要内容,如果未能解决你的问题,请参考以下文章