使用 rustsqlxtokio 时如何取消长时间运行的查询

All期货本质上是可取消的--这是async相对于阻塞多线程代码的好处(和危害)之一.你只是放弃future ，而不是进一步调查它.

您需要做的第一件事是将阻塞通道更改为async通道-这允许在运行查询时混合判断通道.然后，您可以使用各种future 的操作工具来决定是否使用continue来运行查询.我决定用select!来做这件事，它会轮询几个期货，并根据最先完成的那个来运行代码.

(这方面可能有更好的工具；我熟悉异步的工作原理，但还没有编写很多复杂的异步代码.)

use futures::future::OptionFuture;
use std::time::Duration;
use tokio::sync::mpsc;

async fn run_db_thread(mut requests: mpsc::Receiver<Query>) {
    // This variable holds the current query being run, if there is one
    let mut current_query_future = OptionFuture::default();

    loop {
        tokio::select! {
            // If we receive a new query, replace the current query with it.
            Some(query) = requests.recv() => {
                println!("Starting new query {query:?}");
                current_query_future = OptionFuture::from(Some(Box::pin(async move {
                    let answer = do_query(&query).await;
                    println!("Finished query {query:?} => {answer:?}");
                    answer
                })));
                // Note that we did not `.await` the new future.
            }

            // Poll the current query future, and check if it is done yet.
            Some(_answer) = &mut current_query_future => {
                // Stop polling the completed future.
                current_query_future = None.into();
            }

            // We get here if both of the above branches saw None, which means that the
            // channel is closed, *and* there is no query to run.
            else => {
                println!("Channel closed; run_db_thread() exiting");
                break;
            }
        }
    }
}

/// Example to drive the loop.
#[tokio::main]
async fn main() {
    let (sender, receiver) = mpsc::channel(1);
    tokio::spawn(run_db_thread(receiver));

    for (i, delay) in [1000, 1000, 1, 1, 1, 1000, 1000].into_iter().enumerate() {
        sender.send(Query(i)).await.unwrap();
        tokio::time::sleep(Duration::from_millis(delay)).await;
    }
    println!("main() exiting");
}

// Skeleton data types to make the example compile
#[derive(Debug)]
#[allow(dead_code)]
struct Query(usize);
#[derive(Debug)]
struct Answer;
async fn do_query(_q: &Query) -> Answer {
    tokio::time::sleep(Duration::from_millis(100)).await;
    Answer
}

此示例代码打印:

Starting new query Query(0)
Finished query Query(0) => Answer
Starting new query Query(1)
Finished query Query(1) => Answer
Starting new query Query(2)
Starting new query Query(3)
Starting new query Query(4)
Starting new query Query(5)
Finished query Query(5) => Answer
Starting new query Query(6)
Finished query Query(6) => Answer
main() exiting

也就是说，查询0、1、5和6已完成，但查询2、3和4在它们可以完成之前被到达的新查询取消.