问题描述

我尝试使用actix-web提供的ChatServer来实现从rabbitmq中订阅数据，然后实现一个分发到websocket客户端的效果 然后我为此编写了一个函数。 叫做start_mq

// I rename the structure name 
impl RealtimeMq {
    fn send_message(&self, room: &str, message: &str, skip_id: usize) {
        if let Some(sessions) = self.rooms.get(room) {
            for id in sessions {
                if *id != skip_id {
                    if let Some(addr) = self.sessions.get(id) {
                        let _ = addr.do_send(Message(message.to_owned()));
                    }
                }
            }
        }
    }
    fn start_mq(&self) {
        let mq_config = QAEventMQ {
            amqp: CONFIG.AMQP.clone(),
            exchange: CONFIG.EXCHANGE.clone(),
            model: CONFIG.MODEL.clone(),
            routing_key: CONFIG.ROUTING_KEY.clone(),
        };
        let sender = self.sender.clone();
        let receiver = self.receiver.clone();
        thread::spawn(move || {
            QAEventMQ::consume(mq_config, sender)
        });
        thread::spawn(move || {
            loop {
                match receiver.recv() {
                    Ok(msg) => {
                        let data: Value = serde_json::from_str(&msg).unwrap();
                        let code = data["code"].as_str().unwrap();
                        println!("{:?}", data);
                        self.send_message(code, msg.as_str(), 0)
                    }
                    Err(e) => {
                        println!("read error ")
                    }
                }
            }
        });
    }
}

然后我在 default() 里面执行 start_mq 但是崩掉了

错误信息

error[E0277]: `(dyn actix::address::channel::Sender<realtime::Message> + 'static)` cannot be shared between threads safely
   --> src\realtime.rs:126:9
    |
126 |         thread::spawn(move || {
    |         ^^^^^^^^^^^^^ `(dyn actix::address::channel::Sender<realtime::Message> + 'static)` cannot be shared between threads safely
    |
    = help: the trait `std::marker::Sync` is not implemented for `(dyn actix::address::channel::Sender<realtime::Message> + 'static)`
    = note: required because of the requirements on the impl of `std::marker::Sync` for `std::ptr::Unique<(dyn actix::address::channel::Sender<realtime::Message> + 'static)>`
    = note: required because it appears within the type `std::boxed::Box<(dyn actix::address::channel::Sender<realtime::Message> + 'static)>`
    = note: required because it appears within the type `actix::address::Recipient<realtime::Message>`
    = note: required because it appears within the type `(usize, actix::address::Recipient<realtime::Message>)`
    = note: required because of the requirements on the impl of `std::marker::Sync` for `hashbrown::raw::RawTable<(usize, actix::address::Recipient<realtime::Message>)>`
    = note: required because it appears within the type `hashbrown::map::HashMap<usize, actix::address::Recipient<realtime::Message>, std::collections::hash_map::RandomState>`
    = note: required because it appears within the type `std::collections::HashMap<usize, actix::address::Recipient<realtime::Message>>`
    = note: required because it appears within the type `realtime::RealtimeMq`
    = note: required because of the requirements on the impl of `std::marker::Send` for `&realtime::RealtimeMq`
    = note: required because it appears within the type `[closure@src\realtime.rs:126:23: 140:10 receiver:crossbeam_channel::channel::Receiver<std::string::String>, self:&realtime::RealtimeMq]`

error: aborting due to previous error

这个是我拿来改的项目demo addr

我该咋做 才能修复它 ?