120ms 到 30ms：从 Python 到 Rust

我们喜欢看到性能数据。这是我们的核心目标。我们很高兴看到我们持续努力的另一个里程碑：数据管道的写入延迟减少了 4 倍，从 120 毫秒降至 30 毫秒！
这一改进是从通过 Python 应用程序访问的 C 库过渡到完全基于 Rust 的实现的结果。

这是对我们的架构变化、实际结果以及对系统性能和用户体验的影响的简单介绍。

从 Python 切换到 Rust
那么，我们为什么要从 Python 切换到 Rust？我们的数据管道被所有服务使用！

我们的数据管道是我们实时通信平台的支柱。我们的团队负责将事件数据从所有 API 复制到所有内部系统和服务。数据处理、事件存储和索引、连接状态等等。我们的主要目标是确保实时通信的准确性和可靠性。

在迁移之前，旧管道使用通过 Python 服务访问的 C 库，该库缓冲和捆绑数据。这确实是导致我们延迟的关键因素。我们希望进行优化，并且知道这是可以实现的。

我们探索了向 Rust 的过渡，因为我们之前已经看到性能、内存安全性和并发能力对我们有益。是时候再次这样做了！

高度重视 Rust 的性能和异步 IO 优势
Rust 在性能密集型环境中表现出色，尤其是与 Tokio 等异步 IO 库结合使用时。Tokio 支持使用 Rust 编程语言编写异步应用程序的多线程、非阻塞运行时。迁移到 Rust 使我们能够充分利用这些功能，实现高吞吐量和低延迟。所有这些都具有编译时内存和并发安全性。

内存和并发安全
Rust 的所有权模型为内存和并发安全提供了编译时保证，从而避免了最常见的问题，例如数据竞争、内存泄漏和无效内存访问。这对我们来说很有利。

展望未来，我们可以自信地管理代码库的生命周期。如果以后需要，可以进行无情的重构。而且总会有“以后需要”的情况。

使用 MPSC 和 Tokio 进行架构变更、服务到服务以及消息传递的技术实现
以前的架构依赖于服务到服务的消息传递系统，这会带来相当大的开销和延迟。Python 服务使用 C 库来缓冲和捆绑数据。当在多个服务之间交换消息时，会发生延迟，从而增加系统的复杂性。C 库中的缓冲机制是一个很大的瓶颈，导致端到端延迟大约为 120 毫秒。我们认为这是最佳的，因为我们每个事件的平均延迟为 40 微秒。虽然从旧的 Python 服务角度来看这看起来不错，但下游系统在解绑期间受到了影响。这导致总体延迟更高。

当我们部署时，平均每个事件的延迟从原来的 40 微秒增加到 100 微秒。这似乎不是最佳的。

不过，当我们回过头来看原因时，我们可以看到这是怎么回事了。
好消息是，现在下游服务可以更快地逐个使用事件，而无需解绑。

整体端到端延迟有机会从 120 毫秒显着改善到 30 毫秒。

• 新的 Rust 应用程序可以立即并发触发事件。

这种方法在 Python 中是不可能的，因为使用不同的并发模型也需要重写。我们可能可以用 Python 重写。如果要重写，不妨用 Rust 进行最好的重写！

资源减少 CPU 和内存：
我们的 Python 服务会消耗 60% 以上的内核资源。而新的 Rust 服务在多个内核上消耗的资源不到 5%。内存减少也非常显著，Rust 运行时占用的内存约为 200MB，而 Python 则需要占用数 GB 的内存。

基于 Rust 的新架构：

• 新架构利用了 Rust 强大的并发机制和异步 IO 功能。
• 服务到服务的消息传递被利用多生产者、单消费者 (MPSC) 通道的多个实例所取代。
• Tokio 专为高效的异步操作而构建，可减少阻塞并提高吞吐量。
• 我们的数据流程通过消除对中间缓冲阶段的需求而得到简化，转而选择并发和并行。
•  

Rust 应用程序示例
该代码并非直接复制，它只是一个替代示例，用于模拟我们的生产代码的功能。此外，该代码仅显示一个 MPSC，而我们的生产系统使用多个通道。

• Cargo.toml：我们需要包含 Tokio 和我们可能使用的任何其他板条箱的依赖项（例如事件的异步通道）。
• 事件定义：事件类型在代码中使用但未定义，因为我们有许多未在此示例中显示的类型。
• 事件流：event_stream 被引用，但创建方式与许多流不同。取决于您的方法，因此示例保持简单。

以下是带有代码和 Cargo.toml 文件的 Rust 示例。还有事件定义和事件流初始化。

Cargo.toml

[package]
name = "tokio_mpsc_example"
version = "0.1.0"
edition = "2021"

[dependencies]
tokio = { version = "1", features = ["full"] }
main.rs

use tokio::sync::mpsc;
use tokio::task::spawn;
use tokio::time::{sleep, Duration};

// Define the Event type
#[derive(Debug)]
struct Event {
    id: u32,
    data: String,
}

// 处理每个事件的函数
async fn handle_event(event: Event) {
    println!("Processing event: {:?}", event);
    // Simulate processing time
    sleep(Duration::from_millis(200)).await;
}

// 处理接收器接收到的数据的函数
async fn process_data(mut rx: mpsc::Receiver<Event>) {
    while let Some(event) = rx.recv().await {
        handle_event(event).await;
    }
}

#[tokio::main]
async fn main() {
    // 创建缓冲区大小为 100 的通道
    let (tx, rx) = mpsc::channel(100);

    //生成一个任务来处理接收到的数据
    spawn(process_data(rx));

    // 使用虚拟数据模拟事件流以进行演示
    let event_stream = vec![
        Event { id: 1, data: "Event 1".to_string() },
        Event { id: 2, data: "Event 2".to_string() },
        Event { id: 3, data: "Event 3".to_string() },
    ];

    // 通过通道发送事件
    for event in event_stream {
        if tx.send(event).await.is_err() {
            eprintln!("Receiver dropped");
        }
    }
}

Rust 示例文件

1. Cargo.toml：
   • 指定包名称、版本和版次。
   • 包含“完整”功能集所需的 tokio 依赖项。