多线程使用互斥锁并发访问向量

Question

多线程使用互斥锁并发访问向量

8

我正在使用Tokio库提供的示例，并尝试获得所有当前活动TCP连接的向量。最终，我想通过循环遍历它们并向套接字写入消息来能够向每个活动连接广播消息。

首先，在一个线程中我尝试打印出当前连接数量，而在另一个线程中接受连接。

为此，我正在尝试使用共享向量。目前还没有实现从向量中删除连接的功能，当然也要及时进行删除。

// A tiny async echo server with tokio-core
extern crate futures;
extern crate tokio_core;
extern crate tokio_io;

use futures::{Future, Stream};
use tokio_io::{io, AsyncRead};
use tokio_core::net::TcpListener;
use tokio_core::reactor::Core;
use std::thread;
use std::sync::{Arc, Mutex};
use std::io::stdout;
use std::io::Write;

fn main() {
    // Create the event loop that will drive this server
    let mut core = Core::new().unwrap();
    let handle = core.handle();

    // Bind the server's socket
    let addr = "127.0.0.1:12345".parse().unwrap();
    let tcp = TcpListener::bind(&addr, &handle).unwrap();

    let mut connections = Arc::new((Mutex::new(Vec::new())));

    thread::spawn(move || {
        //Every 10 seconds print out the current number of connections
        let mut i;
        loop {              
          i = connections.lock().unwrap().len();
          println!("There are {} connections", i);
          stdout().flush();
          thread::sleep_ms(10000);
        }
    });



    // Iterate incoming connections
    let server = tcp.incoming().for_each(|(tcp, _)| {

        connections.lock().unwrap().push(tcp);
        // Split up the read and write halves
        let (reader, writer) = tcp.split();

        // Future of the copy
        let bytes_copied = io::copy(reader, writer);

        // ... after which we'll print what happened
        let handle_conn = bytes_copied.map(|(n, _, _)| {
            println!("wrote {} bytes", n)
        }).map_err(|err| {
            println!("IO error {:?}", err)
        });

        // Spawn the future as a concurrent task
        handle.spawn(handle_conn);

        Ok(())
    });

    // Spin up the server on the event loop
    core.run(server).unwrap();

}

目前构建失败，出现以下错误：

error[E0382]: capture of moved value: `connections`
  --> src/main.rs:36:42
   |
26 |     thread::spawn(move || {
   |                   ------- value moved (into closure) here
...
36 |     let server = tcp.incoming().for_each(|(tcp, _)| {
   |                                          ^^^^^^^^^^ value captured here after move
   |
   = note: move occurs because `connections` has type `std::sync::Arc<std::sync::Mutex<std::vec::Vec<tokio_core::net::TcpStream>>>`, which does not implement the `Copy` trait

error[E0382]: use of moved value: `tcp`
  --> src/main.rs:40:32
   |
38 |         connections.lock().unwrap().push(tcp);
   |                                          --- value moved here
39 |         // Split up the read and write halves
40 |         let (reader, writer) = tcp.split();
   |                                ^^^ value used here after move
   |
   = note: move occurs because `tcp` has type `tokio_core::net::TcpStream`, which does not implement the `Copy` trait

没有编写任何不安全的代码，是否有可能实现这一点？

- John

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- MartinHaTh · Accepted Answer

你之所以会遇到第一个错误，是因为move闭包的原因：

let mut connections = Arc::new((Mutex::new(Vec::new())));
thread::spawn(move || {
    let mut i = connections.lock().unwrap().len();
    ....
}

实际上，这会移动整个Arc，但你只想移动它的“一部分”（也就是以增加引用计数的方式移动它，以便两个线程都可以使用它）。

为了做到这一点，我们可以使用Arc::clone：

let mut connections = Arc::new((Mutex::new(Vec::new())));
let conn = connections.clone();
thread::spawn(move || {
    let mut i = conn.lock().unwrap().len();
    ....
}

这样，克隆的Arc，conn被移动到闭包中，原始的Arc，connections没有被移动，因此仍可用。

我不确定你在处理第二个错误时具体在做什么，但为了简单地计数连接，你不需要push整个东西。