advance/concurrency-with-threads/message-passing

[giscus[bot]]

advance/concurrency-with-threads/message-passing

https://course.rs/advance/concurrency-with-threads/message-passing.html

[MrRobertYuan]

在文章的 线程是如何结束的 章节，有这么一个例子：

use std::thread;
use std::time::Duration;
fn main() {
    // 创建一个线程A
    let new_thread = thread::spawn(move || {
        // 再创建一个线程B
        thread::spawn(move || {
            loop {
                println!("I am a new thread.");
            }
        })
    });

    // 等待新创建的线程执行完成
    new_thread.join().unwrap();
    println!("Child thread is finish!");

    // 睡眠一段时间，看子线程创建的子线程是否还在运行
    thread::sleep(Duration::from_millis(100));
}

想问一下这里如果新创建的线程一直没有结束，那主线程是不是也会一直等，而不是文章中说的会执行到 sleep 然后 100ms 后结束呢？
因为这里用了 .join() 我理解是它会一直等，只有结束了才会执行后面的语句？

[Ziqi418]

线程A在创建完线程B之后就结束了

[amwps290]

这里的 join 只会等待 A 线程，B 线程没有 join 所以不会等待其结束，如果想要等待 B 结束，也要加 join

[xiandaoyan]

main线程只会等A进程，不会等B进程，但是main进程会休眠100ms，在这100ms里边B进程可以不听的循环，但是100ms结束主线程就会推出，B就会over。 A线程创建完B线程就结束了，这100ms是主线程的休眠时间，不是主线程在等A线程。

[xiandaoyan]

main线程只会等A线程，不会等B线程，但是main线程会休眠100ms，在这100ms里边B线程可以不停的循环，但是100ms结束主线程就会退出，B就会over。 A线程创建完B线程就结束了，这100ms是主线程的休眠时间，不是主线程在等A线程。

[ANIDENTIFIER]

借用ss会在s之前释放，但编译器会报错怎么办

fn main() {
    let s = "1212".to_string();
    let ss = &s;

    let handle = thread::spawn(move || {
        println!("{}", ss);
    });

    handle.join().unwrap();
}

[oleolema]

在线程中借用外部的引用必须拥有'static生命周期

[BingCoke]

当你将ss放入到新得线程，就说明已经把变量放到新的上下文了，两个上下文（线程）是不互通的，编译器无法得知子线程使用的时候是否主线程已经把结构体销毁了，即使你在主线程join了，因为主线程的代码和子线程的代码是隔离起来的。

[Ziqi-Yang]

接上，不要使用引用，使用Arc或者s.clone()

[horizonshd]

同上面提到的，这里使用 Arc 更合适一些：

use std::{sync::Arc, thread};

fn main() {
    //let s = "1212".to_string();
    // let ss = &s;

    let s = Arc::new("1212".to_string());
    let ss = Arc::clone(&s);

    let handle = thread::spawn(move || {
        println!("{}", ss);
    });

    handle.join().unwrap();
}

[EnderHane]

如果需要捕获非 'static 引用，可以使用 std::thread::scope
它保证创建的线程在作用域内 join

[Leojangh]

妈的那个新手容易遇到的坑那里恰好困扰了我两天

[doraemonkeys]

crossbeam-channel官方文档的翻译：https://doraemon.xlog.app/crossbeam_channelmd

[TsGanT]

这里感觉很奇怪，那下面代码里 不论是tx还是tx1，都只有在main结束之后 才会被drop? 那岂不是跳不出for循环？

需要所有的发送者都被drop掉后，接收者rx才会收到错误，进而跳出for循环，最终结束主线程

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();
    let tx1 = tx.clone();
    thread::spawn(move || {
        tx.send(String::from("hi from raw tx")).unwrap();
    });

    thread::spawn(move || {
        tx1.send(String::from("hi from cloned tx")).unwrap();
    });

    for received in rx {
        println!("Got: {}", received);
    }
}

[Leojangh]

不是的，tx被move进了第一个thread，第一个thread分号后面就相当于tx被drop了，tx1同理，所以发送者全部被drop了，所以正常运行完毕退出。

[TsGanT]

奥 对对对 有道理 这也是为什么如果要多个发送者 就需要clone多个出来的原因 感谢感谢！

[duanyuluo]

move关键字说明是所有权转移，你在for received in rx这行代码之前使用tx，编译器不会通过的，因为主线程代码中已经没有所有权了。

[dayday2019]

讲的太好了 第一次看到这么高质量的rust的文章

[GIfted-Jarvis]

tx 和 rx 的全称是什么?

[carzpurzkey]

网络上，tx 是 Transmit 的缩写，rx 是 Receive 的缩写。

[horizonshd]

参考 这里 为 transmitter 和 receiver

[Ray-lyy]

请问一下，对于文章中的这段代码

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let vals = vec![
            String::from("hi"),
            String::from("from"),
            String::from("the"),
            String::from("thread"),
        ];

        for val in vals {
            tx.send(val).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

    for received in rx {
        println!("Got: {}", received);
    }
}

它tx每次发送之后都会休眠一秒，作为迭代器接收tx传递数据的rx，为什么能够自动等待到tx发送完全才脱离for循环，而不是接到一个循环一遍之后就直接跳出循环了呢？

[Leojangh]

因为这个自动等待是因为rx.recv()是阻塞的，tx发送完了tx被drop了rx.recv()就不阻塞了，收到中止信号而退出foreach

[Jevon617]

rx 实现了 Iterator trait, 源码如下

impl<T> Iterator for IntoIter<T> {
    type Item = T;
    fn next(&mut self) -> Option<T> {
        self.rx.recv().ok()
    }
}

所以你可以把 for received in rx 看成连续4次调用rx.recv().ok().

[Endermanbugzjfc]

（ok() 把 Result 转换成了 Option）：
Ok => Some
Err => None

[ShyunnY]

由于 rx.recv() 是阻塞的，又因为迭代器会不断调用 Iterator.next()

所以最终效果就是阻塞地在 for 中进行读取

[XPJ1993]

发送者全部drop或接收者被drop
下面的例子中怎么通过关闭 recv 去破坏这个等待呢？

[ADkun]

看本地标准库代码时，发现std::sync::mpsc实际上是封装了mpmc

#[stable(feature = "rust1", since = "1.0.0")]
pub struct Sender<T> {
    inner: mpmc::Sender<T>,
}

点开mpmc的实现会发现：
//! Multi-producer multi-consumer channels.

// This module is not currently exposed publicly, but is used
// as the implementation for the channels in sync::mpsc. The
// implementation comes from the crossbeam-channel crate:
//
// Copyright (c) 2019 The Crossbeam Project Developers

这里表明，标准库里的mpmc是从crossbeam中抄过来的，只不过不对外开放