当前位置: 首页 > news >正文

基于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的服务);
用户100
用户200
2、分别启动消息源服务B和中间服务A
此时在服务B控制台我们可以看到:
在这里插入图片描述
我们模拟的消息发送已经在给用户100和用户200发送,因为我们的用户100和用户200均没有与中间服务A建立连接,故此时测试界面看不到消息;
当我们在用户100的模拟界面点击“开启连接”后,可以在右侧看到发给用户100的模拟消息:
在这里插入图片描述

之后我们再打开用户200的连接:
在这里插入图片描述

好了,到这里就结束了,有任何问题请积极指出,此例子只是个例子,并未经受任何生产的测试,欢迎讨论沟通:)

相关文章:

基于springboot websocket和okhttp实现消息中转

1、业务介绍 消息源服务的消息不能直接推给用户侧&#xff0c;用户与中间服务建立websocket连接&#xff0c;中间服务再与源服务建立websocket连接&#xff0c;源服务的消息推给中间服务&#xff0c;中间服务再将消息推送给用户。流程如下图&#xff1a; 此例中我们定义中间服…...

@PostConstruct 注解的方法用于资源的初始化

PostConstruct 是 Java EE 5 引入的一个注解&#xff0c;主要用于依赖注入完成之后&#xff0c;需要执行的方法上。这个注解的方法会在依赖注入完成后自动被容器&#xff08;如 EJB 容器或 Spring 容器&#xff09;调用&#xff0c;并且只会被调用一次。 PostConstruct 注解的…...

(一)SvelteKit教程:hello world

&#xff08;一&#xff09;SvelteKit教程&#xff1a;hello world sveltekit 的官方教程&#xff0c;在这里&#xff1a;Creating a project • Docs • SvelteKitCreating a project • Docs • SvelteKit 我们可以按照如下的步骤来创建一个项目&#xff1a; npm create s…...

华为Atlas NPU ffmpeg 编译安装

处理器&#xff1a;鲲鹏920 NPU&#xff1a;昇腾 310P3 操作系统&#xff1a;Kylin Linux Advanced Server V10 CANN&#xff1a;Ascend-cann-toolkit_8.0.RC1_linux-aarch64.run FFmpeg&#xff1a;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. 功能和工具&#xff1a; Sass&#xff1a;提供了更多的功能和内置方法&#xff0c;如条件语句、循环、数学函数等。Sass 也支持更复杂的操作和逻辑构建。 Less&#xff1a;功能也很强大&#xff0c;但相比之下&#xff0c;Sass 在功能上更为丰富和成熟。 2、编译环境&…...

力扣-2663

题目 如果一个字符串满足以下条件&#xff0c;则称其为 美丽字符串 &#xff1a; 它由英语小写字母表的前 k 个字母组成。它不包含任何长度为 2 或更长的回文子字符串。 给你一个长度为 n 的美丽字符串 s 和一个正整数 k 。 请你找出并返回一个长度为 n 的美丽字符串&#…...

CausalMMM:基于因果结构学习的营销组合建模

1. 摘要 在线广告中&#xff0c;营销组合建模&#xff08;Marketing Mix Modeling&#xff0c;MMM&#xff09; 被用于预测广告商家的总商品交易量&#xff08;GMV&#xff09;&#xff0c;并帮助决策者调整各种广告渠道的预算分配。传统的基于回归技术的MMM方法在复杂营销场景…...

编译 CUDA 程序的基本知识和步骤

基本工具 NVCC&#xff08;NVIDIA CUDA Compiler&#xff09;: nvcc 是 NVIDIA 提供的 CUDA 编译器&#xff0c;用于将 CUDA 源代码&#xff08;.cu 文件&#xff09;编译成可执行文件或库。它可以处理 CUDA 和主机代码&#xff08;例如 C&#xff09;的混合编译。nvcc 调用底层…...

[SAP ABAP] 排序内表数据

语法格式 整表排序 SORT <itab> [ASCENDING|DESCENDING]. 按指定字段排序 SORT <itab> BY f1 [ASCENDING|DESCENDING] f2 [ASCENDING|DESCENDING] ... fn [ASCENDING|DESCENDING].<itab>&#xff1a;代表内表 不指定排序方式则默认升序排序 示例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、对操作建模 一个活动图从本质上说是一个流程图&#xff0c;展现从活动到活动的控制流 活动图…...

【web2】jquary,bootstrap,vue

文章目录 1.jquary&#xff1a;选择器1.1 jquery框架引入&#xff1a;$("mydiv") 当成id选择器1.2 jquery版本/对象&#xff1a;$(js对象) -> jquery对象1.3 jquery的页面加载事件&#xff1a;$ 想象成 window.onload 1.4 jquery的基本选择器&#xff1a;$()里内容…...

独角兽品牌獭崎酱酒:高性价比的酱香之选

在酱香型白酒领域中&#xff0c;獭崎酱酒以其独特的品牌定位和高性价比迅速崛起&#xff0c;成为市场上备受关注的独角兽品牌。作为贵州茅台镇的一款新秀酱香酒&#xff0c;獭崎酱酒不仅传承了百年酿造工艺&#xff0c;还以创新的商业模式和亲民的价格赢得了广大消费者的青睐。…...

java打印菱形和空心菱形

java打印菱形 菱形分上下两个部分。其中上部分同打印金字塔&#xff1b;下部分循环部分i是递减 &#xff08;ps:菱形层数只能为奇数&#xff09; import java.util.Scanner;public class Lingxing{public static void main(String[] args) {Scanner myScanner new Scanner(S…...

Day10 —— 大数据技术之Scala

Scala编程入门 Scala的概述什么是Scala&#xff1f;Scala的重要特点Scala的使用场景 Scala的安装Scala基础Scala总结 Scala的概述 什么是Scala&#xff1f; Scala是一种将面向对象和函数式编程结合在一起的高级语言&#xff0c;旨在以简洁、优雅和类型安全的方式表达通用编程…...

Linux应用系统快速部署:docker快速部署linux应用程序

目录 一、背景 &#xff08;一&#xff09;引入docker的起因 &#xff08;二&#xff09;docker介绍 &#xff08;三&#xff09;Docker部署的优势 1、轻量级和可移植性 2、快速部署和扩展 3、一致性 4、版本控制 5、安全性 6、资源隔离 7、简化团队协作 8、多容器…...

三目运算符中间的表达式可以省略吗(a?:c)?

熟悉C语言的童靴对三目运算符都非常熟悉&#xff0c;a? b : c; 如果a为true&#xff0c;则整个运算符的值为b,否则为c;那么问题来了&#xff0c;三目运算符中间的表达式可以省略吗?即a? : c; 1、linux内核中出现的省略情况 本人在阅读内核代码是发现了下面的代码: preferr…...

android 彩虹进度条自定义view实现

实现一个彩虹色进度条功能&#xff0c;不说明具体用途大家应该能猜到。想找别人造的轮子&#xff0c;但是没有合适的&#xff0c;所以决定自己实现一个。 相关知识 android 自定义view LinearGradient 线性渐变 实现步骤 自定义view 自定义一个TmcView类继承View 重写两…...

免费一年SSL证书申请——建议收藏

免费一年SSL证书申请——建议收藏 获取免费一年期SSL证书其实挺简单的 准备你的网站&#xff1a; 确保你的网站已经有了域名&#xff0c;而且这个域名已经指向你的服务器。还要检查你的服务器支持HTTPS&#xff0c;也就是443端口要打开&#xff0c;这是HTTPS默认用的。 验证域…...

【docker1】指令,docker-compose,Dockerfile

文章目录 1.pull/image&#xff0c;run/ps&#xff08;进程&#xff09;&#xff0c;exec/commit2.save/load&#xff1a;docker save 镜像id&#xff0c;不是容器id3.docker-compose&#xff1a;多容器&#xff1a;宿主机&#xff08;eth0网卡&#xff09;安装docker会生成一…...

玩转openrgb

缘由我的asus b760m有rgb&#xff0c;但是华硕Armoury Crate 确实比较臃肿&#xff0c;经常啥也没干它占用3-5%。而开源界有个openrgb&#xff0c;虽然看似简陋但是它小啊。于是采用python脚本openrgb来玩转它。本方案应该也适用于其他rgb主板。准备工作1、下载openrgb&#xf…...

小白必看:霜儿-汉服-造相Z-Turbo从部署到出图全流程解析

小白必看&#xff1a;霜儿-汉服-造相Z-Turbo从部署到出图全流程解析 1. 镜像简介与核心优势 霜儿-汉服-造相Z-Turbo是一款专为汉服写真生成优化的AI模型镜像&#xff0c;基于Xinference框架部署&#xff0c;通过Gradio提供简洁易用的Web界面。与通用文生图模型相比&#xff0…...

leetcode 1550. 存在连续三个奇数的数组-耗时100-Three Consecutive Odds

Problem: 1550. 存在连续三个奇数的数组-耗时100-Three Consecutive Odds 耗时100%&#xff0c;检查连续的三个数字是否奇数 Code class Solution { public:bool threeConsecutiveOdds(vector<int>& arr) {int n arr.size();for(int i 0; i < n - 2; i) {if((a…...

芯片缺货潮下的应对策略与国产替代方案

1. 芯片缺货潮下的行业现状最近我的一个产品项目中&#xff0c;原本采购价仅5元的ST品牌MCU&#xff08;微控制器&#xff09;价格飙升至70元&#xff0c;涨幅高达14倍。这个案例并非个例&#xff0c;而是当前全球半导体行业供应链危机的缩影。作为从业十余年的硬件工程师&…...

30个核心概念一次讲明白,小白也能轻松入门大模型(收藏版)

这几年&#xff0c;AI 几乎成了人人都在谈的话题。 有人在聊大模型&#xff0c;有人在说智能体&#xff0c;有人担心算力不够&#xff0c;也有人被“参数”、“微调”、“多模态”、“RAG”这些词绕得头晕。 结果就是&#xff1a;听了很多&#xff0c;越听越乱。 这篇文章是用尽…...

秒杀系统主库宕机不丢单方案-03-本地消息表

秒杀系统主库宕机不丢单方案&#xff1a;本地消息表&#xff08;事务分离补偿机制&#xff09; 方案概述 本地消息表方案通过在应用层引入消息表机制&#xff0c;将事务操作与消息发送分离&#xff0c;实现最终一致性。该方案是秒杀系统主库宕机不丢单的兜底设计&#xff0c;即…...

YouTube视频一直转圈?加载卡顿原因分析与排查方法(2026)

在日常开发或使用在线视频平台时&#xff0c;常见一个问题&#xff1a;视频播放过程中出现持续加载、卡顿甚至无法播放的情况。这类问题并不一定由带宽不足引起&#xff0c;而往往与浏览器、网络链路以及设备性能等多方面因素有关。本文从技术角度出发&#xff0c;对视频加载流…...

马年市场快报分析:欧美组合式一氧化碳及可燃气体报警器指南

马年市场快报分析&#xff1a;欧美组合式一氧化碳及可燃气体报警器指南根据您提供的快报内容&#xff0c;我将从专业角度逐步分析欧美组合式一氧化碳&#xff08;CO&#xff09;及可燃气体报警器的关键信息&#xff0c;包括安全标准、风险因素、探测器区别、安装建议以及相关产…...

三极管信号滤波原理与工程实践

1. 三极管在信号滤波中的独特应用作为一名嵌入式硬件工程师&#xff0c;我经常需要处理各种传感器信号。最近在无刷电机驱动项目中&#xff0c;遇到了霍尔信号毛刺干扰的问题。传统教科书上总是强调三极管的放大作用&#xff0c;但实际工程中&#xff0c;我发现三极管在信号滤波…...

PX4固件二次开发入门:从源码结构到第一个自定义模块(基于v1.11版本)

PX4固件二次开发实战&#xff1a;从源码解析到自定义模块开发&#xff08;v1.11版本&#xff09; 当你第一次打开PX4的源码仓库&#xff0c;面对数十个文件夹和数千个文件时&#xff0c;那种扑面而来的压迫感我深有体会。作为过来人&#xff0c;我想分享一套系统性的二次开发方…...