5

我有多个线程执行一些繁重的操作，我需要在工作中使用客户端。我使用 Hyper v0.11 作为 HTTP 客户端，并且我想重用连接，因此我需要共享相同hyper::Client的连接以保持打开连接（在keep-alive模式下）。

客户端在线程之间不可共享（它不实现Syncor Send）。这是我尝试做的代码的一个小片段：

let mut core = Core::new().expect("Create Client Event Loop");
let handle = core.handle();

let remote = core.remote();

let client = Client::new(&handle.clone());

thread::spawn(move || {

    // intensive operations...

    let response = &client.get("http://google.com".parse().unwrap()).and_then(|res| {
        println!("Response: {}", res.status());
        Ok(())
    });

    remote.clone().spawn(|_| {
        response.map(|_| { () }).map_err(|_| { () })
    });

    // more intensive operations...
});
core.run(futures::future::empty::<(), ()>()).unwrap();

此代码无法编译：

thread::spawn(move || {
^^^^^^^^^^^^^ within `[closure@src/load-balancer.rs:46:19: 56:6 client:hyper::Client<hyper::client::HttpConnector>, remote:std::sync::Arc<tokio_core::reactor::Remote>]`, the trait `std::marker::Send` is not implemented for `std::rc::Weak<std::cell::RefCell<tokio_core::reactor::Inner>>`

thread::spawn(move || {
^^^^^^^^^^^^^ within `[closure@src/load-balancer.rs:46:19: 56:6 client:hyper::Client<hyper::client::HttpConnector>, remote:std::sync::Arc<tokio_core::reactor::Remote>]`, the trait `std::marker::Send` is not implemented for `std::rc::Rc<std::cell::RefCell<hyper::client::pool::PoolInner<tokio_proto::util::client_proxy::ClientProxy<tokio_proto::streaming::message::Message<hyper::http::MessageHead<hyper::http::RequestLine>, hyper::Body>, tokio_proto::streaming::message::Message<hyper::http::MessageHead<hyper::http::RawStatus>, tokio_proto::streaming::body::Body<hyper::Chunk, hyper::Error>>, hyper::Error>>>>`
...
remote.clone().spawn(|_| {
               ^^^^^ the trait `std::marker::Sync` is not implemented for `futures::Future<Error=hyper::Error, Item=hyper::Response> + 'static`

有没有办法从不同的线程或其他方法重用同一个客户端？

4

1 回答 1

4

简短的回答是否定的，但这样更好。

每个Client对象都拥有一个连接池。以下是 HyperPool在 0.11.0 版本中的定义：

pub struct Pool<T> {
    inner: Rc<RefCell<PoolInner<T>>>,
}

正如在运行时使用inner引用计数Rc和借用检查一样RefCell，池肯定不是线程安全的。当您尝试将其移至Client新线程时，该对象将持有一个存在于另一个线程中的池，这将成为数据竞争的来源。

这种实现是可以理解的。尝试跨多个线程重用 HTTP 连接并不常见，因为它需要对主要是 I/O 密集型资源的同步访问。这与 Tokio 的异步特性非常吻合。在同一个线程中执行多个请求实际上更合理，让 Tokio 的核心负责异步发送和接收消息，而不用依次等待每个响应。此外，计算密集型任务可以由futures_cpupool. 考虑到这一点，下面的代码可以正常工作：

extern crate tokio_core;
extern crate hyper;
extern crate futures;
extern crate futures_cpupool;

use tokio_core::reactor::Core;
use hyper::client::Client;
use futures::Future;
use futures_cpupool::CpuPool;

fn main() {

    let mut core = Core::new().unwrap();
    let handle = core.handle();
    let client = Client::new(&handle.clone());
    let pool = CpuPool::new(1);

    println!("Begin!");
    let req = client.get("http://google.com".parse().unwrap())
        .and_then(|res| {
            println!("Response: {}", res.status());
            Ok(())
        });
    let intensive = pool.spawn_fn(|| {
        println!("I'm working hard!!!");
        std::thread::sleep(std::time::Duration::from_secs(1));
        println!("Phew!");
        Ok(())
    });

    let task = req.join(intensive)
        .map(|_|{
            println!("End!");
        });
    core.run(task).unwrap();
}

如果响应没有收到太晚，输出将是：

Begin!
I'm working hard!!!
Response: 302 Found
Phew!
End!

如果您有多个任务在不同的线程中运行，那么问题就会变得开放，因为有多种架构是可行的。其中之一是将所有通信委托给单个参与者，因此需要所有其他工作线程将其数据发送给它。或者，您可以为每个工作人员提供一个客户端对象，因此也可以拥有单独的连接池。

于 2017-07-03T00:10:31.410 回答