search

BruceChen7 / gitblog

My blog
6 stars 1 forks source link

tokio使用中的注意事项 #53

Open BruceChen7 opened 1 year ago

BruceChen7 commented 1 year ago

参考资料

(usage:: 什么是 tokio runtime )

tokio::runtime::Builder

tokio::task::LocalSet

什么是阻塞

如果我想 block，该怎么办

(usage:: 在同步代码中使用异步函数)

尝试一

impl Sequencer for PlainSequencer {
    fn generate(&self) -> Vec<i32> {
        RUNTIME.block_on(async{
            self.generate_async().await
        })
    }
}

#[cfg(test)]
mod tests{
    #[tokio::test]
    async fn test_sync_method() {
        let sequencer = PlainSequencer {
            bound: 3
        };
        let vec = sequencer.generate();
        println!("vec: {:?}", vec);
    }
}
impl Sequencer for PlainSequencer {
    fn generate(&self) -> Vec<i32> {
        futures::executor::block_on(async {
            self.generate_async().await
        })
    }
}

尝试三

impl Sequencer for PlainSequencer {
    fn generate(&self) -> Vec<i32> {
        let bound = self.bound;
        futures::executor::block_on(async move {
            RUNTIME.spawn(async move {
                let mut res = vec![];
                for i in 0..bound {
                    res.push(i);
                    tokio::time::sleep(Duration::from_millis(100)).await;
                }
                res
            }).await.unwrap()
        })
    }
}

tokio::main 和 tokio::test 区别

一些结论

常见的功能

// single-threaded executor for your async tasks
#[tokio::main(flavor = "current_thread")]
pub async fn run(self) -> Result<()> {
    Builder::new()
        .filter_level(log::LevelFilter::Off)
    ...
}

scoped_thread_local!

use scoped_thread_local::scoped_thread_local;
scoped_thread_local!(static COUNTER: u32);

fn main() {
    let threads = (0..4).map(|_| {
        std::thread::spawn(|| {
            COUNTER.with(|counter| {
                for i in 0..10 {
                    let value = counter.get().unwrap_or(&0) + 1;
                    counter.set(value);
                    println!("Thread {}: counter = {}", std::thread::current().id(), value);
                }
            });
        })
    }).collect::<Vec<_>>();

    for thread in threads {
        thread.join().unwrap();
    }
}

并发编程（类似 goroutine）

'static 约束

send

概念解释

use tokio::task::yield_now;
use std::rc::Rc;
#[tokio::main]
async fn main() {
    tokio::spawn(async {
        // The scope forces `rc` to drop before `.await`.
        {
            let rc = Rc::new("hello");
            println!("{}", rc);
        }

        // `rc` is no longer used. It is **not** persisted when
        // the task yields to the scheduler
        yield_now().await;
    });
}

锁

channel

为啥需要 channel

use mini_redis::client;
#[tokio::main]
async fn main() {
    // 创建到服务器的连接
    let mut client = client::connect("127.0.0.1:6379").await.unwrap();

    // 生成两个任务，一个用于获取 key, 一个用于设置 key
    let t1 = tokio::spawn(async {
        let res = client.get("hello").await;
    });

    let t2 = tokio::spawn(async {
        client.set("foo", "bar".into()).await;
    });

    t1.await.unwrap();
    t2.await.unwrap();
}

channel 的种类

mpsc 的使用

use tokio::sync::mpsc;
#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(32);
    let tx2 = tx.clone();

    tokio::spawn(async move {
        tx.send("sending from first handle").await;
    });

    tokio::spawn(async move {
        tx2.send("sending from second handle").await;
    });

    while let Some(message) = rx.recv().await {
        println!("GOT = {}", message);
    }
}

select

<pattern> = <async expression> => <handler>,

取消

错误

返回值

async fn computation1() -> String {
    // 计算 1
}

async fn computation2() -> String {
    // 计算 2
}

#[tokio::main]
async fn main() {
    let out = tokio::select! {
        res1 = computation1() => res1,
        res2 = computation2() => res2,
    };

    println!("Got = {}", out);
}

模式匹配 (Pattern matching)

借用

循环

恢复异步操作 (Resuming an async operation)

async fn action() {
    // 一些异步逻辑
}

#[tokio::main]
async fn main() {
    let (mut tx, mut rx) = tokio::sync::mpsc::channel(128);

    let operation = action();
    tokio::pin!(operation);

    loop {
        tokio::select! {
            _ = &mut operation => break,
            Some(v) = rx.recv() => {
                if v % 2 == 0 {
                    break;
                }
            }
        }
    }
}

每个任务的并发

type/rust #type/networking #public