本文讲解在爬虫和流处理场景下，如何利用 Redis Pipeline 实现批量游标持久化，大幅减少网络往返，提升系统吞吐。
在数据流处理系统中，游标(cursor) 是标记处理进度的重要机制。每次处理完一批数据后，需要保存当前位置，以便故障恢复时从断点继续。
传统做法是每处理一条数据就更新一次游标：
rust// 传统方式：每次单独写入
for (platform, cursor) in cursors {
redis::cmd("SET").arg(&key).arg(&cursor).query(&mut conn)?;
}
```
问题在于：**每次 SET 操作都是一次完整的网络往返**。假设 Redis 延迟 1ms，处理 10000 个游标就需要 10 秒！
---
## Pipeline 原理
Redis Pipeline 允许将多条命令打包发送，一次性获取所有响应：
```
传统模式:
Client -> SET key1 -> Server -> OK -> Client (1 RTT)
Client -> SET key2 -> Server -> OK -> Client (1 RTT)
... 总计 N 次 RTT
Pipeline 模式:
Client -> SET key1, SET key2, SET key3... -> Server
Server -> OK, OK, OK... -> Client
... 总计 1 次 RTT
```
**性能对比**：
| 操作数量 | 传统方式 (RTT=1ms) | Pipeline |
|----------|-------------------|----------|
| 100 | 100ms | ~2ms |
| 1000 | 1000ms | ~3ms |
| 10000 | 10000ms | ~5ms |
---
## 实战：游标批处理持久化
以下代码来自实际的 Solana 数据流处理项目。
### 批量缓存策略
首先在内存中缓存游标，达到阈值后批量提交：
```rust
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Mutex;
use std::collections::HashMap;
const CURSOR_BATCH_SIZE: usize = 10000; // 达到 10000 个时批量提交
const CURSOR_FLUSH_INTERVAL_MS: u64 = 3600000; // 每小时强制刷新
lazy_static! {
static ref LATEST_CURSORS: Mutex<HashMap<String, String>> = Mutex::new(HashMap::new());
static ref CURSOR_COUNTER: AtomicU64 = AtomicU64::new(0);
}
fn enqueue_cursor(platform: &str, cursor: String) {
let mut cursors = LATEST_CURSORS.lock().unwrap();
cursors.insert(platform.to_string(), cursor);
let count = CURSOR_COUNTER.fetch_add(1, Ordering::SeqCst);
if count % 10 == 0 {
debug!("已缓存 {} 个游标", count);
}
}
```
### Pipeline 批量持久化
```rust
use redis::Commands;
use std::time::Instant;
fn flush_cursors(force: bool) -> Result<usize, anyhow::Error> {
let flush_start = Instant::now();
let mut cursors = LATEST_CURSORS.lock().unwrap();
// 不满足条件则跳过
if cursors.is_empty() || (!force && cursors.len() < CURSOR_BATCH_SIZE) {
return Ok(0);
}
let mut conn = get_conn();
// 创建 Pipeline
let mut pipeline = redis::pipe();
pipeline.atomic(); // 标记为原子操作
let count = cursors.len();
for (platform, cursor) in cursors.iter() {
let key = format!("{}_last_cursor", platform);
pipeline.set(&key, cursor); // 添加命令到 Pipeline
}
// 一次网络往返执行所有命令
match pipeline.query::<()>(&mut conn) {
Ok(_) => {
debug!("批量持久化 {} 个游标完成，耗时: {:.0}ms",
count, flush_start.elapsed().as_millis() as f64);
cursors.clear();
Ok(count)
},
Err(e) => {
error!("批量持久化游标失败: {:?}", e);
Err(anyhow::anyhow!("Redis批量操作失败: {}", e))
}
}
}
```
### 定时刷新任务
结合 Tokio 异步运行时，实现定时刷新：
```rust
use tokio::sync::mpsc::Receiver;
use tokio::time::interval;
use std::time::Duration;
async fn start_cursor_flush_task(mut shutdown_rx: Receiver<()>) {
let mut interval = interval(Duration::from_millis(CURSOR_FLUSH_INTERVAL_MS));
loop {
tokio::select! {
_ = interval.tick() => {
match flush_cursors(true) {
Ok(count) if count > 0 => {
debug!("定时刷新了 {} 个游标", count);
},
Err(e) => error!("定时刷新游标失败: {:?}", e),
_ => {}
}
}
_ = shutdown_rx.recv() => {
info!("接收到关闭信号，进行最终游标刷新");
let _ = flush_cursors(true); // 退出前确保数据不丢失
break;
}
}
}
}
```
---
## 关键设计要点
### 1. 双重触发机制
- **数量触发**：达到 `CURSOR_BATCH_SIZE` 时提交
- **时间触发**：定时强制刷新，避免数据积压过久
### 2. 原子性保证
```rust
pipeline.atomic(); // 使用 MULTI/EXEC 包裹
```
这确保所有命令作为一个事务执行，要么全部成功，要么全部失败。
### 3. 优雅关闭
```rust
_ = shutdown_rx.recv() => {
let _ = flush_cursors(true); // 最终刷新
break;
}
```
程序退出前执行最后一次刷新，确保内存中的游标不丢失。
---
## 性能优化建议
### Pipeline 大小控制
Pipeline 不是越大越好：
- **过大**：内存占用高，可能阻塞其他操作
- **过小**：网络往返次数多
建议根据业务场景测试确定最佳批量大小：
```rust
// 测试不同批量大小
for batch_size in [100, 500, 1000, 5000, 10000] {
benchmark_flush(batch_size);
}
```
### 连接池复用
```rust
lazy_static! {
static ref REDIS_CLIENT: redis::Client = {
redis::Client::open(redis_endpoint).expect("Failed to create Redis client")
};
}
fn get_conn() -> redis::Connection {
REDIS_CLIENT.get_connection().unwrap()
}
```
避免每次操作都创建新连接。
---
## 应用场景
1. **流式数据处理**：Kafka consumer offset、Substreams cursor
2. **爬虫进度追踪**：记录爬取位置，支持断点续爬
3. **分布式任务调度**：Worker 心跳和进度汇报
4. **实时数据同步**：CDC 场景的 binlog 位置记录
---
## 完整示例
```rust
use redis::Commands;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
pub struct CursorManager {
cursors: Mutex<HashMap<String, String>>,
batch_size: usize,
redis_client: redis::Client,
}
impl CursorManager {
pub fn new(redis_url: &str, batch_size: usize) -> Self {
Self {
cursors: Mutex::new(HashMap::new()),
batch_size,
redis_client: redis::Client::open(redis_url).unwrap(),
}
}
pub fn update(&self, key: &str, cursor: String) {
let mut cursors = self.cursors.lock().unwrap();
cursors.insert(key.to_string(), cursor);
}
pub fn flush(&self) -> Result<usize, redis::RedisError> {
let mut cursors = self.cursors.lock().unwrap();
if cursors.len() < self.batch_size {
return Ok(0);
}
let mut conn = self.redis_client.get_connection()?;
let mut pipe = redis::pipe();
pipe.atomic();
let count = cursors.len();
for (key, cursor) in cursors.iter() {
pipe.set(key, cursor);
}
pipe.query(&mut conn)?;
cursors.clear();
Ok(count)
}
}
```
---
## 总结
Redis Pipeline 是提升批量操作性能的利器：
| 优化点 | 效果 |
|--------|------|
| 减少网络往返 | 延迟降低 10-100 倍 |
| 批量提交 | 提高吞吐量 |
| 原子操作 | 保证数据一致性 |
| 定时刷新 | 防止数据积压 |
在流处理、爬虫等需要频繁更新状态的场景中，Pipeline + 批量缓存是标准的高性能实践。
---
## 参考资料
- [Redis Pipeline 文档](https://redis.io/docs/manual/pipelining/)
- [redis-rs 库文档](https://docs.rs/redis/)
- [Tokio 异步运行时](https://tokio.rs/)
本文作者:yowayimono
本文链接:
版权声明:本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议。转载请注明出处！