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

【Rust基础】使用Rocket构建基于SSE的流式回复

news 来源：原创 2025/8/22 1:53:15

背景

我们正在使用Rust开发基于RAG的知识库系统，其中对于模型的回复使用了常用的SSE，Web框架使用Rocket，Rocket提供了一个简单的方式支持SSE，但没有会话保持、会话恢复等功能，因此我们自己简单实现这两个功能。

在这里插入图片描述

使用Rocket推送消息流

以下，是Rocket给出的示例：

#[get("/text/stream")]
fn stream() -> TextStream![String] {TextStream! {let mut interval = time::interval(Duration::from_secs(1));for i in 0..10 {yield format!("n: {}", i);interval.tick().await;}}
}

我们需要改造这个示例，以满足将模型回复的消息推送给前端的需求。

首先，对于既然推流，需要知道将流推送给谁，也就是要推送到哪个会话中，所以我们在发起会话的时候，需要一个会话ID来标识一个唯一的会话。

我们使用sse.js这个库作为SSE的客户端，用于发起SSE连接，该库可以通过发起POST请求来建立连接，可以携带额外的数据和请求头。

使用以下结构来接收一个对话请求：

pub struct ChatMessageReq {/// 会话IDpub session_id: String,/// 消息内容pub content: String,
}

于是我们的接口需要修改为这样：

#[post("/chat", data = "<req>")]
async fn question(req: Json<ChatMessageReq>) -> (ContentType, TextStream<impl Stream<Item = String>>) 
{//TODO
}

其中TextStream<impl Stream<Item = String>>等价于TextStream![String]。

需要注意的是，如果返回值没有指定ContentType，那么Rocket默认响应的ContentType是文本类型，而非stream类型，会导致前端无法解析。

接下来我们实现会话管理功能。

我们定义了一个名为SsePool的struct，来存储并管理SSE连接：

struct SsePool {/// 消息流传输通道channel: DashMap<String, Sender<SseEvent>>,
}impl SsePool {/// 初始化连接池pub fn init() -> Self {Self {channel: DashMap::new(),}}/// 移除连接fn remove(&self, id: &String) {if let Some((_, sender)) = self.channel.remove(id) {drop(sender);}}/// 获取连接fn get_sender(&self, id: &String) -> Option<Sender<SseEvent>> {self.channel.get(id).map(|v| v.value().clone())}/// 新建channelpub fn new_channel(&self, id: String) -> (Sender<SseEvent>, Receiver<SseEvent>) {let (sender, receiver) = tokio::sync::mpsc::channel(10_0000);// 获取并移除旧senderlet old_sender = self.channel.remove(&id).map(|(_, s)| s);// 插入新senderself.channel.insert(id, sender.clone());// 处理旧senderif let Some(old_sender) = old_sender {tokio::spawn(async move {// 发送终止信号let _ = old_sender.send(SseEvent::Abort).await;});}(sender, receiver)}/// 发送消息pub async fn send_message(&self, id: &String, message: ChatMessage) {if let Some(sender) = self.get_sender(id) {if let Err(e) = sender.send(SseEvent::ChatMessage(message)).await {log::warn!("消息发送失败，session id: {}，失败原因：{}", &id, e);};// drop(sender);}}
}

其中channel使用的是tokio中mpsc的channel。

值得注意的是，new_channel中，新建连接时，需要向channel发送一条终止事件，确保已有的receiver关闭，返回新的receiver，这一点是用于后续的会话恢复使用。new_channel会返回receiver和sender，用于消息接收和发送。

当收到模型回复是，调用SsePool::send_message发送消息到channel，再头通过receiver接收消息，转发到前端。

可以把它初始化到静态变量中，方便全局调用：

static SSE_POOL: LazyLock<SsePool> = LazyLock::new(|| SsePool::init());

于是，我们的接口可以完善为以下内容：

#[post("/chat", data = "<req>")]
async fn question(req: Json<ChatMessageReq>,
) -> (ContentType, TextStream<impl Stream<Item = String>>){// 请求新消息，并返回receiverlet (_, _, mut receiver) = service::new_message(req).await.unwrap();let stream = TextStream! {// 持续接收receiver的消息，然后推送到前端while let Some(item) = receiver.recv().await {match item{//模型回复的消息SseEvent::ChatMessage(message) => {yield SseEvent::ChatMessage(message.clone()).to_message();if SseEvent::is_done(&message) {// 推送消息yield SseEvent::Abort.to_message();break;}},// 关闭通道SseEvent::Abort => {yield SseEvent::Abort.to_message();break;},_ => {}}}yield SseEvent::Abort.to_message();drop(receiver);};(ContentType::EventStream, stream)
}

至此，新会话的接口就完成了。

接下来是会话的恢复。

当前端切换会话或刷新页面时，我们希望能够继续收到未回复完成的消息，所以需要一个用于会话恢复的接口。同样的，接口需要会话ID来区分恢复哪一个会话。

#[post("/resume-stream", data = "<req>")]
async fn resume_stream(req: Json<ResumeStreamReq>,
) -> (ContentType, TextStream<impl Stream<Item = String>>){// 会话IDlet session_id = req.session_id.clone();let stream = TextStream! {// 尝试恢复会话，并返回receiver，如果能够返回receiver说明会话未完成，否则已经完成if let Some(mut receiver) = service::resume_stream(&req.session_id).await.unwrap(){// 持续接收未回复完成的消息while let Some(item) = receiver.recv().await {match item {// 模型回复的消息SseEvent::ChatMessage(message) => {yield SseEvent::ChatMessage(message.clone()).to_message();if SseEvent::is_done(&message) {yield SseEvent::Abort.to_message();break;}}// 关闭通道SseEvent::Abort => {yield SseEvent::Abort.to_message();break;}_ => {}}}drop(receiver);}yield SseEvent::Abort.to_message();};(ContentType::EventStream, stream)
}

在service::resume_stream中，首先检查对应会话ID的channel是否存在，存在则新建channel返回receiver，不存在则表明已经回复完成。

pub async fn resume_stream(session_id: &String,
) -> AppResult<Option<Receiver<SseEvent>>> {if let None = chat::get_connection(session_id) {return Ok(None);}// 获取会话对应的channel，如果channel存在则标识消息仍在回复中let (_, receiver) = chat::new_connection(session_id);Ok(Some(receiver))
}

至此，便实现了会话恢复，刷新页面后仍能后接收strem消息。

总结

使用Rust写这些业务代码的速度，终归是没有Java快，一些常用的库，没有Java系列封装的简单易用，不过应用占用资源确实比Java小很多。

本次使用的一些库：

tokio：异步运行环境，以及mpsc的channel，
dashmap：支持并发的hashmap，但是使用不当容易造成死锁。

【Rust基础】使用Rocket构建基于SSE的流式回复

背景我们正在使用Rust开发基于RAG的知识库系统，其中对于模型的回复使用了常用的SSE，Web框架使用Rocket，Rocket提供了一个简单的方式支持SSE，但没有会话保持、会话恢复等功能，因此我们自己简单实现这两个功能。使用R…...

编程日记 2025/8/22 1:53:15

一种改进的CFAR算法用于目标检测（解决多目标掩蔽）

摘要恒虚警率（CFAR）技术在雷达自动检测过程中起着关键作用。单元平均（CA）CFAR算法在几乎所有的多目标情况下都会受到掩蔽效应的影响。最小单元平均（SOCA）CFAR算法仅当干扰目标位于参考窗口的前后方时才具有…...

编程日记 2025/8/22 1:46:48

什么是人工智能芯片？

行业专家指出，许多智能设备和物联网设备都是由某种形式的人工智能(AI)驱动的——无论是语音助理、面部识别摄像头，还是电脑。这些设备需要采用某种技术为它们进行的数据处理提供支持。有些设备需要在云平台的大型数据中心处理数据，而也有一些…...

编程日记 2025/8/22 1:50:43

0.深入探秘 Rust Web 框架 Axum

在当今的 Web 开发领域，Rust 凭借其出色的性能、内存安全性和并发处理能力，正逐渐崭露头角。而 Axum 作为 Rust 生态系统中一款备受瞩目的 Web 框架，更是为开发者提供了高效、灵活且强大的工具，用于构建现代化的 Web 应用程序。本…...

编程日记 2025/8/22 1:46:21

深度监听 ref 和 reactive 的区别详解

深度监听 ref 和 reactive 的区别详解一、ref 的深度监听（示例代码）关键点：1. ref 的存储方式：2. 监听 ref 的特性二、reactive 的深度监听（示例代码）关键点：1. reactive 的深度响应性2. 监听…...

编程日记 2025/8/22 1:47:42

面向对象—有理数类的设计

目录 1.代码呈现 1.1编写toString、equals方法 1.2测试代码 1.3有理数类的代码 2.论述题 3.有理类设计 1.代码呈现 1.1编写toString、equals方法 (1)toString方法 Overridepublic String toString(){if(this.v20){return "Undefined";}return this.v1 "/…...

编程日记 2025/8/21 19:25:03

OpenHarmony Camera开发指导（四）：相机会话管理（ArkTS）

概述相机在使用预览、拍照、录像、获取元数据等功能前，都需要先创建相机会话。相机会话Session的功能如下： 配置相机的输入流和输出流。配置输入流即添加设备输入，通俗来讲即选择某一个摄像头进行拍照录像；配置输出流&#x…...

编程日记 2025/8/22 1:52:50

Linux电源管理(三)，CPUIdle 和 ARM的PSCI

更多linux系统电源管理相关的内容请看：Linux电源管理、功耗管理和发热管理 (CPUFreq、CPUIdle、RPM、thermal、睡眠和唤醒)-CSDN博客 1 简介 Linux下的空闲进程cpuidle在内核中是一个子系统。cpuidle子系统所需要做的事情就是在CPU进入idle状态后，…...

编程日记 2025/8/21 2:34:36

【测试工具】JMeter使用小记

JMeter 使用小记下载与安装 jdk 下载地址：https://www.oracle.com/java/technologies/downloads/#jdk18-windowsJMeter 下载地址：https://jmeter.apache.org/download_jmeter.cgi 教程参考：JMeter下载及安装详细教程-CSDN博客设置中文界…...

编程日记 2025/8/19 1:40:40

Obsidian的简单使用

一、安装并配置仓耳今楷字体优化阅读体验，个人实测觉得正文用仓耳今楷04-W03最合适（前面的数字代表字体，数字越大，越偏向于楷体，而01就很像黑体。后面的数字代表粗细，正常粗细是W03，最粗是W0…...

编程日记 2025/8/19 1:40:34

docker的基础知识

Docker https://www.yuque.com/leifengyang/sutong 下载镜像检索： docker search下载: docker pull列表: docker images删除 docker rmi启动容器运行： docker run查看： docker ps停止： docker stop启动： …...

编程日记 2025/8/17 18:27:59

PcVue助力立讯：精密制造的智能化管控实践！

PcVue助力立讯： 精密制造的智能化管控实践！ 客户介绍立讯精密（Luxshare ICT，股票代码：002475）成立于2004年5月24日，专注于为消费电子产品、汽车领域产品以及企业通讯产品提供从核心零部件、…...

编程日记 2025/8/17 21:17:06

深度学习-157-Dify工具之创建知识库

文章目录 1 硅基流动1.1 模型广场1.1.1 对话模型(免费)1.1.2 嵌入模型(免费)1.1.3 重排序模型(免费)1.2 模型调用1.2.1 文本对话1.2.2 文本嵌入2 构建知识库2.1 准备文档2.2 点击创建知识库2.3 设置嵌入参数2.4 召回测试3 创建聊天助手3.1 仅使用大模型3.2 结合知识库的大模型3…...

编程日记 2025/8/21 4:46:29

Oracle--安装Oracle Database23ai Free

前言：本博客仅作记录学习使用，部分图片出自网络，如有侵犯您的权益，请联系删除官方文档： Get Started with Oracle Database 23ai | Oracle 一、安装的环境要求本文同步使用Oracle Linux9的虚拟机进行操作 1、Orac…...

编程日记 2025/8/21 11:13:26

【JavaEE初阶】多线程重点知识以及常考的面试题－多线程进阶（三）

本篇博客给大家带来的是集合类在多线程下的使用和死锁的知识点还包括常见的面试题. 🐎文章专栏: JavaEE初阶 🚀若有问题评论区见 ❤ 欢迎大家点赞评论收藏分享如果你不知道分享给谁,那就分享给薯条. 你们的支持是我不断创作的动力 . 王子,公主请阅&…...

编程日记 2025/8/20 7:28:07

【verilog】多个 if 控制同一个变量（后面会覆盖前面）非阻塞赋值真的并行吗？

非阻塞赋值 (<) 是“并行”的，但是代码顺序会影响结果？”这正是 Verilog 的硬件描述本质 vs 行为语义之间的微妙之处。 💡1. 非阻塞赋值真的并行吗？ 是的！非阻塞赋值 < 从行为上是并行的，也就是说&a…...

编程日记 2025/8/19 2:04:06

C++事件驱动编程从入门到实战：深入理解与高效应用

C事件驱动编程从入门到实战：深入理解与高效应用在现代软件开发中，事件驱动编程（Event-Driven Programming）作为一种流行的编程范式，被广泛应用于图形用户界面（GUI）、网络通信、游戏开发等众多…...

编程日记 2025/8/17 19:57:42

问题｜ MATLAB比Python更有优势的特定领域

以下是关于MATLAB在特定领域相较于Python的优势的详细分析，结合其核心功能、行业应用及技术特性展开论述： 一、科学研究与工程计算 1. 数值计算的高效性 MATLAB的核心设计围绕矩阵运算展开，其底层对线性代数和数值计算进行了深度优化。例如…...

编程日记 2025/8/21 11:46:53

黑马商城项目(三)微服务

一、单体架构测试高并发软件二、微服务三、SpringCloud 四、微服务拆分黑马商城模块： 服务拆分原则： 拆分服务： 独立project： maven聚合： 拆分案例： 远程调用： package com.hmall.cart.…...

编程日记 2025/8/20 5:03:24

Qt界面卡住变慢的解决方法

本质原因: 当Qt界面出现卡顿或无响应时，通常是因为主线程（GUI线程）被耗时操作阻塞。完全忘了。。。 Qt Creater解决方法 1. 定位耗时操作目标：找到阻塞主线程的代码段。方法： 使用QElapsedTimer测量代码执行时间…...

编程日记 2025/8/21 10:09:52

Flutter的原理及美团的实践（下）

Flutter的原理及性能实践 Flutter和原生性能对比虽然使用原生实现（左）和Flutter实现（右）的全品类页面在实际使用过程中几乎分辨不出来： 但是我们还需要在性能方面有一个比较明确的数据对比。我们最关心的两个页面…...

编程日记 2025/8/20 2:06:49

时序预测 | Matlab实现基于VMD-WOA-ELM和VMD-ELM变分模态分解结合鲸鱼算法优化极限学习机时间序列预测

时序预测 | Matlab实现基于VMD-WOA-ELM和VMD-ELM变分模态分解结合鲸鱼算法优化极限学习机时间序列预测目录时序预测 | Matlab实现基于VMD-WOA-ELM和VMD-ELM变分模态分解结合鲸鱼算法优化极限学习机时间序列预测预测效果基本介绍程序设计参考资料预测效果基本介绍 1.Matlab…...

编程日记 2025/8/19 11:23:25

【云安全】云原生- K8S IngressNightmare CVE-2025-1974（漏洞复现完整教程）

漏洞原理 CVE-2025-1974: The IngressNightmare in Kubernetes | Wiz Blog 分两方面： a、配置注入过程构造一个恶意的Ingress资源，其中注入ssl_engine指令指向恶意共享库向准入控制器验证端点（AdmissionWebhook）发送Admissio…...

编程日记 2025/8/17 1:45:13

上篇文章： Tomcat与Servlethttps://blog.csdn.net/sniper_fandc/article/details/147278469?fromshareblogdetail&sharetypeblogdetail&sharerId147278469&sharereferPC&sharesourcesniper_fandc&sharefromfrom_link 上篇文章介绍了To…...

编程日记 2025/8/21 19:08:47

在高数据速度下确保信号完整性的 10 个关键策略

随着越来越多的传感器连接到系统，需要快速、可靠和安全地传输更多数据，对带宽和设计复杂性的需求也在增加。优先考虑的是确保从 A 发送到 B 的信号不会失真。确保信号完整性对于设计依赖于持续准确数据流的数据密集型应用程序的工程师来说&#xff0c…...

编程日记 2025/8/17 20:03:07

2025华中杯数学建模B题完整分析论文（共42页）（含模型、数据、可运行代码）

2025华中杯大学生数学建模B题完整分析论文目录一、问题重述二、问题分析三、模型假设四、模型建立与求解 4.1问题1 4.1.1问题1解析 4.1.2问题1模型建立 4.1.3问题1样例代码（仅供参考） 4.1.4问题1求解结果（仅供参考&am…...

编程日记 2025/8/20 22:37:51

UE5 自带的视频播放器

文章目录文件夹准备添加一个文件媒体源方法1方法2 添加一个视频播放器播放视频直接播放使用网格体播放使用UI播放播放视频的音乐媒体播放器常用的节点设置循环是用绝对路径播放视频，视频无需导入注册播放完成事件文件夹准备视频必须被放在Content/Moveis文件下…...

编程日记 2025/8/17 18:25:11

是德科技E5080B网络分析仪深度评测：5G/车载雷达测试实战指南

是德科技E5080B网络分析仪（ENA系列）是一款高性能射频测试仪器，广泛应用于通信、航空航天、半导体等领域，以下是其核心功能详解： 一、核心测试功能多参数网络分析 S参数测量：支持全双端口S参数测试&#xf…...

编程日记 2025/8/17 10:11:16

javaSE————网络编程套接字

网络编程套接字~~~~~ 好久没更新啦，蓝桥杯爆掉了，从今天开始爆更嗷； 1，网络编程基础为啥要有网络编程呢，我们进行网络通信就是为了获取丰富的网络资源，说实话真的很神奇，想想我们躺在床上&a…...

编程日记 2025/8/17 19:52:39

力扣349 == 两个数组交集的两种解法

目录解法一：利用 Set 特性高效去重解法二：双重遍历与 Set 去重方法对比与总结关键点总结题目描述给定两个整数数组 nums1 和 nums2，要求返回它们的交集。输出结果中的每个元素必须是唯一的，且顺序不限。示例输入&…...

编程日记 2025/8/19 22:24:17

笔试专题（十）

文章目录对称之美（双指针）题解代码连续子数组最大和（线性dp）题解代码最长回文子序列（区间dp）题解代码对称之美（双指针） 题目链接题解 1. 双指针 2. 用left标记左边的字符串…...

编程日记 2025/8/17 18:22:59

YOLOv12即插即用---RFAConv

1.模块介绍接受域注意卷积（RFAConv）：更聪明地感知空间特征在传统卷积神经网络中，卷积核参数的共享机制虽有效提升了模型的泛化能力与计算效率，但却忽略了不同空间位置特征在感知范围（即接受域）内的重要性差异。为此，我们提出了一种更具感知能力的模块 —— 接受域注…...

编程日记 2025/8/19 22:01:58

使用datax通过HbaseShell封装writer和reader同步hbase数据到hbase_踩坑_细节总结---大数据之DataX工作笔记008

最近在做大数据相关功能,有个需求,使用datax同步hbase到hbase中,其中还是有很多细节值得记录: 首先来看一下datax的源码中,如果你使用phoenix创建的表,那么你就需要使用对应的hbase带有sql字样的,reader和writer. 然后如果你使用datax-web来进行测试的,那么,他默认使用的是h…...

编程日记 2025/8/19 21:05:27

Python解决“小D的abc字符变换”问题

小D的“abc”变换问题问题描述测试样例解题思路代码问题描述小D拿到了一个仅由 “abc” 三种字母组成的字符串。她每次操作会对所有字符同时进行以下变换： 将 ‘a’ 变成 ‘bc’ 将 ‘b’ 变成 ‘ca’ 将 ‘c’ 变成 ‘ab’ 小D将重复该操作 k 次。你的任务是输…...

编程日记 2025/8/21 6:41:47

C++学习：六个月从基础到就业——面向对象编程：重载运算符（下）

C学习：六个月从基础到就业——面向对象编程：重载运算符（下） 本文是我C学习之旅系列的第十三篇技术文章，是面向对象编程中运算符重载主题的下篇。本篇文章将继续深入探讨高级运算符重载技术、特殊运算符、常见应用场景和…...

编程日记 2025/8/17 20:15:48

电压模式控制学习

电压模式控制在开关电源中，大的可分为三大控制模式，分别是电压模式控制，电流模式控制，迟滞模式控制。今天简要介绍下电压模式控制的优缺点。原理架构图如下如图所示，电压模式控制可以分为三部分：误…...

编程日记 2025/8/19 11:34:25

vue3 Ts axios 封装

vue3 Ts axios 封装 axios的封装 import axios, { AxiosError, AxiosInstance, InternalAxiosRequestConfig, AxiosResponse, AxiosRequestConfig, AxiosHeaders } from axios import qs from qs import { config } from ./config import { ElMessage } from element-plus// …...

编程日记 2025/8/21 14:24:58

背景

使用Rocket推送消息流

总结

相关文章：