tokio tcp通信

引入crate

tokio = { version = "1.35.1", features = ["full"] }

服务端

use std::time::Duration;
use tokio::{io::{AsyncBufReadExt, AsyncWriteExt},net::{tcp::{OwnedReadHalf, OwnedWriteHalf},TcpListener, TcpStream,},sync::mpsc,
};#[tokio::main]
async fn main() {println!("Begin Start Server...");let server = TcpListener::bind("127.0.0.1:10888").await.unwrap();while let Ok((client_stream, client_addr)) = server.accept().await{println!("accept client: {}", client_addr);tokio::spawn(async move{process_client(client_stream).await;});}
}async fn process_client(client_stream: TcpStream){let (client_reader, client_writer) = client_stream.into_split();let (msg_tx, msg_rx) = mpsc::channel::<String>(100);let mut read_task = tokio::spawn(async move {read_from_client(client_reader, msg_tx).await;});let mut write_task = tokio::spawn(async move{write_to_client(client_writer, msg_rx).await;});if tokio::try_join!(&mut read_task, &mut write_task).is_err() {read_task.abort();write_task.abort();};
}async fn read_from_client(reader: OwnedReadHalf, mst_tx: mpsc::Sender<String>){let mut buf_reader = tokio::io::BufReader::new(reader);let mut buf = String::new();loop{match buf_reader.read_line(&mut buf).await{Err(_e) =>{eprintln!("read from client error");break;}Ok(0) =>{println!("client closed");break;}Ok(n) => {buf.pop(); //去除末尾的回车符let mut content = buf.drain(..).as_str().to_string();println!("read {} bytes from client. content: {}", n, content);tokio::time::sleep(Duration::from_secs(1)).await;content.push('\n');if mst_tx.send(content).await.is_err(){eprintln!("receiver closed");break;}}}}
}async fn write_to_client(writer: OwnedWriteHalf, mut msg_rx: mpsc::Receiver<String>){let mut buf_writer = tokio::io::BufWriter::new(writer);while let Some(mut str) = msg_rx.recv().await{//str.push('\n');if let Err(e) = buf_writer.write_all(str.as_bytes()).await {eprintln!("write to client failed: {}", e);break;}buf_writer.flush().await;print!("write to client: {}", str);}
}

客户端

use std::sync;
use std::time::Duration;
use std::time::{SystemTime, UNIX_EPOCH};
use chrono::{Local, NaiveDateTime};use tokio::{io::{Interest, AsyncBufReadExt, AsyncWriteExt},net::{tcp::{OwnedReadHalf, OwnedWriteHalf},TcpStream,},sync::mpsc,
};#[tokio::main]
async fn main() {let stream = TcpStream::connect("127.0.0.1:10888").await.unwrap();let (reader, writer) = stream.into_split();let mut buf = String::new();//[0u8, 12];let mut buf_reader = tokio::io::BufReader::new(reader);let mut buf_writer = tokio::io::BufWriter::new(writer);loop{let now=Local::now();let formatted=now.format("%Y-%m-%d %H:%M:%S");let content = format!("hello world {}\n", formatted);//buf_writer.write_all(b"hello world\n").await;buf_writer.write_all(content.as_bytes()).await;buf_writer.flush().await;println!("send:{}", content);match buf_reader.read_line(&mut buf).await {Err(_e) =>{eprintln!("read from server error");break;}Ok(n) =>{buf.pop();let content = buf.as_str().to_string();println!("received: {}  {}", n, content);}};}}

客户端、服务端都使用TcpStream的into_split方法获取网络通信读和写实例，进而获取buffer读写对象，通过channel实现线程执行同步。子线程使用tokio::spawn函数启动。