rust异步编程

fn a(){
  b()
  //...something else
}
 
fn b(){
  ...
}

read(file, buf);

read(file, buff);
// read函数立即返回
// 不会阻塞当前程序

# 处理一次用户请求需要经过的步骤：
A;
B;
C;
数据库读取;
D；
E；
F；

// 主线程
main_thread() {
    A;
    B;
    C;
    发送数据库查询请求;
    D;
    E;
    F;
}
// 数据库线程
DataBase_thread() {
while(1) {
        处理数据库读取请求;
        返回结果;
    }
}

function callBack(){
  D()
  E()
  F()
}

function dbQuery(request, callBack){
   //query database
   //call callback handler
  callBack()
}

fn get_two_sites() {
    // 创建两个新线程执行任务
    let thread_one = thread::spawn(|| download("https://course.rs"));
    let thread_two = thread::spawn(|| download("https://fancy.rs"));
 
    // 等待两个线程的完成
    thread_one.join().expect("thread one panicked");
    thread_two.join().expect("thread two panicked");
}

async fn get_two_sites_async() {
    // 创建两个不同的`future`，你可以把`future`理解为未来某个时刻会被执行的计划任务
    // 当两个`future`被同时执行后，它们将并发的去下载目标页面
    let future_one = download_async("https://www.foo.com");
    let future_two = download_async("https://www.bar.com");
 
    // 同时运行两个`future`，直至完成
    join!(future_one, future_two);
}

async fn say_hi() {
    println!("gogogo!")
}
fn main() {
    say_hi();
}

use futures::executor::block_on;
 
async fn say_hi() {
    println!("gogogo!")
}
 
fn main() {
    let future = say_hi();
    block_on(future)
}

use futures::executor::block_on;
 
async fn hello_world() {
    hello_cat();
    println!("hello, world!");
}
 
async fn hello_cat() {
    println!("hello, kitty!");
}
fn main() {
    let future = hello_world();
    block_on(future);
}

➜ cargo run
   Compiling exercise v0.1.0 (/Users/chenzilong/workspace/rust-master/exercise)
warning: unused implementer of `futures::Future` that must be used
 --> src/main.rs:4:5
  |
4 |     hello_cat();
  |     ^^^^^^^^^^^
  |
  = note: futures do nothing unless you `.await` or poll them
  = note: `#[warn(unused_must_use)]` on by default
 
warning: `exercise` (bin "exercise") generated 1 warning
    Finished dev [unoptimized + debuginfo] target(s) in 0.11s
     Running `target/debug/exercise`
hello, world!

use futures::executor::block_on;
 
async fn hello_world() {
    hello_cat().await;
    println!("hello, world!");
}
 
async fn hello_cat() {
    println!("hello, kitty!");
}
fn main() {
    let future = hello_world();
    block_on(future);
}

use futures::executor::block_on;
 
struct Song {
    author: String,
    name: String,
}
 
async fn learn_song() -> Song {
    Song {
        author: "周杰伦".to_string(),
        name: String::from("《菊花台》"),
    }
}
 
async fn sing_song(song: Song) {
    println!(
        "给大家献上一首{}的{} ~ {}",
        song.author, song.name, "菊花残，满地伤~ ~"
    );
}
 
async fn dance() {
    println!("唱到情深处，身体不由自主的动了起来~ ~");
}
 
fn main() {
    let song = block_on(learn_song());
    block_on(sing_song(song));
    block_on(dance());
}

use futures::executor::block_on;
 
struct Song {
    author: String,
    name: String,
}
 
async fn learn_song() -> Song {
    Song {
        author: "曲婉婷".to_string(),
        name: String::from("《我的歌声里》"),
    }
}
 
async fn sing_song(song: Song) {
    println!("sing song: {} - {}", song.author, song.name);
}
 
async fn dance() {
    println!("dance");
}
 
async fn learn_and_sing() {
    let song = learn_song().await;
    sing_song(song).await;
}
async fn async_main() {
    let f1 = learn_and_sing();
    let f2 = dance();
 
    // `join!`可以并发的处理和等待多个`Future`，若`learn_and_sing Future`被阻塞，那`dance Future`可以拿过线程的所有权继续执行。若`dance`也变成阻塞状态，那`learn_and_sing`又可以再次拿回线程所有权，继续执行。
    // 若两个都被阻塞，那么`async main`会变成阻塞状态，然后让出线程所有权，并将其交给`main`函数中的`block_on`执行器
    futures::join!(f1, f2);
}
 
fn main() {
    block_on(async_main());
}

struct SelfRef {
    value: String,
    pointer_to_value: *mut String,
}

#[derive(Debug)]
struct Test {
    a: String,
    b: *const String,
}
 
impl Test {
    fn new(a: &str) -> Self {
        Self {
            a: String::from(a),
            b: std::ptr::null(),
        }
    }
 
    fn init(&mut self) {
        let self_ref: *const String = &self.a;
        self.b = self_ref;
    }
 
    fn a(&self) -> &str {
        &self.a
    }
 
    fn b(&self) -> &String {
        assert!(
            !self.b.is_null(),
            "Test::b called without Test::init being called first"
        );
        unsafe { &*self.b }
    }
}

fn main() {
    let mut test1 = Test::new("test1");
    test1.init();
    let mut test2 = Test::new("test2");
    test2.init();
 
    println!("a: {}, b: {}", test1.a(), test1.b());
    println!("a: {}, b: {}", test2.a(), test2.b());
}
// a: test1 b: test1
// a: test2 b: test2

fn main() {
    let mut test1 = Test::new("test1");
    test1.init();
    let mut test2 = Test::new("test2");
    test2.init();
 
    println!("a: {}, b: {}", test1.a(), test1.b());
    std::mem::swap(&mut test1,&mut test2);
    println!("a: {}, b: {}", test2.a(), test2.b());
}
// a: test1 b: test1
// a: test1 b: test2

use std::{marker::PhantomPinned, pin::Pin};
 
#[derive(Debug)]
struct Test {
    a: String,
    b: *const String,
    _marker: PhantomPinned,
}
 
impl Test {
    fn new(a: &str) -> Self {
        Self {
            a: String::from(a),
            b: std::ptr::null(),
            //使用这么标记来让我们的类型自动实现特征!Unpin
            _marker: PhantomPinned,
        }
    }
 
    fn init(self: Pin<&mut Self>) {
        let self_ptr: *const String = &self.a;
        let this = unsafe { self.get_unchecked_mut() };
        this.b = self_ptr;
    }
 
    fn a(&self) -> &str {
        &self.a
    }
 
    fn b(&self) -> &String {
        assert!(
            !self.b.is_null(),
            "Test::b called without Test::init being called first"
        );
        unsafe { &*self.b }
    }
}

fn main() {
    // 此时的`test1`可以被安全的移动
    let mut test1 = Test::new("test1");
    // 新的`test1`由于使用了`Pin`，因此无法再被移动，这里的声明会将之前的`test1`遮蔽掉(shadow)
    let mut test1 = unsafe { Pin::new_unchecked(&mut test1) };
    Test::init(test1.as_mut());
 
    let mut test2 = Test::new("test2");
    let mut test2 = unsafe { Pin::new_unchecked(&mut test2) };
    Test::init(test2.as_mut());
 
    println!("a: {}, b: {}", Test::a(test1.as_ref()), Test::b(test1.as_ref()));
    std::mem::swap(test1.get_mut(), test2.get_mut());
    println!("a: {}, b: {}", Test::a(test2.as_ref()), Test::b(test2.as_ref()));
}

use std::marker::PhantomPinned;
use std::pin::Pin;
 
#[derive(Debug)]
struct Test {
    a: String,
    b: *const String,
    _marker: PhantomPinned,
}
 
impl Test {
    fn new(txt: &str) -> Pin<Box<Self>> {
        let t = Test {
            a: String::from(txt),
            b: std::ptr::null(),
            _marker: PhantomPinned,
        };
        let mut boxed = Box::pin(t);
        let self_ptr: *const String = &boxed.as_ref().a;
        unsafe { boxed.as_mut().get_unchecked_mut().b = self_ptr };
 
        boxed
    }
 
    fn a(self: Pin<&Self>) -> &str {
        &self.get_ref().a
    }
 
    fn b(self: Pin<&Self>) -> &String {
        unsafe { &*(self.b) }
    }
}
 
pub fn main() {
    let test1 = Test::new("test1");
    let test2 = Test::new("test2");
 
    println!("a: {}, b: {}", test1.as_ref().a(), test1.as_ref().b());
    println!("a: {}, b: {}", test2.as_ref().a(), test2.as_ref().b());
}

async fn foo(x: &u8) -> u8 { *x }
 
// 上面的函数跟下面的函数是等价的:
fn foo_expanded<'a>(x: &'a u8) -> impl Future<Output = u8> + 'a {
    async move { *x }
}

use std::future::Future;
fn bad() -> impl Future<Output = u8> {
    let x = 5;
    borrow_x(&x) // ERROR: `x` does not live long enough
}
 
async fn borrow_x(x: &u8) -> u8 { *x }

use std::future::Future;
 
async fn borrow_x(x: &u8) -> u8 { *x }
 
fn good() -> impl Future<Output = u8> {
    async {
        let x = 5;
        borrow_x(&x).await
    }
}

use futures::{executor::block_on, Future};
 
// 多个不同的 `async` 语句块可以访问同一个本地变量，只要它们在该变量的作用域内执行
async fn blocks() {
    let my_string = "foo".to_string();
 
    let future_one = async {
        // ...
        println!("{my_string}");
    };
 
    let future_two = async {
        // ...
        println!("{my_string}");
    };
 
    // 运行两个 Future 直到完成
    let ((), ()) = futures::join!(future_one, future_two);
}
 
// 由于 `async move` 会捕获环境中的变量，因此只有一个 `async move` 语句块可以访问该变量，
// 但是它也有非常明显的好处： 变量可以转移到返回的 Future 中，不再受借用生命周期的限制
fn move_block() -> impl Future<Output = ()> {
    let my_string = "foo".to_string();
    async move {
        // ...
        println!("{my_string}");
    }
}
 
fn main() {
    block_on(move_block());
}

use futures::{channel::mpsc, executor::block_on, SinkExt, StreamExt};
 
async fn send_recv() {
    const BUFFER_SIZE: usize = 10;
    let (mut tx, mut rx) = mpsc::channel::<i32>(BUFFER_SIZE);
 
    tx.send(1).await.unwrap();
    tx.send(2).await.unwrap();
    drop(tx);
 
    // `StreamExt::next` 类似于 `Iterator::next`, 但是前者返回的不是值，而是一个 `Future<Output = Option<T>>`，
    // 因此还需要使用`.await`来获取具体的值
    assert_eq!(Some(1), rx.next().await);
    assert_eq!(Some(2), rx.next().await);
    assert_eq!(None, rx.next().await);
}
 
fn main() {
    block_on(send_recv());
}

use std::{io, pin::Pin};
 
use futures::{Stream, StreamExt, TryStreamExt};
 
async fn sum_with_next(mut stream: Pin<&mut dyn Stream<Item = u32>>) -> u32 {
    let mut sum = 0;
    while let Some(value) = stream.next().await {
        sum += value;
    }
    sum
}
 
async fn sum_with_try_next(
    mut stream: Pin<&mut dyn Stream<Item = Result<i32, io::Error>>>,
) -> Result<i32, io::Error> {
    let mut sum = 0;
    while let Some(value) = stream.try_next().await? {
        sum += value;
    }
    Result::Ok(sum)
}

use futures::stream::{FuturesUnordered, StreamExt};
 
// 异步函数，模拟数据处理
async fn process_data(data: i32) -> i32 {
    // 假设这里有一些异步处理
    data * 2
}
 
#[tokio::main]
async fn main() {
    let data_vec = vec![1, 2, 3, 4, 5];
 
    // 并发处理每个数据项
    let futures: FuturesUnordered<_> = data_vec
        .into_iter()
        .map(|data| process_data(data))
        .collect();
 
    futures
        .for_each_concurrent(None, |result| async move {
            println!("Result: {}", result);
        })
        .await;
}

async fn enjoy_book_and_music() -> (Book, Music) {
    let book = enjoy_book().await;
    let music = enjoy_music().await;
    (book, music)
}

use futures::join;
async fn enjoy_book_and_music() -> (Book, Music) {
    let book = enjoy_book().await;
    let music = enjoy_music().await;
	  join!(book, music)
}

use futures::try_join;
 
async fn get_book() -> Result<Book, String> { /* ... */ Ok(Book) }
async fn get_music() -> Result<Music, String> { /* ... */ Ok(Music) }
 
async fn get_book_and_music() -> Result<(Book, Music), String> {
    let book_fut = get_book();
    let music_fut = get_music();
    try_join!(book_fut, music_fut)
}

use futures::{
    future::TryFutureExt,
    try_join,
};
 
async fn get_book() -> Result<Book, ()> { /* ... */ Ok(Book) }
async fn get_music() -> Result<Music, String> { /* ... */ Ok(Music) }
 
async fn get_book_and_music() -> Result<(Book, Music), String> {
    let book_fut = get_book().map_err(|()| "Unable to get book".to_string());
    let music_fut = get_music();
    try_join!(book_fut, music_fut)
}

use futures::{
    future::FutureExt, // for `.fuse()`
    pin_mut,
    select,
};
 
async fn task_one() { /* ... */ }
async fn task_two() { /* ... */ }
 
async fn race_tasks() {
    let t1 = task_one().fuse();
    let t2 = task_two().fuse();
 
    pin_mut!(t1, t2);
 
    select! {
        () = t1 => println!("任务1率先完成"),
        () = t2 => println!("任务2率先完成"),
    }
}

use futures::future;
use futures::select;
pub fn main() {
    let mut a_fut = future::ready(4);
    let mut b_fut = future::ready(6);
    let mut total = 0;
 
    loop {
        select! {
            a = a_fut => total += a,
            b = b_fut => total += b,
            complete => break,
            default => panic!(), // 该分支永远不会运行，因为 `Future` 会先运行，然后是 `complete`
        };
    }
    assert_eq!(total, 10);
}

use futures::{
    stream::{Stream, StreamExt, FusedStream},
    select,
};
 
async fn add_two_streams(
    mut s1: impl Stream<Item = u8> + FusedStream + Unpin,
    mut s2: impl Stream<Item = u8> + FusedStream + Unpin,
) -> u8 {
    let mut total = 0;
 
    loop {
        let item = select! {
            x = s1.next() => x,
            x = s2.next() => x,
            complete => break,
        };
        if let Some(next_num) = item {
            total += next_num;
        }
    }
 
    total
}

use futures::{
    future::{Fuse, FusedFuture, FutureExt},
    stream::{FusedStream, Stream, StreamExt},
    pin_mut,
    select,
};
 
async fn get_new_num() -> u8 { /* ... */ 5 }
 
async fn run_on_new_num(_: u8) { /* ... */ }
 
async fn run_loop(
    mut interval_timer: impl Stream<Item = ()> + FusedStream + Unpin,
    starting_num: u8,
) {
    let run_on_new_num_fut = run_on_new_num(starting_num).fuse();
    let get_new_num_fut = Fuse::terminated();
    pin_mut!(run_on_new_num_fut, get_new_num_fut);
    loop {
        select! {
            () = interval_timer.select_next_some() => {
                // 定时器已结束，若`get_new_num_fut`没有在运行，就创建一个新的
                if get_new_num_fut.is_terminated() {
                    get_new_num_fut.set(get_new_num().fuse());
                }
            },
            new_num = get_new_num_fut => {
                // 收到新的数字 -- 创建一个新的`run_on_new_num_fut`并丢弃掉旧的
                run_on_new_num_fut.set(run_on_new_num(new_num).fuse());
            },
            // 运行 `run_on_new_num_fut`
            () = run_on_new_num_fut => {},
            // 若所有任务都完成，直接 `panic`， 原因是 `interval_timer` 应该连续不断的产生值，而不是结束
            //后，执行到 `complete` 分支
            complete => panic!("`interval_timer` completed unexpectedly"),
        }
    }
}

use futures::{
    future::{Fuse, FusedFuture, FutureExt},
    stream::{FusedStream, FuturesUnordered, Stream, StreamExt},
    pin_mut,
    select,
};
 
async fn get_new_num() -> u8 { /* ... */ 5 }
 
async fn run_on_new_num(_: u8) -> u8 { /* ... */ 5 }
 
 
// 使用从 `get_new_num` 获取的最新数字 来运行 `run_on_new_num`
//
// 每当计时器结束后，`get_new_num` 就会运行一次，它会立即取消当前正在运行的`run_on_new_num` ,
// 并且使用新返回的值来替换
async fn run_loop(
    mut interval_timer: impl Stream<Item = ()> + FusedStream + Unpin,
    starting_num: u8,
) {
    let mut run_on_new_num_futs = FuturesUnordered::new();
    run_on_new_num_futs.push(run_on_new_num(starting_num));
    let get_new_num_fut = Fuse::terminated();
    pin_mut!(get_new_num_fut);
    loop {
        select! {
            () = interval_timer.select_next_some() => {
                 // 定时器已结束，若 `get_new_num_fut` 没有在运行，就创建一个新的
                if get_new_num_fut.is_terminated() {
                    get_new_num_fut.set(get_new_num().fuse());
                }
            },
            new_num = get_new_num_fut => {
                 // 收到新的数字 -- 创建一个新的 `run_on_new_num_fut` (并没有像之前的例子那样丢弃掉旧值)
                run_on_new_num_futs.push(run_on_new_num(new_num));
            },
            // 运行 `run_on_new_num_futs`, 并检查是否有已经完成的
            res = run_on_new_num_futs.select_next_some() => {
                println!("run_on_new_num_fut returned {:?}", res);
            },
            // 若所有任务都完成，直接 `panic`， 原因是 `interval_timer` 应该连续不断的产生值，而不是结束
            //后，执行到 `complete` 分支
            complete => panic!("`interval_timer` completed unexpectedly"),
        }
    }
}