Rust의 async/await는 제로 코스트 추상화입니다. 비동기 코드를 동기 코드처럼 작성할 수 있습니다.
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
// Future 트레이트의 정의
pub trait Future {
type Output;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}
// 간단한 Future 구현
struct TimerFuture {
completed: bool,
}
impl Future for TimerFuture {
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
if self.completed {
Poll::Ready(())
} else {
self.completed = true;
cx.waker().wake_by_ref();
Poll::Pending
}
}
}
// async 함수
async fn fetch_data() -> String {
// 비동기 작업 시뮬레이션
tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
"Data fetched".to_string()
}
// async 블록
async fn process() {
let result = async {
let data = fetch_data().await;
data.to_uppercase()
}.await;
println!("Result: {}", result);
}
// 여러 futures 동시 실행
async fn concurrent_tasks() {
let (result1, result2, result3) = tokio::join!(
fetch_data(),
fetch_data(),
fetch_data()
);
println!("{}, {}, {}", result1, result2, result3);
}
Tokio는 Rust의 사실상 표준 비동기 런타임입니다.
use tokio;
#[tokio::main]
async fn main() {
println!("Hello from Tokio!");
let handle = tokio::spawn(async {
println!("Spawned task");
42
});
let result = handle.await.unwrap();
println!("Result: {}", result);
}
// 멀티스레드 런타임
#[tokio::main(flavor = "multi_thread", worker_threads = 4)]
async fn main_multithread() {
// 4개의 워커 스레드로 실행
}
// 싱글스레드 런타임
#[tokio::main(flavor = "current_thread")]
async fn main_singlethread() {
// 현재 스레드에서만 실행
}
use tokio::time::{timeout, Duration};
async fn long_running_task() -> String {
tokio::time::sleep(Duration::from_secs(10)).await;
"Completed".to_string()
}
#[tokio::main]
async fn main() {
// 타임아웃 설정
match timeout(Duration::from_secs(2), long_running_task()).await {
Ok(result) => println!("Success: {}", result),
Err(_) => println!("Timeout!"),
}
// select! 매크로로 먼저 완료된 것 선택
tokio::select! {
_ = tokio::time::sleep(Duration::from_secs(1)) => {
println!("Timeout first");
}
result = fetch_data() => {
println!("Data first: {}", result);
}
}
}
스트림은 비동기 버전의 이터레이터입니다.
use tokio_stream::{Stream, StreamExt};
use std::pin::Pin;
// 스트림 생성
fn number_stream() -> impl Stream<Item = i32> {
tokio_stream::iter(1..=5)
}
// 스트림 처리
#[tokio::main]
async fn main() {
let mut stream = number_stream()
.map(|x| x * 2)
.filter(|x| x % 4 == 0);
while let Some(value) = stream.next().await {
println!("Value: {}", value);
}
// 비동기 생성
let stream = async_stream::stream! {
for i in 0..3 {
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
yield i;
}
};
tokio::pin!(stream);
while let Some(value) = stream.next().await {
println!("Generated: {}", value);
}
}
Pin은 메모리에서 값이 이동하지 못하게 고정합니다. 자체 참조 구조체에 필수적입니다.
use std::pin::Pin;
use std::marker::PhantomPinned;
// 자체 참조 구조체
struct SelfReferential {
value: String,
pointer: *const String, // value를 가리킴
_pin: PhantomPinned,
}
impl SelfReferential {
fn new(txt: &str) -> Pin<Box<Self>> {
let mut boxed = Box::pin(SelfReferential {
value: txt.to_string(),
pointer: std::ptr::null(),
_pin: PhantomPinned,
});
let ptr = &boxed.value as *const String;
unsafe {
let mut_ref = Pin::as_mut(&mut boxed);
Pin::get_unchecked_mut(mut_ref).pointer = ptr;
}
boxed
}
fn print(self: Pin<&Self>) {
let value = unsafe { &*self.pointer };
println!("Value: {}", value);
}
}
fn main() {
let pinned = SelfReferential::new("Hello");
pinned.as_ref().print();
}
use tokio::sync::mpsc;
#[tokio::main]
async fn main() {
let (tx, mut rx) = mpsc::channel(32);
// 여러 sender
for i in 0..5 {
let tx = tx.clone();
tokio::spawn(async move {
tx.send(format!("Message {}", i)).await.unwrap();
});
}
drop(tx); // 모든 sender를 닫음
// receiver에서 수신
while let Some(msg) = rx.recv().await {
println!("Received: {}", msg);
}
}
use tokio::sync::broadcast;
#[tokio::main]
async fn main() {
let (tx, mut rx1) = broadcast::channel(16);
let mut rx2 = tx.subscribe();
let mut rx3 = tx.subscribe();
tokio::spawn(async move {
for i in 0..3 {
tx.send(format!("Broadcast {}", i)).unwrap();
}
});
// 모든 receiver가 메시지 수신
tokio::spawn(async move {
while let Ok(msg) = rx1.recv().await {
println!("RX1: {}", msg);
}
});
tokio::spawn(async move {
while let Ok(msg) = rx2.recv().await {
println!("RX2: {}", msg);
}
});
while let Ok(msg) = rx3.recv().await {
println!("RX3: {}", msg);
}
}
use tokio;
use reqwest;
use std::collections::HashSet;
use std::sync::{Arc, Mutex};
struct Crawler {
visited: Arc<Mutex<HashSet<String>>>,
client: reqwest::Client,
}
impl Crawler {
fn new() -> Self {
Crawler {
visited: Arc::new(Mutex::new(HashSet::new())),
client: reqwest::Client::new(),
}
}
async fn fetch(&self, url: &str) -> Result<String, reqwest::Error> {
let response = self.client.get(url).send().await?;
response.text().await
}
async fn crawl(&self, url: String, depth: usize) {
if depth == 0 {
return;
}
// 이미 방문했는지 확인
{
let mut visited = self.visited.lock().unwrap();
if visited.contains(&url) {
return;
}
visited.insert(url.clone());
}
println!("Crawling: {}", url);
// 페이지 가져오기
match self.fetch(&url).await {
Ok(content) => {
println!("Fetched {} bytes from {}", content.len(), url);
// 여기서 링크를 추출하고 재귀적으로 크롤링
// (간단한 예제를 위해 생략)
}
Err(e) => {
println!("Error fetching {}: {}", url, e);
}
}
}
async fn crawl_concurrent(&self, urls: Vec<String>, depth: usize) {
let mut tasks = vec![];
for url in urls {
let crawler = Crawler {
visited: Arc::clone(&self.visited),
client: self.client.clone(),
};
let task = tokio::spawn(async move {
crawler.crawl(url, depth).await;
});
tasks.push(task);
}
// 모든 태스크 완료 대기
for task in tasks {
let _ = task.await;
}
}
}
#[tokio::main]
async fn main() {
let crawler = Crawler::new();
let urls = vec![
"https://example.com".to_string(),
"https://rust-lang.org".to_string(),
"https://github.com".to_string(),
];
crawler.crawl_concurrent(urls, 2).await;
println!("Crawling complete!");
let visited = crawler.visited.lock().unwrap();
println!("Visited {} pages", visited.len());
}