Rust Tokio
Build high-performance async applications in Rust using the Tokio runtime.
Quick Start
use tokio::net::TcpListener;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use std::sync::Arc;
use tokio::sync::Mutex;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let listener = TcpListener::bind("127.0.0.1:8080").await?;
let counter = Arc::new(Mutex::new(0u64));
println!("Server listening on :8080");
loop {
let (socket, addr) = listener.accept().await?;
let counter = counter.clone();
tokio::spawn(async move {
println!("New connection: {}", addr);
let (reader, mut writer) = socket.into_split();
let mut buf_reader = BufReader::new(reader);
let mut line = String::new();
loop {
line.clear();
match buf_reader.read_line(&mut line).await {
Ok(0) => break, // EOF
Ok(_) => {
let response = format!("Echo: {}", line.trim());
// Update shared counter
let mut count = counter.lock().await;
*count += 1;
drop(count);
if let Err(e) = writer.write_all(response.as_bytes()).await {
eprintln!("Write error: {}", e);
break;
}
if let Err(e) = writer.write_all(b"\n").await {
eprintln!("Write error: {}", e);
break;
}
}
Err(e) => {
eprintln!("Read error: {}", e);
break;
}
}
}
println!("Connection closed: {}", addr);
});
}
}// Async HTTP client
use reqwest;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Concurrent requests with join
let urls = vec![
"https://api.example.com/users",
"https://api.example.com/products",
"https://api.example.com/orders",
];
let mut handles = Vec::new();
for url in urls {
handles.push(tokio::spawn(async move {
reqwest::get(url).await.unwrap().text().await.unwrap()
}));
}
// Join all concurrent requests
for handle in handles {
let result = handle.await?;
println!("Got response: {} bytes", result.len());
}
Ok(())
}Key Concepts
Tokio is an async runtime for Rust providing I/O, timers, synchronization primitives, and task scheduling. Use `#[tokio::main]` to enter async context. `tokio::spawn` creates concurrent tasks. Use `Mutex`/`RwLock` for shared state.
When to Use
- High-throughput network services (HTTP servers, proxies)
- Real-time systems (chat, gaming, streaming)
- Concurrent data processing pipelines
- Microservices requiring maximum performance
Step-by-Step
1. Add Tokio: `cargo add tokio --features full` and set an async `main` with `#[tokio::main]`.
2. Pick the I/O primitive: `TcpListener`/`TcpStream` for sockets, `UnixListener` for local pipes, `tokio::fs` for async file access.
3. Handle each connection by `tokio::spawn` — never block tokio worker threads with synchronous work.
4. Share state via `tokio::sync::Mutex`/`RwLock` (or actor channels) cloned into tasks.
5. Add concurrency: use `tokio::join!`/`try_join!` for parallel awaits and `tokio::time::timeout` for deadlines.
6. Tune the runtime: set worker threads and run tests with `cargo test` under `--cfg tokio_unstable` if multi-thread stats are needed.
Examples
// Fan-out to N concurrent tasks and aggregate results
use tokio::time::{sleep, Duration};
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let handles: Vec<_> = (0..5)
.map(|i| tokio::spawn(async move {
sleep(Duration::from_millis(100 * i)).await;
i * i
}))
.collect();
let mut total = 0;
for h in handles {
total += h.await?;
}
println!("sum of squares = {total}");
Ok(())
}// Graceful shutdown with CancellationToken
use tokio_util::sync::CancellationToken;
#[tokio::main]
async fn main() {
let token = CancellationToken::new();
let child = token.clone();
tokio::spawn(async move {
tokio::select! {
_ = child.cancelled() => println!("task cancelled"),
_ = tokio::time::sleep(Duration::from_secs(60)) => println!("task done"),
}
});
tokio::time::sleep(Duration::from_millis(50)).await;
token.cancel();
tokio::time::sleep(Duration::from_millis(50)).await;
}1. Add Tokio: `cargo add tokio --features full` and set an async `main` with `#[tokio::main]`.
2. Pick the I/O primitive: `TcpListener`/`TcpStream` for sockets, `UnixListener` for local pipes, `tokio::fs` for async file access.
3. Handle each connection by `tokio::spawn` — never block tokio worker threads with synchronous work.
4. Share state via `tokio::sync::Mutex`/`RwLock` (or actor channels) cloned into tasks.
5. Add concurrency: use `tokio::join!`/`try_join!` for parallel awaits and `tokio::time::timeout` for deadlines.
6. Tune the runtime: set worker threads and run tests with `cargo test` under `--cfg tokio_unstable` if multi-thread stats are needed.
// Fan-out to N concurrent tasks and aggregate results
use tokio::time::{sleep, Duration};
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let handles: Vec<_> = (0..5)
.map(|i| tokio::spawn(async move {
sleep(Duration::from_millis(100 * i)).await;
i * i
}))
.collect();
let mut total = 0;
for h in handles {
total += h.await?;
}
println!("sum of squares = {total}");
Ok(())
}// Graceful shutdown with CancellationToken
use tokio_util::sync::CancellationToken;
#[tokio::main]
async fn main() {
let token = CancellationToken::new();
let child = token.clone();
tokio::spawn(async move {
tokio::select! {
_ = child.cancelled() => println!("task cancelled"),
_ = tokio::time::sleep(Duration::from_secs(60)) => println!("task done"),
}
});
tokio::time::sleep(Duration::from_millis(50)).await;
token.cancel();
tokio::time::sleep(Duration::from_millis(50)).await;
}Best Practices
- Use Tokio's async runtime for I/O-bound work; threads for CPU-heavy.
- Prefer `tokio::spawn` with owned tasks; avoid blocking the executor.
- Use channels for communication and `Arc<Mutex>` sparingly.
- Add timeouts with `tokio::time` to avoid hangs.
- Use `tokio::select!` for concurrent awaitable sources.
- Benchmark with `tokio` instrumented tracing when latency matters.
Troubleshooting
- Runtime not found: add `tokio` with the `full` or `rt-multi-thread` feature.
- Blocking executor: move sync work to `spawn_blocking`.
- Deadlocks: keep lock scopes tiny and avoid awaits while holding locks.
- Task panic: use JoinHandle or a supervisor to restart.
Validation
1. `cargo run` starts server without panics
2. Multiple concurrent connections are handled
3. Shared state synchronization is race-condition free
4. Performance benchmarks show expected throughput