rust学习十六.2、并发-利用消息传递进行线程间通讯
通过信道(channel)传递消息是rust的解决线程之间通信的2个工具之一，另外1个是共享内存状态。注：channel有多种译法，有的地方翻译为通道、频道，此处循例称为“信道"
rust推出这个，明显地是因为受到go之类的影响。
在书籍中，作者提到go编程文档中的内容:
不要通过共享内存来通讯；而是通过通讯来共享内存（Do not communicate by sharing memory; instead, share memory by communicating）
因此，我这里就按部就班复述下书本上关于这个篇章的内容。
一、概述
* 1.mpsc 是 multiply producer single consumer 的缩写，即多生产者单消费者模式.意味着* rust的主要消息模式不允许一个生产者多个消费者。但rust也没有说不支持一个生产者多个消费者模式* 2.(tx, rx) = mpsc::channel() 会返回一个元组，分别表示发送和接收者* 3. rx有两个重要方法，分别是recv() 和 try_recv().前者会阻塞，后者不会。* 4. 信道和所有权问题. tx.send方法会默认夺取消息的所有权，移动这个值归接收者所有.所以发送者需要拥有消息的所有权。* 这也同时意味着，如果没有特别措施，那么发送之后，被发送的消息不能再访问了。* 那么，我们是否宁愿clone()一下，然后再发送呢？是的。* rust的所有权系统，导致rust会把消息视为一个实物，而不是一个拷贝，意味着发送就是“脱手”、“出手”* 5. 发送者可以关闭信道，这样接收者就会收到一个错误。* 6. 发送者可以克隆，这样多个线程都可以发送消息。发送者虽然克隆了，但是接收者还是同一个. 7. 这种方式是否可能存在这样的问题：后发先至。 这个问题暂时没有明确答案，因为暂时没有找到答案。据说概率很小，但不是么有？二、简要示例
2.1、例子1_正常发送迭代接收
这个例子把原书的简单地做了一些修改，以便更有乐趣些！
和原书稍微不同的是，接收者会根据消息来源把收到的内容分别保存到v0,v1的向量中,最后打印各自拼接的消息
大体模拟了碟战中，收到不同消息来源的情景！
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
#
#
struct Message {
sender: String, // 发送者标识
content: String, // 消息内容
order: u8, // 消息序号
}
fn main() {
// --snip--
let (tx, rx) = mpsc::channel(); //channel() 返回一个元组，分别表示发送和接收者
let tx1 = tx.clone();
thread::spawn(move || { // 移动tx1所有权到新线程中。为什么需要move？规定就是这样
let vals = vec![
Message{sender: String::from("x1"),content: String::from("风"),order:1},
Message{sender: String::from("x1"),content: String::from("紧"),order:2},
Message{sender: String::from("x1"),content: String::from("扯"),order:3},
Message{sender: String::from("x1"),content: String::from("呼"),order:4}
];
for val in vals {
tx1.send(val).unwrap();
println!("tx1 is sending....");
thread::sleep(Duration::from_millis(200));
}
});
thread::spawn(move || {
let vals = vec![
Message{sender: String::from("x0"),content: String::from("乘"),order:1},
Message{sender: String::from("x0"),content: String::from("胜"),order:2},
Message{sender: String::from("x0"),content: String::from("追"),order:3},
Message{sender: String::from("x0"),content: String::from("击"),order:4}
];
for val in vals {
tx.send(val).unwrap();
println!("tx0 is sending....");
thread::sleep(Duration::from_millis(100));
}
});
//把收到的消息分别放入到两个Vec中，以便于观察顺序。
let mut v0 = Vec::new();
let mut v1 = Vec::new();
let mut qty=0;
//利用rx的迭代器接收消息. 无需显示调用rec(),或者try_recv()，迭代器会自动调用。
for received in rx {
if received.sender == "x0" {
v0.push(received);
}
else{
v1.push(received);
}
qty += 1;
if qty==5 { // 收到5条消息就退出循环。看看会不会有什么问题
break;
}
}
let msg0 = get_msg(v0);
let msg1 = get_msg(v1);
println!("从t0收到的消息是: {}", msg0);
println!("从t1收到的消息是: {}", msg1);
if msg0!=msg1 {
println!("那一份消息是真的？应该信任谁？还是说都是真的？\n 如果都是真的，为什么会这样？");
}
}
fn get_msg(msg:Vec<Message>) -> String {
let mut result = String::new();
for m in msg {
result.push_str(&m.content);
}
return result;
}当主线程中，设置为接收5条消息：
当主线程中，设置为接收20条（根本不可能）：
2.2、例子2_try_recv()
很遗憾，第一个例子么有try_recv(),所以这里采用这个试试。
根据描述try_recv()不阻塞主线程，所以可以干一些其它的事情！！！这是我感兴趣的地方，所以把例子稍微修改了下！
//利用try_recv()接收消息，并打印出来。
loop {
match rx.try_recv() {
Ok(val) => {
if val.sender == "x0" {
v0.push(val);
}
else{
v1.push(val);
}
qty+=1;
}
Err(_) => {
if qty==8 {
break;
}
thread::sleep(Duration::from_millis(200));
}
}
println!("...快点啊！");
}其余部分同示例1，具体略。
结果如下：
用这个loop结构，有个问题：如何知道消息已经接收完毕了？ 示例代码只能写死。 但实际应该不能这样，太不友好！
从这个方面来说，不如用迭代友好！ 反之，try_recv()让计算机可以干一点别的事情...，充分利用资源，如果有的话！
2.3、例子3 _不接收
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
#
#
struct Message {
sender: String, // 发送者标识
content: String, // 消息内容
order: u8, // 消息序号
}
fn main() {
// --snip--
let (tx, rx) = mpsc::channel(); //channel() 返回一个元组，分别表示发送和接收者
let tx1 = tx.clone();
let handle1=thread::spawn(move || { // 移动tx1所有权到新线程中。为什么需要move？规定就是这样
let vals = vec![
Message{sender: String::from("x1"),content: String::from("风"),order:1},
Message{sender: String::from("x1"),content: String::from("紧"),order:2},
Message{sender: String::from("x1"),content: String::from("扯"),order:3},
Message{sender: String::from("x1"),content: String::from("呼"),order:4}
];
for val in vals {
tx1.send(val).unwrap();
println!("tx1 is sending....");
thread::sleep(Duration::from_millis(200));
}
});
let handle0=thread::spawn(move || {
let vals = vec![
Message{sender: String::from("x0"),content: String::from("乘"),order:1},
Message{sender: String::from("x0"),content: String::from("胜"),order:2},
Message{sender: String::from("x0"),content: String::from("追"),order:3},
Message{sender: String::from("x0"),content: String::from("击"),order:4}
];
for val in vals {
tx.send(val).unwrap();
println!("tx0 is sending....");
thread::sleep(Duration::from_millis(100));
}
});
handle0.join().unwrap();
handle1.join().unwrap();
}
三、疑问
1.后发先到问题
这个暂时没有结论！ 待后续补充!!
2.如何中断发送或者接收
接收者退出即可，即不执行recv,或者try_recv即可! 但这个不会终端发送，只会终端接收。
不是我想象的那样：如果用电话线通话，砍断电话线，是不是应该两端都会收到故障响应？ 但是我已经习得的告诉我不是那么回事。
当然，我要求的不是主动触发错误（异常）。
有关内容，有待继续完善！
3.如果没有接收动作，会发送得出去吗？
可以，因为没有报告异常！！！
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！
页:
[1]