基于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会生成一…...
【JVM】- 内存结构
引言 JVM:Java Virtual Machine 定义:Java虚拟机,Java二进制字节码的运行环境好处: 一次编写,到处运行自动内存管理,垃圾回收的功能数组下标越界检查(会抛异常,不会覆盖到其他代码…...
【解密LSTM、GRU如何解决传统RNN梯度消失问题】
解密LSTM与GRU:如何让RNN变得更聪明? 在深度学习的世界里,循环神经网络(RNN)以其卓越的序列数据处理能力广泛应用于自然语言处理、时间序列预测等领域。然而,传统RNN存在的一个严重问题——梯度消失&#…...
oracle与MySQL数据库之间数据同步的技术要点
Oracle与MySQL数据库之间的数据同步是一个涉及多个技术要点的复杂任务。由于Oracle和MySQL的架构差异,它们的数据同步要求既要保持数据的准确性和一致性,又要处理好性能问题。以下是一些主要的技术要点: 数据结构差异 数据类型差异ÿ…...
ETLCloud可能遇到的问题有哪些?常见坑位解析
数据集成平台ETLCloud,主要用于支持数据的抽取(Extract)、转换(Transform)和加载(Load)过程。提供了一个简洁直观的界面,以便用户可以在不同的数据源之间轻松地进行数据迁移和转换。…...
基于Docker Compose部署Java微服务项目
一. 创建根项目 根项目(父项目)主要用于依赖管理 一些需要注意的点: 打包方式需要为 pom<modules>里需要注册子模块不要引入maven的打包插件,否则打包时会出问题 <?xml version"1.0" encoding"UTF-8…...
全面解析各类VPN技术:GRE、IPsec、L2TP、SSL与MPLS VPN对比
目录 引言 VPN技术概述 GRE VPN 3.1 GRE封装结构 3.2 GRE的应用场景 GRE over IPsec 4.1 GRE over IPsec封装结构 4.2 为什么使用GRE over IPsec? IPsec VPN 5.1 IPsec传输模式(Transport Mode) 5.2 IPsec隧道模式(Tunne…...
代码随想录刷题day30
1、零钱兑换II 给你一个整数数组 coins 表示不同面额的硬币,另给一个整数 amount 表示总金额。 请你计算并返回可以凑成总金额的硬币组合数。如果任何硬币组合都无法凑出总金额,返回 0 。 假设每一种面额的硬币有无限个。 题目数据保证结果符合 32 位带…...
Java编程之桥接模式
定义 桥接模式(Bridge Pattern)属于结构型设计模式,它的核心意图是将抽象部分与实现部分分离,使它们可以独立地变化。这种模式通过组合关系来替代继承关系,从而降低了抽象和实现这两个可变维度之间的耦合度。 用例子…...
【电力电子】基于STM32F103C8T6单片机双极性SPWM逆变(硬件篇)
本项目是基于 STM32F103C8T6 微控制器的 SPWM(正弦脉宽调制)电源模块,能够生成可调频率和幅值的正弦波交流电源输出。该项目适用于逆变器、UPS电源、变频器等应用场景。 供电电源 输入电压采集 上图为本设计的电源电路,图中 D1 为二极管, 其目的是防止正负极电源反接, …...
RabbitMQ入门4.1.0版本(基于java、SpringBoot操作)
RabbitMQ 一、RabbitMQ概述 RabbitMQ RabbitMQ最初由LShift和CohesiveFT于2007年开发,后来由Pivotal Software Inc.(现为VMware子公司)接管。RabbitMQ 是一个开源的消息代理和队列服务器,用 Erlang 语言编写。广泛应用于各种分布…...
