当前位置：首页 > news >正文

Rust 实战练习 - 4. 网络 TCP/UDP/Channel

news 来源：原创 2024/5/10 3:01:46

目标：

tcp 服务器和客户端
udp服务器和客户端
多线程实现服务器，异步并发
channel

tcp server & client

use std::{prelude::*, net, thread, env};
use std::io::{Read, Write};fn main() {let args = env::args().into_iter().collect::<Vec<_>>();if args.len()>1 {match args[1].as_str() {"-s" => tcp_server(),"-c" => tcp_client(),_ => println!("unknown cmd {}", args[1]),}}else{println!("Usage:\r\n\t-s Open Tcp Server.\r\n\t-c Open Tcp client to connect the server.")}
}
fn tcp_server() {let s = net::TcpListener::bind("0.0.0.0:8000").unwrap();println!("Listen on addr: {}", s.local_addr().unwrap().to_string());for req in s.incoming() {if let Ok(req_s) = req {_ = req_s.set_nodelay(true);thread::spawn( move || {handler_tcp(req_s);});}}
}
fn handler_tcp(mut c: net::TcpStream) {let mut buf = [0u8;1024];let info = format!("[{:?}] => client in: {}", thread::current().id(), c.peer_addr().unwrap().to_string());let n = c.read(&mut buf).unwrap();println!("{} {}", n, String::from_utf8_lossy(&buf[..n]));println!("{}", info);_ = c.write(format!("HTTP/1.1 200 OK\r\n\r\n{}\r\n", info).as_bytes());
}fn tcp_client(){let mut c = net::TcpStream::connect("127.0.0.1:8000").unwrap();c.set_nodelay(true).unwrap();_ = c.write("GET / HTTP/1.1\r\nAccept: */*\r\n\r\n".as_bytes());let mut strbuf = String::new();_ = c.read_to_string(&mut strbuf);println!("resp: {}", strbuf);
}

异步版本

use std::fs;
use std::io::prelude::*;
use std::net::TcpListener;
use std::net::TcpStream;fn main() {// Listen for incoming TCP connections on localhost port 7878let listener = TcpListener::bind("127.0.0.1:7878").unwrap();// Block forever, handling each request that arrives at this IP addressfor stream in listener.incoming() {let stream = stream.unwrap();handle_connection(stream);}
}fn handle_connection(mut stream: TcpStream) {// Read the first 1024 bytes of data from the streamlet mut buffer = [0; 1024];stream.read(&mut buffer).unwrap();let get = b"GET / HTTP/1.1\r\n";// Respond with greetings or a 404,// depending on the data in the requestlet (status_line, filename) = if buffer.starts_with(get) {("HTTP/1.1 200 OK\r\n\r\n", "hello.html")} else {("HTTP/1.1 404 NOT FOUND\r\n\r\n", "404.html")};let contents = fs::read_to_string(filename).unwrap();// Write response back to the stream,// and flush the stream to ensure the response is sent back to the clientlet response = format!("{status_line}{contents}");stream.write_all(response.as_bytes()).unwrap();stream.flush().unwrap();
}

改写

[package]
name = "d5"
version = "0.1.0"
edition = "2021"# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html[dependencies]
futures="*"[dependencies.async-std]
version = "*"
features = ["attributes"]

use async_std::net::TcpListener;
use async_std::net::TcpStream;
use futures::stream::StreamExt;
use async_std::task::spawn;
use async_std::prelude::*;
use std::fs;#[async_std::main]
async fn main() {let listener = TcpListener::bind("127.0.0.1:7878").await.unwrap();listener.incoming().for_each_concurrent(/* limit */ None, |tcpstream| async move {let tcpstream = tcpstream.unwrap();//spawn(handle_connection(stream));handle_connection(tcpstream).await;}).await;
}async fn handle_connection(mut stream: TcpStream) {// Read the first 1024 bytes of data from the streamlet mut buffer = [0; 1024];stream.read(&mut buffer).await.unwrap();let get = b"GET / HTTP/1.1\r\n";// Respond with greetings or a 404,// depending on the data in the requestlet (status_line, filename) = if buffer.starts_with(get) {("HTTP/1.1 200 OK\r\n\r\n", "hello.html")} else {("HTTP/1.1 404 NOT FOUND\r\n\r\n", "404.html")};let contents = fs::read_to_string(filename).unwrap();// Write response back to the stream,// and flush the stream to ensure the response is sent back to the clientlet response = format!("{status_line}{contents}");stream.write_all(response.as_bytes()).await.unwrap();stream.flush().await.unwrap();
}

UDP

广播使用广播地址255.255.255.255，将消息发送到在同一广播网络上的每个主机，广播仅仅在同一局域网上才能进行，但是广播还是要指明接收者的端口号的
多播，也称为“组播”，与单播一样，多播是允许在广域网即Internet上进行传输的，多播的地址是特定的，D类地址用于多播。即224.0.0.0至239.255.255.255之间的IP地址。

1、局部多播地址：在224.0.0.0～224.0.0.255之间，这是为路由协议和其他用途保留的地址，路由器并不转发属于此范围的IP包。

2、预留多播地址：在224.0.1.0～238.255.255.255之间，可用于全球范围（如Internet）或网络协议。

3、管理权限多播地址：在239.0.0.0～239.255.255.255之间，可供组织内部使用，类似于私有IP地址，不能用于Internet，可限制多播范围。
单播的数据只是收发数据的特定主机进行处理，根据地址不同，可以跨越局域网进行internet通讯。
- 单播流程：主机A向主机B发送UDP数据报，发送的目的IP为192.168.1.151，端口为 80，目的MAC地址为00:00:00:00:00:02。
- 广播的流程：主机A向整个网络发送广播数据，发送的目的IP为192.168.1.255，端口为 80，目的MAC地址为FF:FF:FF:FF:FF:FF。

use std::{prelude::*, net, thread, env};
use std::io::{Read, Write};fn main() {let args = env::args().into_iter().collect::<Vec<_>>();if args.len()>1 {match args[1].as_str() {"-s" => udp_server(),"-c" => udp_client(),_ => println!("unknown cmd {}", args[1]),}}else{println!("Usage:\r\n\t-s Open UDP Server.\r\n\t-c Open UDP client to connect the server.")}
}
fn udp_server() {let s = net::UdpSocket::bind("0.0.0.0:8000").unwrap();println!("Listen on addr: {}", s.local_addr().unwrap().to_string());let mut buf = [0u8; 1024];loop{if let Ok((n, addr)) = s.recv_from(&mut buf) {println!("addr: {}, content: {}", addr.to_string(), String::from_utf8_lossy(&buf[..n]));_ = s.send_to(format!("ok! you are: {}", addr.to_string()).as_bytes(), addr);}}
}fn udp_client(){// 不指定地址let mut c = net::UdpSocket::bind("0.0.0.0:0").unwrap();println!("client addr: {}", c.local_addr().unwrap().to_string());let s_addr = "127.0.0.1:8000";let mut buf = [0u8; 1024];for i in 0..10 {_ = c.send_to(format!("client: I'm coming! on {}!", i).as_bytes(), s_addr);if let Ok((n, addr)) = c.recv_from(&mut buf) {println!("client: s addr: {}, content: {}", addr.to_string(), String::from_utf8_lossy(&buf[..n]));}}
}

Channel

通道（Channel）是一种用于在多个线程之间传递数据的并发原语。通道提供了一种安全且高效的方式，允许线程之间进行通信和同步。

我们可以使用 std::sync::mpsc 模块提供的 channel 函数来创建一个通道。mpsc 是“多个生产者，单个消费者”（Multiple Producers, Single Consumer）的缩写，意味着多个线程可以同时向通道发送数据，但只有一个线程可以从通道接收数据。

use std::sync::mpsc;
use std::thread;
use std::time::Duration;fn main() {// 创建通道，返回发送者和接收者// 默认是异步的，不阻塞对方。// mpsc::sync_channel(0) 就是同步的，缓冲区满了就会阻塞let (tx, rx) = mpsc::channel();// thread 1let tx2 = tx.clone();thread::spawn(move || {for i in 0..30 {let message = format!("2 - {}. Hello from the sender!", i);tx2.send(message).unwrap();thread::sleep(Duration::from_secs(1));}println!("Thread 2 End");// drop(tx2);});// thread 2thread::spawn(move || {for i in 0..20 {let message = format!("1 - {}. Hello from the sender!", i);tx.send(message).unwrap();thread::sleep(Duration::from_secs(1));}println!("Thread 1 End");// drop(tx);});// 在主线程接收数据// for received in rx {//     println!("Got: {}", received);// }loop {if let Ok(r) = rx.recv() {println!("Received: {}", r);}else{println!("Receiced error!");break;}}println!("End");
}

一个常见的坑

use std::sync::mpsc;
fn main() {use std::thread;let (send, recv) = mpsc::channel();let num_threads = 3;for i in 0..num_threads {let thread_send = send.clone();thread::spawn(move || {thread_send.send(i).unwrap();println!("thread {:?} finished", i);});}// 在这里如果不drop,会因为Send的生命周期一直到main结束，所以recv也不会结束，所以无法退出// drop(send);for x in recv {println!("Got: {}", x);}println!("finished iterating");
}