multithreading - 无法跨mpsc::Sender作为字段发送跨线程结构

Question

我有一个结构，其字段为Sender类型。

pub struct GenericConnectionPool<E>
where
    E: ConnectionConnector,
{
    _sender: Sender<()>,
    _reciever: Arc<Mutex<Receiver<()>>>,
    _num_of_live_connections: AtomicU8,
    _max_connections: u8,
    _min_connections: u8,
    _connections: Arc<Mutex<Vec<<E as ConnectionConnector>::Conn>>>,
    _connector: E,
}

我在多个线程中使用该结构，这就是为什么在Arc中使用它并克隆它时的原因。

let pool = Arc::new(GenericConnectionPool::new(2, 1, cc));
println!("here");
{
    for _ in 0..3 {
        let pool = Arc::clone(&pool);
        std::thread::spawn(move || {
            pool.get_connection();
            thread::sleep(Duration::from_secs(1));
        });
    }
}

但是我收到一个错误，我的结构无法跨线程发送。

`std::sync::mpsc::Sender<()>` cannot be shared between threads safely
within `GenericConnectionPool<tests::connector_works::DummyConnectionConnector>`, the trait `std::marker::Sync` is not implemented for `std::sync::mpsc::Sender<()>`
required because it appears within the type `GenericConnectionPool<tests::connector_works::DummyConnectionConnector>`
required because of the requirements on the impl of `std::marker::Send` for `std::sync::Arc<GenericConnectionPool<tests::connector_works::DummyConnectionConnector>>`
required because it appears within the type `[closure@src/lib.rs:169:36: 172:18 pool:std::sync::Arc<GenericConnectionPool<tests::connector_works::DummyConnectionConnector>>]`

我了解的是Sender类型不能在线程间安全地发送，但是由于它是cloneable，您可以克隆它，然后在线程间发送它。但是在我的情况下，发件人位于我的Struct内部。我想不出办法解决此问题。
我认为我可能必须更改设计。

Answer 1

您可以使用Crossbeam。它的crossbeam::Sender似乎可以在线程之间转移。大概，您还需要使用其crossbeam::Receiver。
或者，您可以将GenericConnectionPool重构为如下形式:

pub struct ExtraData {}

#[derive(Clone)]
pub struct GenericConnectionPool {
    _sender: Sender<()>,
    _extra_data: Arc<ExtraData>,
}

那么您可以直接克隆GenericConnectionPool而不是克隆包含它的Arc并获得正确的行为:

let pool = GenericConnectionPool{_sender:s, _extra_data:Arc::new(ExtraData{}) };

for _ in 0..3 {
   let pool = pool.clone();
   std::thread::spawn(move || {
       pool.get_connection();
       thread::sleep(Duration::from_secs(1));
    });
}

您可以看到一个编译版本here: