rust-async-pattern

异步代码的生命周期怎么这么难管？

async 让生命周期问题更复杂。

---

问题代码

```rust

// ❌ Stream 实现中返回借用内部缓冲区的 slice

pub struct SessionStream<'buf> {

buf: Vec,

cache: Vec>,

}

impl Stream for SessionStream<'buf> {

type Item = Result, Status>;

fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> {

// ❌ 返回的 CachedResponse<'buf> 生命周期依赖于 self.buf

// 但 Stream trait 的 Item 必须能在任意时刻被使用

}

}

```

错误信息

```

error[E0700]: hidden type for impl futures_core::Stream captures lifetime that does not appear in bounds

error[E0310]: the parameter type may not live long enough

```

原因

• Stream 的 Item 可以被任意持有
• 'buf 和 self 绑定在一起
• 返回的 Item 逃逸了 self 的生命周期

解决：Worker + Channel 模式

```rust

// ✅ 内部 worker 持有缓冲区，对外只发 owned snapshot

pub struct SessionWorker {

rx_events: Receiver,

tx_snapshots: Sender,

buf: Vec,

}

impl SessionWorker {

pub async fn run(&mut self) {

while let Some(event) = self.rx_events.recv().await {

let snapshot = self.process_event(event);

self.tx_snapshots.send(snapshot).await;

}

}

fn process_event(&mut self, event: Bytes) -> SnapshotResponse {

// 内部可以借用 self.buf

let start = self.buf.len();

self.buf.extend_from_slice(&event);

// 但对外发的是 owned SnapshotResponse

SnapshotResponse {

id: self.next_id,

payload: Bytes::copy_from_slice(&self.buf[start..]),

}

}

}

// ✅ Stream 只读 channel，发的都是 owned

pub struct SessionStream {

rx_snapshots: Receiver,

}

impl Stream for SessionStream {

type Item = Result;

fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> {

// 这里发的都是 SnapshotResponse (owned)，没问题

}

}

```

---

问题代码

```rust

// ❌ tokio::spawn 要求 'static，但 BorrowedMessage<'a> 不是

pub struct BorrowedMessage<'a> {

pub raw: &'a [u8],

pub meta: MessageMeta,

}

pub trait Plugin: Send + Sync {

fn handle<'a>(&'a self, msg: BorrowedMessage<'a>)

-> Pin> + Send + 'a>>;

}

fn dispatch_to_plugins(msg: BorrowedMessage<'a>) {

for p in &plugins {

let fut = p.handle(msg);

tokio::spawn(fut); // ❌ fut 不是 'static

}

}

```

原因

• tokio::spawn 不知道任务何时完成
• 如果任务持有 'a 引用，'a 可能已过期

解决：事件循环 + Actor 模式

```rust

// ✅ 不 spawn，每个插件是一个长期存在的 actor

struct PluginActor {

plugin: M,

queue: Receiver,

arena: MessageArena,

}

impl PluginActor {

pub async fn run(&mut self) {

while let Some(msg) = self.queue.recv().await {

// 在 arena 域内处理消息

self.arena.with_message(msg, |msg_ref| {

self.plugin.handle(msg_ref);

});

}

}

}

// ✅ 用索引代替直接借用

pub struct MessageRef {

index: usize,

generation: u64,

}

struct MessageArena {

buffers: Vec>,

}

impl MessageArena {

pub fn get(&self, ref: MessageRef) -> Option<&[u8]> {

// 通过索引安全获取

self.buffers.get(ref.index)?.get(ref.generation)

}

}

```

---

约束

1. 零拷贝缓冲区复用
2. 插件热插拔
3. 异步 handler
4. 可重试/延后 ack

最终架构

```

┌─────────────────────────────────────┐

│ Decode Layer │ 持有缓冲区

├─────────────────────────────────────┤

│ MessageArena │ 缓冲区管理

├─────────────────────────────────────┤

│ Event Loop │ 协作式调度

├─────────────────────────────────────┤

│ Plugin Actor │ 每个插件一个

└─────────────────────────────────────┘

│

↓ API 层只看到 owned 数据

┌─────────────────────────────────────┐

│ GraphQL / gRPC │ 'static 要求

└─────────────────────────────────────┘

```

关键设计

```rust

// 1. 缓冲区管理域

struct MessageArena {

buffers: Vec>,

free_list: Vec,

}

impl MessageArena {

// 分配时返回索引，不是引用

fn alloc(&mut self, data: &[u8]) -> MessageRef {

let idx = self.buffers.len();

self.buffers.push(Arc::new(data.to_vec()));

MessageRef { index: idx, generation: 0 }

}

}

// 2. API 层只暴露 owned

pub trait Plugin: Send + Sync {

async fn handle(&self, msg: OwnedMessage); // owned

}

```

---

| 问题 | 原因 | 解决 |

|-----|------|-----|

| Stream 返回借用 | Item 生命周期逃逸 | Worker + Channel |

| tokio::spawn 非 'static | 任务可能持有临时引用 | 事件循环模式 |

| 插件 handler 生命周期 | 插件持有消息 | Actor + 索引 |

| async-graphql + GAT | 'static 要求 | owned DTO |

| tonic Stream 自引用 | 缓冲区复用冲突 | Snapshot 模式 |

---

| 场景 | 方案 |

|-----|-----|

| 独立任务，可并行 | tokio::spawn |

| 需要协作调度 | Event Loop |

| 插件系统 | Actor 模式 |

| 长期运行的状态ful | Actor |

| 短命任务 | spawn |

| 需要背压控制 | Channel + actor |