代码之家 › 专栏 › 技术社区 › Mr.Wang from Next Door

如何在actix web中执行异步函数?

actix-web rust-actix rust

1

Mr.Wang from Next Door · 技术社区 · 5 年前

有一个 async connect() 功能如下。

use actix_web::client::Client;
use futures::compat::Future01CompatExt;
use futures::future::{FutureExt, TryFutureExt};

pub async fn connect() {
    let request = Client::default().get("https://www.google.com").send();

    // compat() converts a v0.1 futures::future::Future<Item = T, Error = E>
    // into a std::future::Future<Output = Result<T, E>>.
    let result = request.compat().await;

    if let Err(e) = result {
        println!("{:?}", e);
        return;
    }
    if let Ok(response) = result {
        println!("{:?}", response);
        return;
    }
}

首先我试着执行 连接() 通过 futures::executor::ThreadPool::run 。它会惊慌失措因为 actix_web::client::Client 必须在actix框架上下文中使用。

如何在actix上下文中执行connect()?

我也试过下面的代码。它不编译。

let future03 = async {
    connect().await;
};
let future01 = future03.unit_error().boxed().compat();
actix::Arbiter::spawn(future01);

它以编译错误结束。

error[E0277]: `(dyn futures::future::Future<Item = awc::response::ClientResponse, Error = actix_http::client::error::SendRequestError> + 'static)` cannot be sent between threads safely
  --> src/main.rs:29:42
   |
29 |     let future01 = future03.unit_error().boxed().compat();
   |                                          ^^^^^ `(dyn futures::future::Future<Item = awc::response::ClientResponse, Error = actix_http::client::error::SendRequestError> + 'static)` cannot be sent between threads safely
   |
   = help: the trait `std::marker::Send` is not implemented for `(dyn futures::future::Future<Item = awc::response::ClientResponse, Error = actix_http::client::error::SendRequestError> + 'static)`
   = note: required because of the requirements on the impl of `std::marker::Send` for `std::ptr::Unique<(dyn futures::future::Future<Item = awc::response::ClientResponse, Error = actix_http::client::error::SendRequestError> + 'static)>`
   = note: required because it appears within the type `std::boxed::Box<(dyn futures::future::Future<Item = awc::response::ClientResponse, Error = actix_http::client::error::SendRequestError> + 'static)>`
   = note: required because it appears within the type `futures::future::map::Map<std::boxed::Box<(dyn futures::future::Future<Item = awc::response::ClientResponse, Error = actix_http::client::error::SendRequestError> + 'static)>, [closure@DefId(150:269 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[1]) 0:bool]>`
   = note: required because it appears within the type `futures::future::either::Either<futures::future::map_err::MapErr<tokio_timer::timeout::Timeout<futures::future::map::Map<std::boxed::Box<(dyn futures::future::Future<Item = awc::response::ClientResponse, Error = actix_http::client::error::SendRequestError> + 'static)>, [closure@DefId(150:269 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[1]) 0:bool]>>, [closure@DefId(150:272 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[3])]>, futures::future::map::Map<std::boxed::Box<(dyn futures::future::Future<Item = awc::response::ClientResponse, Error = actix_http::client::error::SendRequestError> + 'static)>, [closure@DefId(150:269 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[1]) 0:bool]>>`
   = note: required because it appears within the type `futures::future::either::Either<futures::future::result_::FutureResult<awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>, actix_http::client::error::SendRequestError>, futures::future::either::Either<futures::future::map_err::MapErr<tokio_timer::timeout::Timeout<futures::future::map::Map<std::boxed::Box<(dyn futures::future::Future<Item = awc::response::ClientResponse, Error = actix_http::client::error::SendRequestError> + 'static)>, [closure@DefId(150:269 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[1]) 0:bool]>>, [closure@DefId(150:272 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[3])]>, futures::future::map::Map<std::boxed::Box<(dyn futures::future::Future<Item = awc::response::ClientResponse, Error = actix_http::client::error::SendRequestError> + 'static)>, [closure@DefId(150:269 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[1]) 0:bool]>>>`
   = note: required because it appears within the type `impl futures::future::Future`
   = note: required because it appears within the type `impl futures::future::Future`
   = note: required because it appears within the type `{impl futures::future::Future, futures_util::compat::compat01as03::Compat01As03<impl futures::future::Future>, ()}`
   = note: required because it appears within the type `[static generator@src/main.rs:5:24: 20:2 {impl futures::future::Future, futures_util::compat::compat01as03::Compat01As03<impl futures::future::Future>, ()}]`
   = note: required because it appears within the type `std::future::GenFuture<[static generator@src/main.rs:5:24: 20:2 {impl futures::future::Future, futures_util::compat::compat01as03::Compat01As03<impl futures::future::Future>, ()}]>`
   = note: required because it appears within the type `impl core::future::future::Future`
   = note: required because it appears within the type `impl core::future::future::Future`
   = note: required because it appears within the type `{impl core::future::future::Future, ()}`
   = note: required because it appears within the type `[static generator@src/main.rs:26:26: 28:6 {impl core::future::future::Future, ()}]`
   = note: required because it appears within the type `std::future::GenFuture<[static generator@src/main.rs:26:26: 28:6 {impl core::future::future::Future, ()}]>`
   = note: required because it appears within the type `impl core::future::future::Future`
   = note: required because it appears within the type `futures_util::future::unit_error::UnitError<impl core::future::future::Future>`

error[E0277]: `std::rc::Rc<std::cell::RefCell<actix_http::h1::payload::Inner>>` cannot be sent between threads safely
  --> src/main.rs:29:42
   |
29 |     let future01 = future03.unit_error().boxed().compat();
   |                                          ^^^^^ `std::rc::Rc<std::cell::RefCell<actix_http::h1::payload::Inner>>` cannot be sent between threads safely
   |
   = help: within `futures_util::future::unit_error::UnitError<impl core::future::future::Future>`, the trait `std::marker::Send` is not implemented for `std::rc::Rc<std::cell::RefCell<actix_http::h1::payload::Inner>>`
   = note: required because it appears within the type `actix_http::h1::payload::Payload`
   = note: required because it appears within the type `actix_http::payload::Payload<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>`
   = note: required because it appears within the type `awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>`
   = note: required because it appears within the type `std::result::Result<awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>, actix_http::client::error::SendRequestError>`
   = note: required because it appears within the type `std::option::Option<std::result::Result<awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>, actix_http::client::error::SendRequestError>>`
   = note: required because it appears within the type `futures::future::result_::FutureResult<awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>, actix_http::client::error::SendRequestError>`
   = note: required because it appears within the type `futures::future::either::Either<futures::future::result_::FutureResult<awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>, actix_http::client::error::SendRequestError>, futures::future::either::Either<futures::future::map_err::MapErr<tokio_timer::timeout::Timeout<futures::future::map::Map<std::boxed::Box<(dyn futures::future::Future<Item = awc::response::ClientResponse, Error = actix_http::client::error::SendRequestError> + 'static)>, [closure@DefId(150:269 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[1]) 0:bool]>>, [closure@DefId(150:272 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[3])]>, futures::future::map::Map<std::boxed::Box<(dyn futures::future::Future<Item = awc::response::ClientResponse, Error = actix_http::client::error::SendRequestError> + 'static)>, [closure@DefId(150:269 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[1]) 0:bool]>>>`
   = note: required because it appears within the type `impl futures::future::Future`
   = note: required because it appears within the type `impl futures::future::Future`
   = note: required because it appears within the type `{impl futures::future::Future, futures_util::compat::compat01as03::Compat01As03<impl futures::future::Future>, ()}`
   = note: required because it appears within the type `[static generator@src/main.rs:5:24: 20:2 {impl futures::future::Future, futures_util::compat::compat01as03::Compat01As03<impl futures::future::Future>, ()}]`
   = note: required because it appears within the type `std::future::GenFuture<[static generator@src/main.rs:5:24: 20:2 {impl futures::future::Future, futures_util::compat::compat01as03::Compat01As03<impl futures::future::Future>, ()}]>`
   = note: required because it appears within the type `impl core::future::future::Future`
   = note: required because it appears within the type `impl core::future::future::Future`
   = note: required because it appears within the type `{impl core::future::future::Future, ()}`
   = note: required because it appears within the type `[static generator@src/main.rs:26:26: 28:6 {impl core::future::future::Future, ()}]`
   = note: required because it appears within the type `std::future::GenFuture<[static generator@src/main.rs:26:26: 28:6 {impl core::future::future::Future, ()}]>`
   = note: required because it appears within the type `impl core::future::future::Future`
   = note: required because it appears within the type `futures_util::future::unit_error::UnitError<impl core::future::future::Future>`

error[E0277]: `(dyn actix_http::error::ResponseError + 'static)` cannot be sent between threads safely
  --> src/main.rs:29:42
   |
29 |     let future01 = future03.unit_error().boxed().compat();
   |                                          ^^^^^ `(dyn actix_http::error::ResponseError + 'static)` cannot be sent between threads safely
   |
   = help: the trait `std::marker::Send` is not implemented for `(dyn actix_http::error::ResponseError + 'static)`
   = note: required because of the requirements on the impl of `std::marker::Send` for `std::ptr::Unique<(dyn actix_http::error::ResponseError + 'static)>`
   = note: required because it appears within the type `std::boxed::Box<(dyn actix_http::error::ResponseError + 'static)>`
   = note: required because it appears within the type `actix_http::error::Error`
   = note: required because it appears within the type `actix_http::client::error::SendRequestError`
   = note: required because it appears within the type `std::result::Result<awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>, actix_http::client::error::SendRequestError>`
   = note: required because it appears within the type `std::option::Option<std::result::Result<awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>, actix_http::client::error::SendRequestError>>`
   = note: required because it appears within the type `futures::future::result_::FutureResult<awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>, actix_http::client::error::SendRequestError>`
   = note: required because it appears within the type `futures::future::either::Either<futures::future::result_::FutureResult<awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>, actix_http::client::error::SendRequestError>, futures::future::either::Either<futures::future::map_err::MapErr<tokio_timer::timeout::Timeout<futures::future::map::Map<std::boxed::Box<(dyn futures::future::Future<Item = awc::response::ClientResponse, Error = actix_http::client::error::SendRequestError> + 'static)>, [closure@DefId(150:269 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[1]) 0:bool]>>, [closure@DefId(150:272 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[3])]>, futures::future::map::Map<std::boxed::Box<(dyn futures::future::Future<Item = awc::response::ClientResponse, Error = actix_http::client::error::SendRequestError> + 'static)>, [closure@DefId(150:269 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[1]) 0:bool]>>>`
   = note: required because it appears within the type `impl futures::future::Future`
   = note: required because it appears within the type `impl futures::future::Future`
   = note: required because it appears within the type `{impl futures::future::Future, futures_util::compat::compat01as03::Compat01As03<impl futures::future::Future>, ()}`
   = note: required because it appears within the type `[static generator@src/main.rs:5:24: 20:2 {impl futures::future::Future, futures_util::compat::compat01as03::Compat01As03<impl futures::future::Future>, ()}]`
   = note: required because it appears within the type `std::future::GenFuture<[static generator@src/main.rs:5:24: 20:2 {impl futures::future::Future, futures_util::compat::compat01as03::Compat01As03<impl futures::future::Future>, ()}]>`
   = note: required because it appears within the type `impl core::future::future::Future`
   = note: required because it appears within the type `impl core::future::future::Future`
   = note: required because it appears within the type `{impl core::future::future::Future, ()}`
   = note: required because it appears within the type `[static generator@src/main.rs:26:26: 28:6 {impl core::future::future::Future, ()}]`
   = note: required because it appears within the type `std::future::GenFuture<[static generator@src/main.rs:26:26: 28:6 {impl core::future::future::Future, ()}]>`
   = note: required because it appears within the type `impl core::future::future::Future`
   = note: required because it appears within the type `futures_util::future::unit_error::UnitError<impl core::future::future::Future>`

error[E0277]: `(dyn futures::stream::Stream<Error = actix_http::error::PayloadError, Item = bytes::bytes::Bytes> + 'static)` cannot be sent between threads safely
  --> src/main.rs:29:42
   |
29 |     let future01 = future03.unit_error().boxed().compat();
   |                                          ^^^^^ `(dyn futures::stream::Stream<Error = actix_http::error::PayloadError, Item = bytes::bytes::Bytes> + 'static)` cannot be sent between threads safely
   |
   = help: the trait `std::marker::Send` is not implemented for `(dyn futures::stream::Stream<Error = actix_http::error::PayloadError, Item = bytes::bytes::Bytes> + 'static)`
   = note: required because of the requirements on the impl of `std::marker::Send` for `std::ptr::Unique<(dyn futures::stream::Stream<Error = actix_http::error::PayloadError, Item = bytes::bytes::Bytes> + 'static)>`
   = note: required because it appears within the type `std::boxed::Box<(dyn futures::stream::Stream<Error = actix_http::error::PayloadError, Item = bytes::bytes::Bytes> + 'static)>`
   = note: required because it appears within the type `actix_http::payload::Payload`
   = note: required because it appears within the type `actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>`
   = note: required because it appears within the type `actix_http::payload::Payload<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>`
   = note: required because it appears within the type `awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>`
   = note: required because it appears within the type `std::result::Result<awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>, actix_http::client::error::SendRequestError>`
   = note: required because it appears within the type `std::option::Option<std::result::Result<awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>, actix_http::client::error::SendRequestError>>`
   = note: required because it appears within the type `futures::future::result_::FutureResult<awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>, actix_http::client::error::SendRequestError>`
   = note: required because it appears within the type `futures::future::either::Either<futures::future::result_::FutureResult<awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>, actix_http::client::error::SendRequestError>, futures::future::either::Either<futures::future::map_err::MapErr<tokio_timer::timeout::Timeout<futures::future::map::Map<std::boxed::Box<(dyn futures::future::Future<Item = awc::response::ClientResponse, Error = actix_http::client::error::SendRequestError> + 'static)>, [closure@DefId(150:269 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[1]) 0:bool]>>, [closure@DefId(150:272 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[3])]>, futures::future::map::Map<std::boxed::Box<(dyn futures::future::Future<Item = awc::response::ClientResponse, Error = actix_http::client::error::SendRequestError> + 'static)>, [closure@DefId(150:269 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[1]) 0:bool]>>>`
   = note: required because it appears within the type `impl futures::future::Future`
   = note: required because it appears within the type `impl futures::future::Future`
   = note: required because it appears within the type `{impl futures::future::Future, futures_util::compat::compat01as03::Compat01As03<impl futures::future::Future>, ()}`
   = note: required because it appears within the type `[static generator@src/main.rs:5:24: 20:2 {impl futures::future::Future, futures_util::compat::compat01as03::Compat01As03<impl futures::future::Future>, ()}]`
   = note: required because it appears within the type `std::future::GenFuture<[static generator@src/main.rs:5:24: 20:2 {impl futures::future::Future, futures_util::compat::compat01as03::Compat01As03<impl futures::future::Future>, ()}]>`
   = note: required because it appears within the type `impl core::future::future::Future`
   = note: required because it appears within the type `impl core::future::future::Future`
   = note: required because it appears within the type `{impl core::future::future::Future, ()}`
   = note: required because it appears within the type `[static generator@src/main.rs:26:26: 28:6 {impl core::future::future::Future, ()}]`
   = note: required because it appears within the type `std::future::GenFuture<[static generator@src/main.rs:26:26: 28:6 {impl core::future::future::Future, ()}]>`
   = note: required because it appears within the type `impl core::future::future::Future`
   = note: required because it appears within the type `futures_util::future::unit_error::UnitError<impl core::future::future::Future>`

error[E0277]: `(dyn std::any::Any + 'static)` cannot be sent between threads safely
  --> src/main.rs:29:42
   |
29 |     let future01 = future03.unit_error().boxed().compat();
   |                                          ^^^^^ `(dyn std::any::Any + 'static)` cannot be sent between threads safely
   |
   = help: the trait `std::marker::Send` is not implemented for `(dyn std::any::Any + 'static)`
   = note: required because of the requirements on the impl of `std::marker::Send` for `std::ptr::Unique<(dyn std::any::Any + 'static)>`
   = note: required because it appears within the type `std::boxed::Box<(dyn std::any::Any + 'static)>`
   = note: required because it appears within the type `(std::any::TypeId, std::boxed::Box<(dyn std::any::Any + 'static)>)`
   = note: required because of the requirements on the impl of `std::marker::Send` for `hashbrown::raw::RawTable<(std::any::TypeId, std::boxed::Box<(dyn std::any::Any + 'static)>)>`
   = note: required because it appears within the type `hashbrown::map::HashMap<std::any::TypeId, std::boxed::Box<(dyn std::any::Any + 'static)>>`
   = note: required because it appears within the type `actix_http::extensions::Extensions`
   = note: required because of the requirements on the impl of `std::marker::Send` for `std::cell::RefCell<actix_http::extensions::Extensions>`
   = note: required because it appears within the type `actix_http::message::ResponseHead`
   = note: required because it appears within the type `awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>`
   = note: required because it appears within the type `std::result::Result<awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>, actix_http::client::error::SendRequestError>`
   = note: required because it appears within the type `std::option::Option<std::result::Result<awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>, actix_http::client::error::SendRequestError>>`
   = note: required because it appears within the type `futures::future::result_::FutureResult<awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>, actix_http::client::error::SendRequestError>`
   = note: required because it appears within the type `futures::future::either::Either<futures::future::result_::FutureResult<awc::response::ClientResponse<actix_http::encoding::decoder::Decoder<actix_http::payload::Payload>>, actix_http::client::error::SendRequestError>, futures::future::either::Either<futures::future::map_err::MapErr<tokio_timer::timeout::Timeout<futures::future::map::Map<std::boxed::Box<(dyn futures::future::Future<Item = awc::response::ClientResponse, Error = actix_http::client::error::SendRequestError> + 'static)>, [closure@DefId(150:269 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[1]) 0:bool]>>, [closure@DefId(150:272 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[3])]>, futures::future::map::Map<std::boxed::Box<(dyn futures::future::Future<Item = awc::response::ClientResponse, Error = actix_http::client::error::SendRequestError> + 'static)>, [closure@DefId(150:269 ~ awc[6b54]::request[0]::{{impl}}[0]::send_body[0]::{{closure}}[1]) 0:bool]>>>`
   = note: required because it appears within the type `impl futures::future::Future`
   = note: required because it appears within the type `impl futures::future::Future`
   = note: required because it appears within the type `{impl futures::future::Future, futures_util::compat::compat01as03::Compat01As03<impl futures::future::Future>, ()}`
   = note: required because it appears within the type `[static generator@src/main.rs:5:24: 20:2 {impl futures::future::Future, futures_util::compat::compat01as03::Compat01As03<impl futures::future::Future>, ()}]`
   = note: required because it appears within the type `std::future::GenFuture<[static generator@src/main.rs:5:24: 20:2 {impl futures::future::Future, futures_util::compat::compat01as03::Compat01As03<impl futures::future::Future>, ()}]>`
   = note: required because it appears within the type `impl core::future::future::Future`
   = note: required because it appears within the type `impl core::future::future::Future`
   = note: required because it appears within the type `{impl core::future::future::Future, ()}`
   = note: required because it appears within the type `[static generator@src/main.rs:26:26: 28:6 {impl core::future::future::Future, ()}]`
   = note: required because it appears within the type `std::future::GenFuture<[static generator@src/main.rs:26:26: 28:6 {impl core::future::future::Future, ()}]>`
   = note: required because it appears within the type `impl core::future::future::Future`
   = note: required because it appears within the type `futures_util::future::unit_error::UnitError<impl core::future::future::Future>`

error: aborting due to 5 previous errors

我还在学生锈。有人能帮忙吗?很感激。

[dependencies]
actix = "0.8.3"
actix-web = "1.0.5"

# https://rust-lang-nursery.github.io/futures-rs/blog/2019/04/18/compatibility-layer.html
# Rustâs futures ecosystem is currently split in two: 
# On the one hand we have the vibrant ecosystem built around futures@0.1 with its many libraries working on stable Rust 
# and on the other hand thereâs std::future ecosystem with support for the ergonomic and powerful async/await language feature. 
# To bridge the gap between these two worlds we have introduced a compatibility layer as part of the futures@0.3 extension to std::future. 
[dependencies.futures-preview]
version = "0.3.0-alpha.18"
default-features = false
features = ["compat", "async-await", "nightly"]

0 回复 | 直到 5 年前

1

0

Mr.Wang from Next Door 5 年前

let future01 = future03.unit_error().boxed_local().compat();

boxed_local() 把未来包装在一个盒子里,钉住它。类似 boxed() ,但没有发送要求。