SpringBoot webSocket实现发送广播、点对点消息和Android接收
1、SpringBoot webSocket
SpringBoot 使用的websocket 协议，不是标准的websocket协议，使用的是名称叫做STOMP的协议。
1.1 STOMP协议说明
STOMP，Streaming Text Orientated Message Protocol，是流文本定向消息协议，是一种为MOM(Message Oriented Middleware，面向消息的中间件)设计的简单文本协议。
它提供了一个可互操作的连接格式，允许STOMP客户端与任意STOMP消息代理(Broker)进行交互，类似于OpenWire(一种二进制协议)。
由于其设计简单，很容易开发客户端，因此在多种语言和多种平台上得到广泛应用。其中最流行的STOMP消息代理是Apache ActiveMQ。
1.2 搭建
本人使用的是Inject idea 搭建的springBoot websocket，并未采用熟悉的gradle，而是采用了maven方式搭建。
项目结构如下
pom.xml：
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.drawthink</groupId>
<artifactId>websocketdemo</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<name>webSocketdemo</name>
<description>webSocketDemo project for Spring Boot</description>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.3.6.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-thymeleaf</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
Application：
package com.drawthink;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class WebSocketdemoApplication {
public static void main(String[] args) {
SpringApplication.run(WebSocketdemoApplication.class, args);
}
}
WebSocketConfig
package com.drawthink.websocket;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.web.socket.config.annotation.AbstractWebSocketMessageBrokerConfigurer;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
/**
* Created by lincoln on 16-10-25
*/
@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer {
@Override
public void registerStompEndpoints(StompEndpointRegistry stompEndpointRegistry) {
//允许使用socketJs方式访问，访问点为hello，允许跨域
stompEndpointRegistry.addEndpoint("/hello").setAllowedOrigins("*").withSockJS();
}
@Override
public void configureMessageBroker(MessageBrokerRegistry registry) {
//订阅Broker名称
registry.enableSimpleBroker("/topic","/user");
//全局使用的订阅前缀（客户端订阅路径上会体现出来）
registry.setApplicationDestinationPrefixes("/app/");
//点对点使用的订阅前缀（客户端订阅路径上会体现出来），不设置的话，默认也是/user/
//registry.setUserDestinationPrefix("/user/");
}
}
WebSocketController
package com.drawthink.websocket.controller;
import com.drawthink.message.ClientMessage;
import com.drawthink.message.ServerMessage;
import com.drawthink.message.ToUserMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Controller;
/**
* Created by lincoln on 16-10-25
*/
@Controller
public class WebSocketController {
@MessageMapping("/welcome")
//SendTo 发送至 Broker 下的指定订阅路径
@SendTo("/topic/getResponse")
public ServerMessage say(ClientMessage clientMessage){
//方法用于广播测试
System.out.println("clientMessage.getName() = " + clientMessage.getName());
return new ServerMessage("Welcome , "+clientMessage.getName()+" !");
}
//注入SimpMessagingTemplate 用于点对点消息发送
@Autowired
private SimpMessagingTemplate messagingTemplate;
@MessageMapping("/cheat")
// 发送的订阅路径为/user/{userId}/message
// /user/路径是默认的一个，如果想要改变，必须在config 中setUserDestinationPrefix
public void cheatTo(ToUserMessage toUserMessage){
//方法用于点对点测试
System.out.println("toUserMessage.getMessage() = " + toUserMessage.getMessage());
System.out.println("toUserMessage.getUserId() = " + toUserMessage.getUserId()); messagingTemplate.convertAndSendToUser(toUserMessage.getUserId(),"/message",toUserMessage.getMessage());
}
}
Vo
package com.drawthink.message;
/**
* Created by lincoln on 16-10-25
*/
public class ClientMessage {
private String name;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
}
package com.drawthink.message;
/**
* Created by lincoln on 16-10-25
*/
public class ServerMessage {
private String responseMessage;
public ServerMessage(String responseMessage) {
this.responseMessage = responseMessage;
}
public String getResponseMessage() {
return responseMessage;
}
public void setResponseMessage(String responseMessage) {
this.responseMessage = responseMessage;
}
}
package com.drawthink.message;
/**
* Created by lincoln on 16-10-25
*/
public class ToUserMessage {
private String userId;
private String message;
public String getUserId() {
return userId;
}
public void setUserId(String userId) {
this.userId = userId;
}
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
}
Android 客户端
STOMP协议在Android系统中没有默认实现，必须自行去实现。不过好消息是，开源大神们已经完成了Android上使用STOMP协议的实现，所以我们只需要使用就好了。
地址：StompProtocolAndroid_jb51.rar
搭建
build.gradle(app)
apply plugin: 'com.android.application'
android {
compileSdkVersion 24
buildToolsVersion "24.0.3"
defaultConfig {
applicationId "com.drawthink.websocket"
minSdkVersion 16
targetSdkVersion 24
versionCode 1
versionName "1.0"
testInstrumentationRunner "android.support.test.runner.AndroidJUnitRunner"
}
buildTypes {
release {
minifyEnabled false
proguardFiles getDefaultProguardFile('proguard-android.txt'), 'proguard-rules.pro'
}
}
}
dependencies {
compile fileTree(include: ['*.jar'], dir: 'libs')
androidTestCompile('com.android.support.test.espresso:espresso-core:2.2.2', {
exclude group: 'com.android.support', module: 'support-annotations'
})
compile 'com.android.support:appcompat-v7:24.2.1'
testCompile 'junit:junit:4.12'
//依赖STOMP协议的Android实现
compile 'com.github.NaikSoftware:StompProtocolAndroid:1.1.1'
//StompProtocolAndroid 依赖于webSocket的标准实现
compile 'org.java-websocket:Java-WebSocket:1.3.0'
}
接收广播实例：
package com.drawthink.websocket;
import android.content.Intent;
import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.view.View;
import android.widget.Button;
import android.widget.EditText;
import android.widget.TextView;
import android.widget.Toast;
import org.java_websocket.WebSocket;
import rx.Subscriber;
import rx.functions.Action1;
import ua.naiksoftware.stomp.LifecycleEvent;
import ua.naiksoftware.stomp.Stomp;
import ua.naiksoftware.stomp.client.StompClient;
import ua.naiksoftware.stomp.client.StompMessage;
import static android.content.ContentValues.TAG;
public class MainActivity extends AppCompatActivity {
private TextView serverMessage;
private Button start;
private Button stop;
private Button send;
private EditText editText;
private StompClient mStompClient;
private Button cheat;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
bindView();
start.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
//创建client 实例
createStompClient();
//订阅消息
registerStompTopic();
}
});
send.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
mStompClient.send("/app/welcome","{\"name\":\""+editText.getText()+"\"}")
.subscribe(new Subscriber<Void>() {
@Override
public void onCompleted() {
toast("发送成功");
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
toast("发送错误");
}
@Override
public void onNext(Void aVoid) {
}
});
}
});
stop.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
mStompClient.disconnect();
}
});
cheat.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
startActivity(new Intent(MainActivity.this,CheatActivity.class));
if(mStompClient != null) {
mStompClient.disconnect();
}
finish();
}
});
}
private void showMessage(final StompMessage stompMessage) {
runOnUiThread(new Runnable() {
@Override
public void run() {
serverMessage.setText("stomp command is --->"+stompMessage.getStompCommand() +" body is --->"+stompMessage.getPayload());
}
});
}
//创建client 实例
private void createStompClient() {
mStompClient = Stomp.over(WebSocket.class, "ws://192.168.0.46:8080/hello/websocket");
mStompClient.connect();
Toast.makeText(MainActivity.this,"开始连接 192.168.0.46:8080",Toast.LENGTH_SHORT).show();
mStompClient.lifecycle().subscribe(new Action1<LifecycleEvent>() {
@Override
public void call(LifecycleEvent lifecycleEvent) {
switch (lifecycleEvent.getType()) {
case OPENED:
Log.d(TAG, "Stomp connection opened");
toast("连接已开启");
break;
case ERROR:
Log.e(TAG, "Stomp Error", lifecycleEvent.getException());
toast("连接出错");
break;
case CLOSED:
Log.d(TAG, "Stomp connection closed");
toast("连接关闭");
break;
}
}
});
}
//订阅消息
private void registerStompTopic() {
mStompClient.topic("/topic/getResponse").subscribe(new Action1<StompMessage>() {
@Override
public void call(StompMessage stompMessage) {
Log.e(TAG, "call: " +stompMessage.getPayload() );
showMessage(stompMessage);
}
});
}
private void toast(final String message) {
runOnUiThread(new Runnable() {
@Override
public void run() {
Toast.makeText(MainActivity.this,message,Toast.LENGTH_SHORT).show();
}
});
}
private void bindView() {
serverMessage = (TextView) findViewById(R.id.serverMessage);
start = (Button) findViewById(R.id.start);
stop = (Button) findViewById(R.id.stop);
send = (Button) findViewById(R.id.send);
editText = (EditText) findViewById(R.id.clientMessage);
cheat = (Button) findViewById(R.id.cheat);
}
}
点对点
package com.drawthink.websocket;
import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.view.View;
import android.widget.Button;
import android.widget.EditText;
import android.widget.LinearLayout;
import android.widget.TextView;
import android.widget.Toast;
import org.java_websocket.WebSocket;
import rx.Subscriber;
import rx.functions.Action1;
import ua.naiksoftware.stomp.LifecycleEvent;
import ua.naiksoftware.stomp.Stomp;
import ua.naiksoftware.stomp.client.StompClient;
import ua.naiksoftware.stomp.client.StompMessage;
import static android.content.ContentValues.TAG;
public class CheatActivity extends AppCompatActivity {
private EditText cheat;
private Button send;
private LinearLayout message;
private StompClient mStompClient;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_cheat);
bindView();
createStompClient();
registerStompTopic();
send.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
// 向/app/cheat发送Json数据
mStompClient.send("/app/cheat","{\"userId\":\"lincoln\",\"message\":\""+cheat.getText()+"\"}")
.subscribe(new Subscriber<Void>() {
@Override
public void onCompleted() {
toast("发送成功");
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
toast("发送错误");
}
@Override
public void onNext(Void aVoid) {
}
});
}
});
}
private void bindView() {
cheat = (EditText) findViewById(R.id.cheat);
send = (Button) findViewById(R.id.send);
message = (LinearLayout) findViewById(R.id.message);
}
private void createStompClient() {
mStompClient = Stomp.over(WebSocket.class, "ws://192.168.0.46:8080/hello/websocket");
mStompClient.connect();
Toast.makeText(CheatActivity.this,"开始连接 192.168.0.46:8080",Toast.LENGTH_SHORT).show();
mStompClient.lifecycle().subscribe(new Action1<LifecycleEvent>() {
@Override
public void call(LifecycleEvent lifecycleEvent) {
switch (lifecycleEvent.getType()) {
case OPENED:
Log.d(TAG, "Stomp connection opened");
toast("连接已开启");
break;
case ERROR:
Log.e(TAG, "Stomp Error", lifecycleEvent.getException());
toast("连接出错");
break;
case CLOSED:
Log.d(TAG, "Stomp connection closed");
toast("连接关闭");
break;
}
}
});
}
// 接收/user/xiaoli/message路径发布的消息
private void registerStompTopic() {
mStompClient.topic("/user/xiaoli/message").subscribe(new Action1<StompMessage>() {
@Override
public void call(StompMessage stompMessage) {
Log.e(TAG, "call: " +stompMessage.getPayload() );
showMessage(stompMessage);
}
});
}
private void showMessage(final StompMessage stompMessage) {
runOnUiThread(new Runnable() {
@Override
public void run() {
TextView text = new TextView(CheatActivity.this);
text.setLayoutParams(new LinearLayout.LayoutParams(LinearLayout.LayoutParams.MATCH_PARENT, LinearLayout.LayoutParams.WRAP_CONTENT));
text.setText(System.currentTimeMillis() +" body is --->"+stompMessage.getPayload());
message.addView(text);
}
});
}
private void toast(final String message) {
runOnUiThread(new Runnable() {
@Override
public void run() {
Toast.makeText(CheatActivity.this,message,Toast.LENGTH_SHORT).show();
}
});
}
}
代码比较乱，说明一下。
1、STOMP 使用的时候，关键是发布订阅的关系，使用过消息队列，例如rabbitMQ的应该很容易理解。
服务器端 WebSocketConfig.Java文件控制的就是订阅发布的路径关系。
2、websocket的路径说明，本例中连接的是ws://192.168.0.46:8080/hello/websocket路径，/hello是在WebSocketConfig的stompEndpointRegistry.addEndpoint(“/hello”).setAllowedOrigins(““).withSockJS();*确定的， 如果有多个endpoint，这个地方的路径也会随之变化。
3、发布路径
发布信息的路径是由WebSocketConfig中的 setApplicationDestinationPrefixes(“/app/”); 和 Controller 中@MessageMapping(“/welcome”) 组合确定的。
例如发广播消息，路径为/app/welcome
例如发点对点消息，路径为/app/cheat
4、消息订阅路径
订阅broker源自WebSocketConfig中的registry.enableSimpleBroker(“/topic”,”/user”);此处开放了两个broker，具体的订阅服务路径给基于Controller中的 @SendTo(“/topic/getResponse”)或SimpMessagingTemplate中给定。（注：此处，服务器和客户端须约定订阅路径）
5、关于心跳
订阅发布模型的心跳很简单，客户端向一个指定的心跳路径发送心跳，服务器处理，服务器使用指定的订阅路径向客户端发心跳，即可。因为没有Socket，只需要记录是否联通的状态即可，重连客户端做一下就好了。
本人菜鸟，肯定有些地方没有搞清楚，如果有误，请大神斧正。
代码下载地址：blogRepository_jb51.rar
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持我们。
相关推荐
-
详解在Spring Boot框架下使用WebSocket实现消息推送
spring Boot的学习持续进行中.前面两篇博客我们介绍了如何使用Spring Boot容器搭建Web项目以及怎样为我们的Project添加HTTPS的支持,在这两篇文章的基础上,我们今天来看看如何在Spring Boot中使用WebSocket. 什么是WebSocket WebSocket为浏览器和服务器之间提供了双工异步通信功能,也就是说我们可以利用浏览器给服务器发送消息,服务器也可以给浏览器发送消息,目前主流浏览器的主流版本对WebSocket的支持都算是比较好的,但是在实际开发中使
-
详解spring boot Websocket使用笔记
本文只作为个人笔记,大部分代码是引用其他人的文章的. 在springboot项目中使用websocket做推送,虽然挺简单的,但初学也踩过几个坑,特此记录. 使用websocket有两种方式:1是使用sockjs,2是使用h5的标准.使用Html5标准自然更方便简单,所以记录的是配合h5的使用方法. 1.pom 核心是@ServerEndpoint这个注解.这个注解是Javaee标准里的注解,tomcat7以上已经对其进行了实现,如果是用传统方法使用tomcat发布项目,只要在pom文件中引入j
-
使用 Spring Boot 实现 WebSocket实时通信
在开发 Web 应用程序时,我们有时需要将服务端事件推送到连接的客户端.但 HTTP 并不能做到.客户端打开与服务端的连接并请求数据,但服务端不能打开与客户端的连接并推送数据. 为了解决这个限制,我们可以建立了一个轮询模式,网页会间隔地轮询服务器以获取新事件.但这种模式不太理想,因为它增加了 HTTP 开销,速度也只能达到与轮询的速率一样快,并且给服务器增加了不必要的负载. 幸运的是,HTML5 WebSocket 出现了.WebSocket 协议允许浏览器与 Web 服务器之间进行低开销的交互
-
关于Spring Boot WebSocket整合以及nginx配置详解
前言 本文主要给大家介绍了关于Spring Boot WebSocket整合及nginx配置的相关内容,分享出来供大家参考学习,下面话不多说了,来一起看看详细的介绍吧. 一:Spring Boot WebSocket整合 创建一个maven项目,加入如下依赖 <dependencyManagement> <dependencies> <dependency> <groupId>org.springframework.boot</groupId>
-
SpringBoot+Websocket实现一个简单的网页聊天功能代码
最近做了一个SpringBoot的项目,被SpringBoot那简介的配置所迷住.刚好项目中,用到了websocket.于是,我就想着,做一个SpringBoot+websocket简单的网页聊天Demo. 效果展示: 当然,项目很简单,没什么代码,一眼就能明白 导入websocket的包. 通过使用SpringBoot导入包的时候,我们可以发现,很多包都是以 spring-boot-starter 开头的,对于我这种强迫症 ,简直是福音 <dependency> <groupId>
-
Spring Boot 开发私有即时通信系统（WebSocket）
1/ 概述 利用Spring Boot作为基础框架,Spring Security作为安全框架,WebSocket作为通信框架,实现点对点聊天和群聊天. 2/ 所需依赖 Spring Boot 版本 1.5.3,使用MongoDB存储数据(非必须),Maven依赖如下: <properties> <java.version>1.8</java.version> <thymeleaf.version>3.0.0.RELEASE</thymeleaf.ve
-
详解spring boot实现websocket
前言 QQ这类即时通讯工具多数是以桌面应用的方式存在.在没有websocket出现之前,如果开发一个网页版的即时通讯应用,则需要定时刷新页面或定时调用ajax请求,这无疑会加大服务器的负载和增加了客户端的流量.而websocket的出现,则完美的解决了这些问题. spring boot对websocket进行了封装,这对实现一个websocket网页即时通讯应用来说,变得非常简单. 一.准备工作 pom.xml引入 <dependency> <groupId>org.spring
-
SpringBoot webSocket实现发送广播、点对点消息和Android接收
1.SpringBoot webSocket SpringBoot 使用的websocket 协议,不是标准的websocket协议,使用的是名称叫做STOMP的协议. 1.1 STOMP协议说明 STOMP,Streaming Text Orientated Message Protocol,是流文本定向消息协议,是一种为MOM(Message Oriented Middleware,面向消息的中间件)设计的简单文本协议. 它提供了一个可互操作的连接格式,允许STOMP客户端与任意STOMP消
-
Springboot Websocket Stomp 消息订阅推送
目录 需求背景 websocket协议 stomp协议 需求背景 闲话不扯,直奔主题.需要和web前端建立长链接,互相实时通讯,因此想到了websocket,后面随着需求的变更,需要用户订阅主题,实现消息的精准推送,发布订阅等,则想到了STOMP(Simple Text-Orientated Messaging Protocol) 面向消息的简单文本协议. websocket协议 想到了之前写的一个websocket长链接的demo,也贴上代码供大家参考. pom文件 直接引入spring-bo
-
SpringBoot+WebSocket+Netty实现消息推送的示例代码
上一篇文章讲了Netty的理论基础,这一篇讲一下Netty在项目中的应用场景之一:消息推送功能,可以满足给所有用户推送,也可以满足给指定某一个用户推送消息,创建的是SpringBoot项目,后台服务端使用Netty技术,前端页面使用WebSocket技术. 大概实现思路: 前端使用webSocket与服务端创建连接的时候,将用户ID传给服务端 服务端将用户ID与channel关联起来存储,同时将channel放入到channel组中 如果需要给所有用户发送消息,直接执行channel组的writ
-
Springboot整合Socket实现单点发送,广播群发,1对1,1对多实战
目录 本篇内容: 功能场景点: ① pom引入核心依赖 ② yml加上配置项 ③ 创建socket配置加载类 MySocketConfig.java ④创建消息实体 MyMessage.java ⑤创建 socket handler 负责记录客户端 连接.下线 ⑥ 封装的socket 小函数 ⑦写1个接口,模拟场景,前端页面调用后端接口,做消息推送 前端简单页面 群发,所有人都能收到 场景2,局部群发,部分人群都能收到 最后一个场景,也就是单点推送,指定某个人收到 本篇内容: 后端 + 前端简单
-
SpringBoot+WebSocket实现消息推送功能
目录 背景 WebSocket简介 协议原理 WebSocket与HTTP协议的区别 WebSocket特点 应用场景 系统集成Websocket jar包引入 Websocket配置 具体实现 测试示例 页面请求websocket 测试效果 背景 项目中经常会用到消息推送功能,关于推送技术的实现,我们通常会联想到轮询.comet长连接技术,虽然这些技术能够实现,但是需要反复连接,对于服务资源消耗过大,随着技术的发展,HtML5定义了WebSocket协议,能更好的节省服务器资源和带宽,并且能够
-
Android Intent发送广播消息实例详解
Android Intent发送广播消息 Intent的另一种用途是发送广播消息,应用程序和Android系统都可以使用Intent发送广播消息,广播消息的内容是可以与应用程序密切相关的数据信息,也可以是Android的系统信息,例如网络连接变化.电池电量变化.接收的短信或系统设置变化等.如果应用程序注册了BroadcastReceiver,则可以接受到指定的广播信息. 使用Intent发送广播消息非常简单,只须创建一个Intent,并调用sendBroadcast()函数就可把Intent携带
-
vue 实现websocket发送消息并实时接收消息
公司做了个数字货币行情的H5,需要用到websocket,刚好自己也不会,可以学习一下,美滋滋. 项目结合vue脚手架和websocket来搭建,主要遇到了两个问题,一个是:断开重连机制要如何处理:另外一个是:如何在页面上随时的发送消息并实时接收返回的数据,断开重连后又如何重新发送消息,接收消息 一.断开重连机制处理 (详细参考这里) 写一个重连的方法(重连方法中必须加一把锁,重连方法执行过程中不再执行重连动作,避免重复连接),然后在websocket的onclose和error中绑定重连方法,
-
Springboot详细讲解RocketMQ实现顺序消息的发送与消费流程
目录 一.创建Springboot项目添加rockermq依赖 二.配置rocketmq 三.新建一个controller来做消息发送 四.创建消费端监听消息消费消息 五.启动服务测试顺序消息发送与消费 如何实现顺序消息? 需要程序保证发送和消费的是同一个 Queue rocketmq默认发送的消息是进入多个消息队列,然后消费端多线程并发消费,所以默认情况,不是順序消费消息的:有時候,我们需要顺序消费一批消息,比如电商系统 订单创建.支付.完成操作,需要順序执行: RocketMQTemplat
-
SpringBoot集成RocketMQ发送事务消息的原理解析
目录 简介 原理 具体实现 消费者 消费者 生产者消息监听器 消息事务测试 正常测试 异常测试 代码调整 执行结果 总结 简介 RocketMQ 事务消息(Transactional Message)是指应用本地事务和发送消息操作可以被定义到全局事务中,要么同时成功,要么同时失败.RocketMQ 的事务消息提供类似 X/Open XA 的分布事务功能,通过事务消息能达到分布式事务的最终一致. 原理 RocketMQ事务消息通过异步确保方式,保证事务的最终一致性.设计的思想可以借鉴两个阶段提交事
-
springboot websocket集群（stomp协议）连接时候传递参数
最近在公司项目中接到个需求.就是后台跟前端浏览器要保持长连接,后台主动往前台推数据. 网上查了下,websocket stomp协议处理这个很简单.尤其是跟springboot 集成. 但是由于开始是单机玩的,很顺利. 但是后面部署到生产搞集群的话,就会出问题了. 假如集群两个节点,浏览器A与节点A建立连接,A节点发的消息浏览器A节点肯定能收到.但是B节点由于没有跟浏览器A建立连接.B节点发的消息浏览器就收不到了. 网上也查了好多,但是没有一个说的很清楚的,也很多都是理论层面的. 还有很多思路都