基于springboot websocket和okhttp实现消息中转
1、业务介绍
消息源服务的消息不能直接推给用户侧,用户与中间服务建立websocket连接,中间服务再与源服务建立websocket连接,源服务的消息推给中间服务,中间服务再将消息推送给用户。流程如下图:

此例中我们定义中间服务A的端口为8082,消息源头服务B的端口为8081,方便阅读下面代码。
说明:此例子只实现了中间服务的转发,连接的关闭等其他逻辑并没有完善,如需要请自行完善;
2、中间服务实现
中间服务即为上图的中间服务A,由于中间服务既要发送(发给用户端)消息,又要接收(从消息源服务B接收)消息,故服务A分为服务端与客户端。
服务A的websocket服务端我们使用springboot websocket实现,客户端使用okhttp实现;会话缓存暂使用内存缓存(实际项目中可置于其他缓存中)
中间服务所需依赖为:
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId>
</dependency>
<dependency><groupId>com.squareup.okhttp3</groupId><artifactId>okhttp</artifactId><version>4.2.2</version>
</dependency>
缓存类:
public class WSCache {//存储客户端session信息, {会话id:ws_session}public static Map<String, Session> clients = new ConcurrentHashMap<>();//存储把不同用户的客户端session信息集合 {userId, [会话id1,会话id2,会话id3,会话id4]}public static Map<String, Set<String>> connection = new ConcurrentHashMap<>();
}
自定义消息类:
@Accessors(chain = true)
@Data
public class MsgInfo {private String massage;//为userId,用于从缓存中获取对应用户的websocket sessionprivate String userKey;
}
2.1 中间服务A的客户端:
客户端也可以使用springboot websocket,当下我们选择使用okhttp实现。
@Slf4j
public class CommonWSClient extends WebSocketListener {/*** websocket连接建立** @param webSocket* @param response*/@Overridepublic void onOpen(WebSocket webSocket, Response response) {super.onOpen(webSocket, response);log.info("客户端连接建立:{}", response.body().string());}/*** 收到消息* @param webSocket* @param text*/@Overridepublic void onMessage(WebSocket webSocket, String text) {super.onMessage(webSocket, text);log.info("okhttp receive=>{}", text);//todo 收到源(8081)的消息,取到对应userId的消息,并将消息通过本地server发送给用户ObjectMapper mapper = new ObjectMapper();try {MsgInfo msgInfo = mapper.readValue(text, MsgInfo.class);Set<String> strings = WSCache.connection.get(msgInfo.getUserKey());if(!CollectionUtils.isEmpty(strings)){for (String sid : strings) {Session session = WSCache.clients.get(sid);session.getBasicRemote().sendText(msgInfo.getMassage());}}} catch (Exception e) {e.printStackTrace();//throw new RuntimeException(e);}}@Overridepublic void onMessage(WebSocket webSocket, ByteString bytes) {super.onMessage(webSocket, bytes);}@Overridepublic void onClosing(WebSocket webSocket, int code, String reason) {super.onClosing(webSocket, code, reason);log.info("okhttp socket closing.");}@Overridepublic void onClosed(WebSocket webSocket, int code, String reason) {super.onClosed(webSocket, code, reason);log.info("okhttp socket closed.");}@Overridepublic void onFailure(WebSocket webSocket, Throwable t, Response response) {super.onFailure(webSocket, t, response);if (response == null) {log.error("okhttp onFailure, response is null.");return;}try {log.error("okhttp onFailure, code: {}, errmsg: {}", response.code(), response.body().string());} catch (IOException e) {log.warn("okhttp onFailure failed, error: {}", e.getMessage());}}}
2.2 中间服务A的服务端:
websocket服务:
@Slf4j
@Component
@ServerEndpoint("/notice/{userId}")
public class WebSocketServer {//会话idprivate String sid = null;//建立连接的用户idprivate String userId;/*** @description: 当与用户端连接成功时,执行该方法* @PathParam 获取ServerEndpoint路径中的占位符信息类似 控制层的 @PathVariable注解**/@OnOpenpublic String onOpen(Session session, @PathParam("userId") String userId){this.sid = UUID.randomUUID().toString();this.userId = userId;WSCache.clients.put(this.sid,session);//判断该用户是否存在会话信息,不存在则添加Set<String> clientSet = WSCache.connection.get(userId);if (CollectionUtils.isEmpty(clientSet)){clientSet = new HashSet<>();clientSet.add(this.sid);}else {clientSet.add(this.sid);}WSCache.connection.put(userId,clientSet);log.info("用户{}与本地(8082)server建立连接", this.userId);//todo 本地client与源server(8081)连接Request requestRemote = new Request.Builder().url("ws://127.0.0.1:8081/api/notice/" + userId).build();OkHttpClient webSocketClientRemote = new OkHttpClient.Builder().build();WebSocket localClientRemote = webSocketClientRemote.newWebSocket(requestRemote, new CommonWSClient());log.info("本地server创建本地client,且本地client与远程(8082)server连接成功");return userId + "与本地server连接";}/*** @description: 当连接失败时,执行该方法**/@OnClosepublic void onClose(){WSCache.clients.remove(this.sid);System.out.println(this.sid+"连接断开");}/*** @description: 当收到client发送的消息时,执行该方法**/@OnMessagepublic void onMessage(String message, Session session) {System.out.println("-----------收到来自用户:" + this.userId + "的信息 " + message);}/*** @description: 当连接发生错误时,执行该方法**/@OnErrorpublic void onError(Throwable error){System.out.println("error--------系统错误");error.printStackTrace();}
}
websocket配置类:
@Configuration
public class WebSocketConfig {@Beanpublic ServerEndpointExporter serverEndpointExporter(){return new ServerEndpointExporter();}
}
3、消息源服务
消息源服务B只需要websocket服务用来发送消息即可,其实现与中间服务A的服务端相同。
服务:
@Slf4j
@Component
@ServerEndpoint("/notice/{userId}")
public class WebSocketServer {//存储客户端session信息, {会话id:ws_session}public static Map<String, Session> clients = new ConcurrentHashMap<>();//存储把不同用户的客户端session信息集合 {userId, [会话id1,会话id2,会话id3,会话id4]}public static Map<String, Set<String>> connection = new ConcurrentHashMap<>();//会话idprivate String sid = null;//建立连接的用户idprivate String userId;/*** @description: 当与客户端的websocket连接成功时,执行该方法* @PathParam 获取ServerEndpoint路径中的占位符信息类似 控制层的 @PathVariable注解**/@OnOpenpublic void onOpen(Session session, @PathParam("userId") String userId){log.info("onOpen-->session.getRequestParameterMap():{}", session.getRequestParameterMap());this.sid = UUID.randomUUID().toString();this.userId = userId;clients.put(this.sid,session);//判断该用户是否存在会话信息,不存在则添加Set<String> clientSet = connection.get(userId);if (clientSet == null){clientSet = new HashSet<>();connection.put(userId,clientSet);}clientSet.add(this.sid);System.out.println(this.userId + "用户建立连接," + this.sid+"连接开启!");}/*** @description: 当连接失败时,执行该方法**/@OnClosepublic void onClose(){clients.remove(this.sid);System.out.println(this.sid+"连接断开");}/*** @description: 当收到客户端发送的消息时,执行该方法**/@OnMessagepublic void onMessage(String message, Session session) {System.out.println("-----------收到来自用户:" + this.userId + "的信息 " + message);//自定义消息实体MsgInfo msgInfo = new MsgInfo().setUserKey(this.userId).setMassage("服务端-" + System.currentTimeMillis() + ":已收到用户" +this.userId + "的信息: " + message);sendMessageByUserId(this.userId, msgInfo);}/*** @description: 当连接发生错误时,执行该方法**/@OnErrorpublic void onError(Throwable error){System.out.println("error--------系统错误");error.printStackTrace();}/*** @description: 通过userId向用户发送信息* 该类定义成静态可以配合定时任务实现定时推送**/public static void sendMessageByUserId(String userId, MsgInfo msgInfo){if (!StringUtils.isEmpty(userId)) {Set<String> clientSet = connection.get(userId);//用户是否存在客户端连接if (Objects.nonNull(clientSet)) {Iterator<String> iterator = clientSet.iterator();while (iterator.hasNext()) {String sid = iterator.next();Session session = clients.get(sid);//向每个会话发送消息if (Objects.nonNull(session)){try {//同步发送数据,需要等上一个sendText发送完成才执行下一个发送ObjectMapper mapper = new ObjectMapper();session.getBasicRemote().sendText(mapper.writeValueAsString(msgInfo));} catch (Exception e) {e.printStackTrace();}}}}}}@Scheduled(cron = "0/10 * * * * ?")public void testSendMessageByCron(){log.info("-----------模拟消息开始发送--------------");//模拟两个用户100和200MsgInfo msg100 = new MsgInfo().setUserKey("100").setMassage("这是8081发给用户100的消息" + System.currentTimeMillis());sendMessageByUserId("100", msg100);MsgInfo msg200 = new MsgInfo().setUserKey("200").setMassage("这是8081发给用户200的消息" + System.currentTimeMillis());sendMessageByUserId("200", msg200);}
}
4、测试
我们使用: wss在线测试工具进行测试;
1、 打开两个该工具窗口,分别模拟用户100和用户200,这两个用户都连接中间服务A(端口8082的服务);


2、分别启动消息源服务B和中间服务A
此时在服务B控制台我们可以看到:

我们模拟的消息发送已经在给用户100和用户200发送,因为我们的用户100和用户200均没有与中间服务A建立连接,故此时测试界面看不到消息;
当我们在用户100的模拟界面点击“开启连接”后,可以在右侧看到发给用户100的模拟消息:

之后我们再打开用户200的连接:

好了,到这里就结束了,有任何问题请积极指出,此例子只是个例子,并未经受任何生产的测试,欢迎讨论沟通:)
相关文章:
基于springboot websocket和okhttp实现消息中转
1、业务介绍 消息源服务的消息不能直接推给用户侧,用户与中间服务建立websocket连接,中间服务再与源服务建立websocket连接,源服务的消息推给中间服务,中间服务再将消息推送给用户。流程如下图: 此例中我们定义中间服…...
@PostConstruct 注解的方法用于资源的初始化
PostConstruct 是 Java EE 5 引入的一个注解,主要用于依赖注入完成之后,需要执行的方法上。这个注解的方法会在依赖注入完成后自动被容器(如 EJB 容器或 Spring 容器)调用,并且只会被调用一次。 PostConstruct 注解的…...
(一)SvelteKit教程:hello world
(一)SvelteKit教程:hello world sveltekit 的官方教程,在这里:Creating a project • Docs • SvelteKitCreating a project • Docs • SvelteKit 我们可以按照如下的步骤来创建一个项目: npm create s…...
华为Atlas NPU ffmpeg 编译安装
处理器:鲲鹏920 NPU:昇腾 310P3 操作系统:Kylin Linux Advanced Server V10 CANN:Ascend-cann-toolkit_8.0.RC1_linux-aarch64.run FFmpeg:AscendFFmpegPlugin(不要用AscendFFmpeg) AscendFFmpegPlugin下载地址&…...
Python 虚拟环境 requirements.txt 文件生成 ;pipenv导出pip安装文件
搜索关键词: Python 虚拟环境Pipenv requirements.txt 文件生成;Pipenv 导出 pip requirements.txt安装文件 本文基于python版本 >3.9 文章内容有效日期2023年01月开始(因为此方法从这个时间开始是完全ok的) 上述为pipenv的演示版本 使用以下命令可精准生成requirement…...
Less与Sass的区别
1. 功能和工具: Sass:提供了更多的功能和内置方法,如条件语句、循环、数学函数等。Sass 也支持更复杂的操作和逻辑构建。 Less:功能也很强大,但相比之下,Sass 在功能上更为丰富和成熟。 2、编译环境&…...
力扣-2663
题目 如果一个字符串满足以下条件,则称其为 美丽字符串 : 它由英语小写字母表的前 k 个字母组成。它不包含任何长度为 2 或更长的回文子字符串。 给你一个长度为 n 的美丽字符串 s 和一个正整数 k 。 请你找出并返回一个长度为 n 的美丽字符串&#…...
CausalMMM:基于因果结构学习的营销组合建模
1. 摘要 在线广告中,营销组合建模(Marketing Mix Modeling,MMM) 被用于预测广告商家的总商品交易量(GMV),并帮助决策者调整各种广告渠道的预算分配。传统的基于回归技术的MMM方法在复杂营销场景…...
编译 CUDA 程序的基本知识和步骤
基本工具 NVCC(NVIDIA CUDA Compiler): nvcc 是 NVIDIA 提供的 CUDA 编译器,用于将 CUDA 源代码(.cu 文件)编译成可执行文件或库。它可以处理 CUDA 和主机代码(例如 C)的混合编译。nvcc 调用底层…...
[SAP ABAP] 排序内表数据
语法格式 整表排序 SORT <itab> [ASCENDING|DESCENDING]. 按指定字段排序 SORT <itab> BY f1 [ASCENDING|DESCENDING] f2 [ASCENDING|DESCENDING] ... fn [ASCENDING|DESCENDING].<itab>:代表内表 不指定排序方式则默认升序排序 示例1 结果显…...
【UML用户指南】-21-对基本行为建模-活动图
目录 1、概念 2、组成结构 2.1、动作 2.2、活动节点 2.3、控制流 2.4、分支 2.5、分岔和汇合 2.6、泳道 2.7、对象流 2.8、扩展区域 3、一般用法 3.1、对工作流建模 3.2、对操作建模 一个活动图从本质上说是一个流程图,展现从活动到活动的控制流 活动图…...
【web2】jquary,bootstrap,vue
文章目录 1.jquary:选择器1.1 jquery框架引入:$("mydiv") 当成id选择器1.2 jquery版本/对象:$(js对象) -> jquery对象1.3 jquery的页面加载事件:$ 想象成 window.onload 1.4 jquery的基本选择器:$()里内容…...
独角兽品牌獭崎酱酒:高性价比的酱香之选
在酱香型白酒领域中,獭崎酱酒以其独特的品牌定位和高性价比迅速崛起,成为市场上备受关注的独角兽品牌。作为贵州茅台镇的一款新秀酱香酒,獭崎酱酒不仅传承了百年酿造工艺,还以创新的商业模式和亲民的价格赢得了广大消费者的青睐。…...
java打印菱形和空心菱形
java打印菱形 菱形分上下两个部分。其中上部分同打印金字塔;下部分循环部分i是递减 (ps:菱形层数只能为奇数) import java.util.Scanner;public class Lingxing{public static void main(String[] args) {Scanner myScanner new Scanner(S…...
Day10 —— 大数据技术之Scala
Scala编程入门 Scala的概述什么是Scala?Scala的重要特点Scala的使用场景 Scala的安装Scala基础Scala总结 Scala的概述 什么是Scala? Scala是一种将面向对象和函数式编程结合在一起的高级语言,旨在以简洁、优雅和类型安全的方式表达通用编程…...
Linux应用系统快速部署:docker快速部署linux应用程序
目录 一、背景 (一)引入docker的起因 (二)docker介绍 (三)Docker部署的优势 1、轻量级和可移植性 2、快速部署和扩展 3、一致性 4、版本控制 5、安全性 6、资源隔离 7、简化团队协作 8、多容器…...
三目运算符中间的表达式可以省略吗(a?:c)?
熟悉C语言的童靴对三目运算符都非常熟悉,a? b : c; 如果a为true,则整个运算符的值为b,否则为c;那么问题来了,三目运算符中间的表达式可以省略吗?即a? : c; 1、linux内核中出现的省略情况 本人在阅读内核代码是发现了下面的代码: preferr…...
android 彩虹进度条自定义view实现
实现一个彩虹色进度条功能,不说明具体用途大家应该能猜到。想找别人造的轮子,但是没有合适的,所以决定自己实现一个。 相关知识 android 自定义view LinearGradient 线性渐变 实现步骤 自定义view 自定义一个TmcView类继承View 重写两…...
免费一年SSL证书申请——建议收藏
免费一年SSL证书申请——建议收藏 获取免费一年期SSL证书其实挺简单的 准备你的网站: 确保你的网站已经有了域名,而且这个域名已经指向你的服务器。还要检查你的服务器支持HTTPS,也就是443端口要打开,这是HTTPS默认用的。 验证域…...
【docker1】指令,docker-compose,Dockerfile
文章目录 1.pull/image,run/ps(进程),exec/commit2.save/load:docker save 镜像id,不是容器id3.docker-compose:多容器:宿主机(eth0网卡)安装docker会生成一…...
Appium+python自动化(十六)- ADB命令
简介 Android 调试桥(adb)是多种用途的工具,该工具可以帮助你你管理设备或模拟器 的状态。 adb ( Android Debug Bridge)是一个通用命令行工具,其允许您与模拟器实例或连接的 Android 设备进行通信。它可为各种设备操作提供便利,如安装和调试…...
三维GIS开发cesium智慧地铁教程(5)Cesium相机控制
一、环境搭建 <script src"../cesium1.99/Build/Cesium/Cesium.js"></script> <link rel"stylesheet" href"../cesium1.99/Build/Cesium/Widgets/widgets.css"> 关键配置点: 路径验证:确保相对路径.…...
Redis相关知识总结(缓存雪崩,缓存穿透,缓存击穿,Redis实现分布式锁,如何保持数据库和缓存一致)
文章目录 1.什么是Redis?2.为什么要使用redis作为mysql的缓存?3.什么是缓存雪崩、缓存穿透、缓存击穿?3.1缓存雪崩3.1.1 大量缓存同时过期3.1.2 Redis宕机 3.2 缓存击穿3.3 缓存穿透3.4 总结 4. 数据库和缓存如何保持一致性5. Redis实现分布式…...
理解 MCP 工作流:使用 Ollama 和 LangChain 构建本地 MCP 客户端
🌟 什么是 MCP? 模型控制协议 (MCP) 是一种创新的协议,旨在无缝连接 AI 模型与应用程序。 MCP 是一个开源协议,它标准化了我们的 LLM 应用程序连接所需工具和数据源并与之协作的方式。 可以把它想象成你的 AI 模型 和想要使用它…...
连锁超市冷库节能解决方案:如何实现超市降本增效
在连锁超市冷库运营中,高能耗、设备损耗快、人工管理低效等问题长期困扰企业。御控冷库节能解决方案通过智能控制化霜、按需化霜、实时监控、故障诊断、自动预警、远程控制开关六大核心技术,实现年省电费15%-60%,且不改动原有装备、安装快捷、…...
测试markdown--肇兴
day1: 1、去程:7:04 --11:32高铁 高铁右转上售票大厅2楼,穿过候车厅下一楼,上大巴车 ¥10/人 **2、到达:**12点多到达寨子,买门票,美团/抖音:¥78人 3、中饭&a…...
解决本地部署 SmolVLM2 大语言模型运行 flash-attn 报错
出现的问题 安装 flash-attn 会一直卡在 build 那一步或者运行报错 解决办法 是因为你安装的 flash-attn 版本没有对应上,所以报错,到 https://github.com/Dao-AILab/flash-attention/releases 下载对应版本,cu、torch、cp 的版本一定要对…...
Ascend NPU上适配Step-Audio模型
1 概述 1.1 简述 Step-Audio 是业界首个集语音理解与生成控制一体化的产品级开源实时语音对话系统,支持多语言对话(如 中文,英文,日语),语音情感(如 开心,悲伤)&#x…...
IT供电系统绝缘监测及故障定位解决方案
随着新能源的快速发展,光伏电站、储能系统及充电设备已广泛应用于现代能源网络。在光伏领域,IT供电系统凭借其持续供电性好、安全性高等优势成为光伏首选,但在长期运行中,例如老化、潮湿、隐裂、机械损伤等问题会影响光伏板绝缘层…...
Map相关知识
数据结构 二叉树 二叉树,顾名思义,每个节点最多有两个“叉”,也就是两个子节点,分别是左子 节点和右子节点。不过,二叉树并不要求每个节点都有两个子节点,有的节点只 有左子节点,有的节点只有…...
