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

Rust 并发编程高级应用：从入门到精通

news 2026/5/8 4:37:47

Rust 并发编程高级应用：从入门到精通

作为一名从Python转向Rust的后端开发者，我深刻体会到Rust并发编程的强大和安全。Rust的并发模型不仅可以帮助我们充分利用多核CPU，还可以在编译时保证线程安全，这让我在编写高并发服务时更加自信。今天，我想分享一下Rust并发编程的高级应用，希望能帮助大家更好地理解和使用这个强大的特性。

一、并发编程的基本概念

1. 什么是并发编程

并发编程是指同时执行多个任务的编程范式，它可以提高程序的性能和响应速度。

2. Rust的并发模型

Rust提供了两种并发模型：

线程（Threads）：使用std::thread创建和管理线程
异步（Async）：使用async/await语法和异步运行时

// 线程示例 use std::thread; use std::time::Duration; fn main() { let handle = thread::spawn(|| { for i in 1..10 { println!("Thread: {}", i); thread::sleep(Duration::from_millis(100)); } }); for i in 1..5 { println!("Main: {}", i); thread::sleep(Duration::from_millis(100)); } handle.join().unwrap(); } // 异步示例 use tokio::time::Duration; #[tokio::main] async fn main() { tokio::spawn(async { for i in 1..10 { println!("Async: {}", i); tokio::time::sleep(Duration::from_millis(100)).await; } }); for i in 1..5 { println!("Main: {}", i); tokio::time::sleep(Duration::from_millis(100)).await; } }

二、高级应用技巧

1. 共享状态管理

我们可以使用Arc<T>和Mutex<T>来管理共享状态。

use std::sync::{Arc, Mutex}; use std::thread; fn main() { let counter = Arc::new(Mutex::new(0)); let mut handles = vec![]; for _ in 0..10 { let counter = Arc::clone(&counter); let handle = thread::spawn(move || { let mut num = counter.lock().unwrap(); *num += 1; }); handles.push(handle); } for handle in handles { handle.join().unwrap(); } println!("Result: {}", *counter.lock().unwrap()); }

2. 无锁数据结构

我们可以使用atomic模块来实现无锁数据结构。

use std::sync::atomic::{AtomicUsize, Ordering}; use std::thread; fn main() { let counter = AtomicUsize::new(0); let mut handles = vec![]; for _ in 0..10 { let counter = &counter; let handle = thread::spawn(move || { counter.fetch_add(1, Ordering::SeqCst); }); handles.push(handle); } for handle in handles { handle.join().unwrap(); } println!("Result: {}", counter.load(Ordering::SeqCst)); }

3. 消息传递

我们可以使用mpsc（multiple producer, single consumer）来实现线程间的消息传递。

use std::sync::mpsc; use std::thread; fn main() { let (tx, rx) = mpsc::channel(); for i in 0..5 { let tx = tx.clone(); thread::spawn(move || { tx.send(i).unwrap(); }); } drop(tx); // 关闭发送端 for received in rx { println!("Received: {}", received); } }

三、实用示例

1. 实现一个线程池

我们可以实现一个线程池来管理和复用线程。

use std::sync::{mpsc, Arc, Mutex}; use std::thread; struct ThreadPool { workers: Vec<Worker>, sender: mpsc::Sender<Job>, } type Job = Box<dyn FnOnce() + Send + 'static>; impl ThreadPool { fn new(size: usize) -> ThreadPool { assert!(size > 0); let (sender, receiver) = mpsc::channel(); let receiver = Arc::new(Mutex::new(receiver)); let mut workers = Vec::with_capacity(size); for id in 0..size { workers.push(Worker::new(id, Arc::clone(&receiver))); } ThreadPool { workers, sender, } } fn execute<F>(&self, f: F) where F: FnOnce() + Send + 'static { let job = Box::new(f); self.sender.send(job).unwrap(); } } struct Worker { id: usize, thread: thread::JoinHandle<()>, } impl Worker { fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker { let thread = thread::spawn(move || { loop { let job = receiver.lock().unwrap().recv(); match job { Ok(job) => { println!("Worker {} got a job; executing.", id); job(); } Err(_) => { println!("Worker {} disconnected; shutting down.", id); break; } } } }); Worker { id, thread, } } } fn main() { let pool = ThreadPool::new(4); for i in 0..8 { pool.execute(move || { println!("Hello from task {}", i); }); } // 线程池会在Drop时自动关闭 }

2. 实现一个异步任务调度器

我们可以使用tokio来实现一个异步任务调度器。

use tokio::time::Duration; use tokio::task; async fn process_task(id: usize) { println!("Processing task {}", id); tokio::time::sleep(Duration::from_millis(100)).await; println!("Task {} completed", id); } #[tokio::main] async fn main() { let mut handles = vec![]; for i in 0..10 { let handle = task::spawn(async move { process_task(i).await; }); handles.push(handle); } for handle in handles { handle.await.unwrap(); } println!("All tasks completed"); }

3. 实现一个并发下载器

我们可以使用tokio和reqwest来实现一个并发下载器。

use tokio::task; use reqwest::Client; async fn download_url(client: &Client, url: &str) -> Result<String, reqwest::Error> { let response = client.get(url).send().await?; response.text().await } #[tokio::main] async fn main() { let client = Client::new(); let urls = [ "https://www.example.com", "https://www.google.com", "https://www.github.com", "https://www.python.org", "https://www.rust-lang.org" ]; let mut tasks = vec![]; for url in &urls { let client = client.clone(); let url = url.to_string(); tasks.push(task::spawn(async move { match download_url(&client, &url).await { Ok(content) => println!("Downloaded {}: {} bytes", url, content.len()), Err(e) => println!("Error downloading {}: {:?}", url, e), } })); } for task in tasks { task.await.unwrap(); } }

四、高级并发技术

1. 异步流

我们可以使用Stream特质来处理异步流数据。

use tokio::stream::StreamExt; use tokio::time::Duration; #[tokio::main] async fn main() { let mut stream = tokio::stream::iter(0..10); while let Some(item) = stream.next().await { println!("Item: {}", item); tokio::time::sleep(Duration::from_millis(100)).await; } }

2. 超时处理

我们可以使用tokio::time::timeout来处理超时。

use tokio::time::{timeout, Duration}; async fn long_running_task() { tokio::time::sleep(Duration::from_secs(2)).await; println!("Task completed"); } #[tokio::main] async fn main() { match timeout(Duration::from_secs(1), long_running_task()).await { Ok(_) => println!("Task completed within timeout"), Err(_) => println!("Task timed out"), } }

3. 优雅关闭

我们可以使用tokio::signal来实现优雅关闭。

use tokio::signal; use tokio::time::Duration; #[tokio::main] async fn main() { println!("Server started"); // 等待中断信号 let ctrl_c = signal::ctrl_c().await; match ctrl_c { Ok(()) => println!("Received Ctrl+C, shutting down gracefully"), Err(e) => println!("Error receiving signal: {:?}", e), } // 执行清理操作 println!("Performing cleanup..."); tokio::time::sleep(Duration::from_secs(1)).await; println!("Server shutdown"); }

五、实战应用

1. 实现一个并发Web服务器

我们可以使用tokio和hyper来实现一个并发Web服务器。

use hyper::service::{make_service_fn, service_fn}; use hyper::{Body, Request, Response, Server, StatusCode}; async fn handle_request(_req: Request<Body>) -> Result<Response<Body>, hyper::Error> { Ok(Response::new(Body::from("Hello, World!"))) } #[tokio::main] async fn main() { let addr = ([127, 0, 0, 1], 3000).into(); let make_svc = make_service_fn(|_conn| async { Ok::<_, hyper::Error>(service_fn(handle_request)) }); let server = Server::bind(&addr).serve(make_svc); println!("Server running on http://127.0.0.1:3000"); if let Err(e) = server.await { eprintln!("Server error: {}", e); } }

2. 实现一个并发数据库连接池

我们可以使用tokio和deadpool-postgres来实现一个并发数据库连接池。

use deadpool_postgres::{Config, Pool}; use tokio_postgres::NoTls; #[tokio::main] async fn main() { let mut config = Config::new(); config.host = Some("localhost".to_string()); config.user = Some("postgres".to_string()); config.password = Some("password".to_string()); config.dbname = Some("test".to_string()); let pool = config.create_pool(NoTls).unwrap(); // 并发执行多个查询 let mut tasks = vec![]; for i in 0..10 { let pool = pool.clone(); tasks.push(tokio::spawn(async move { let client = pool.get().await.unwrap(); let row = client.query_one("SELECT $1::text", &[&format!("Hello, {}!", i)]).await.unwrap(); let message: String = row.get(0); println!("Query {} result: {}", i, message); })); } for task in tasks { task.await.unwrap(); } }

3. 实现一个并发消息处理器

我们可以使用tokio和rdkafka来实现一个并发消息处理器。

use rdkafka::consumer::{Consumer, StreamConsumer}; use rdkafka::message::Message; use rdkafka::ClientConfig; use tokio::stream::StreamExt; #[tokio::main] async fn main() { let consumer: StreamConsumer = ClientConfig::new() .set("group.id", "test-group") .set("bootstrap.servers", "localhost:9092") .set("auto.offset.reset", "earliest") .create() .expect("Consumer creation failed"); consumer .subscribe(&["test-topic"]) .expect("Subscription failed"); let mut stream = consumer.stream(); while let Some(message) = stream.next().await { match message { Ok(m) => { let payload = m.payload().unwrap_or_default(); println!("Received message: {}", String::from_utf8_lossy(payload)); } Err(e) => println!("Error receiving message: {:?}", e), } } }