Rust 2026 经验谈 - 异步 + FFI：桥接同步与异步世界

1
use tokio::task;
2

3
async fn call_blocking_c_lib(input: &[u8]) -> Vec<u8> {
4
    // 将阻塞调用提交到专门的阻塞线程池
5
    let input = input.to_owned();
6
    task::spawn_blocking(move || {
7
        // 这里的代码运行在阻塞线程池中
8
        // 不会占用异步工作线程
9
        expensive_c_computation(&input)
10
    })
11
    .await
12
    .expect("spawn_blocking panicked")
13
}
14

15
fn expensive_c_computation(data: &[u8]) -> Vec<u8> {
16
    // 模拟阻塞的 C 库调用
17
    std::thread::sleep(std::time::Duration::from_millis(100));
18
    data.to_vec()
19
}

1
// 错误！会阻塞异步工作线程，影响同一线程上的其他 task
2
async fn bad_blocking() -> String {
3
    std::thread::sleep(std::time::Duration::from_secs(5)); // 整个线程卡住！
4
    String::from("done")
5
}
6

7
// 正确：用 spawn_blocking
8
async fn good_blocking() -> String {
9
    tokio::task::spawn_blocking(|| {
10
        std::thread::sleep(std::time::Duration::from_secs(5));
11
        String::from("done")
12
    })
13
    .await
14
    .unwrap()
15
}

1
use criterion::{criterion_group, criterion_main, Criterion, BenchmarkId};
2
use tokio::runtime::Runtime;
3

4
fn bench_async_function(c: &mut Criterion) {
5
    let rt = Runtime::new().unwrap();
6

7
    c.bench_function("async_computation", |b| {
8
        b.to_async(&rt).iter(|| async {
9
            some_async_work().await
10
        })
11
    });
12
}
13

14
async fn some_async_work() -> u64 {
15
    tokio::time::sleep(std::time::Duration::from_micros(10)).await;
16
    42
17
}
18

19
criterion_group!(benches, bench_async_function);
20
criterion_main!(benches);

1
fn bench_spawn_vs_block_in_place(c: &mut Criterion) {
2
    let rt = Runtime::new().unwrap();
3

4
    let mut group = c.benchmark_group("blocking_methods");
5

6
    group.bench_function("spawn_blocking", |b| {
7
        b.to_async(&rt).iter(|| async {
8
            tokio::task::spawn_blocking(|| {
9
                std::thread::sleep(std::time::Duration::from_micros(100));
10
            })
11
            .await
12
            .unwrap()
13
        })
14
    });
15

16
    group.bench_function("block_in_place", |b| {
17
        b.to_async(&rt).iter(|| async {
18
            tokio::task::block_in_place(|| {
19
                std::thread::sleep(std::time::Duration::from_micros(100));
20
            })
21
        })
22
    });
23

24
    group.finish();
25
}

1
use std::future::Future;
2
use std::pin::Pin;
3
use std::sync::{Arc, Mutex};
4
use std::task::{Context, Poll, Waker};
5

6
struct CCallbackFuture {
7
    result: Arc<Mutex<Option<i32>>>,
8
    waker: Arc<Mutex<Option<Waker>>>,
9
}
10

11
impl Future for CCallbackFuture {
12
    type Output = i32;
13

14
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
15
        let mut result = self.result.lock().unwrap();
16
        if let Some(val) = result.take() {
17
            Poll::Ready(val)
18
        } else {
19
            // 保存 waker，C 回调触发时唤醒
20
            *self.waker.lock().unwrap() = Some(cx.waker().clone());
21
            Poll::Pending
22
        }
23
    }
24
}
25

26
// C 回调函数
27
unsafe extern "C" fn on_complete(value: i32, user_data: *mut std::ffi::c_void) {
28
    let data = unsafe { &*(user_data as *const CallbackData) };
29
    *data.result.lock().unwrap() = Some(value);
30
    if let Some(waker) = data.waker.lock().unwrap().take() {
31
        waker.wake();
32
    }
33
}
34

35
struct CallbackData {
36
    result: Arc<Mutex<Option<i32>>>,
37
    waker: Arc<Mutex<Option<Waker>>>,
38
}

1
use tokio::io::Interest;
2
use tokio::net::UdpSocket;
3

4
async fn poll_c_fd() -> std::io::Result<()> {
5
    // 将 C 库的 fd 注册到 tokio 的 epoll
6
    // 注意：from_raw_fd 需要 unsafe，且要求 fd 有效且未被 tokio 接管
7
    let socket = unsafe { UdpSocket::from_raw_fd(c_library_fd) };
8

9
    loop {
10
        socket.ready(Interest::READABLE).await?;
11
        // fd 可读，调用 C 库的非阻塞读取
12
        let result = unsafe { c_lib_nonblocking_read() };
13
        if result > 0 {
14
            // 处理数据
15
        }
16
    }
17
}

1
// 危险：在阻塞代码中持有异步 Mutex 的锁
2
async fn bad_pattern() {
3
    let data = Arc::new(tokio::sync::Mutex::new(vec![]));
4
    let data_clone = data.clone();
5

6
    tokio::task::spawn_blocking(move || {
7
        let mut guard = data_clone.blocking_lock(); // 这可以工作...
8
        guard.push(1);
9
        // 但如果另一个 task 在等锁，可能死锁
10
    })
11
    .await
12
    .unwrap();
13
}
14

15
// 更安全：在异步侧获取锁，将数据传入阻塞闭包
16
async fn good_pattern() {
17
    let data = Arc::new(tokio::sync::Mutex::new(vec![]));
18

19
    {
20
        let mut guard = data.lock().await;
21
        let inner = &mut *guard;
22
        tokio::task::spawn_blocking(move || {
23
            // 对 inner 的数据做阻塞处理
24
        })
25
        .await
26
        .unwrap();
27
    }
28
}

1
// Rust 侧
2
#[unsafe(no_mangle)]
3
pub unsafe extern "C" fn rust_async_operation(
4
    input: *const u8,
5
    input_len: usize,
6
    callback: unsafe extern "C" fn(*mut u8, usize),
7
) {
8
    let input_data = unsafe { std::slice::from_raw_parts(input, input_len) }.to_vec();
9

10
    // 在独立运行时中启动异步操作
11
    std::thread::spawn(move || {
12
        let rt = tokio::runtime::Runtime::new().unwrap();
13
        rt.block_on(async {
14
            let result = async_process(&input_data).await;
15
            callback(result.as_ptr() as *mut u8, result.len());
16
        });
17
    });
18
}