嘿，小伙伴们！今天咱们来聊聊怎么用 Rust 搭建一个牛气哄哄的多线程服务器，还能在需要的时候优雅地关机。为啥要用 Rust 呢？因为 Rust 是个超级靠谱的语言，它能保证内存安全，写并发代码的时候不用担心那些让人头疼的并发问题，比如数据竞争和死锁。而且，Rust 的标准库提供了超好用的网络编程工具，能帮咱们轻松搞定网络服务。
一、先搞个线程池
线程池这东西，就像是个“任务分配中心”，能帮咱们把任务合理分配给线程，避免线程频繁创建和销毁，节省资源，提升性能。想象一下，如果你每次有任务就新建一个线程，那线程创建和销毁的开销会让你的程序慢得像蜗牛。有了线程池，这些线程就像是一群随时待命的工人，任务一来，立马就能开工。
下面就是个简单的线程池代码，瞅瞅：
use std::sync::{mpsc, Arc, Mutex};
use std::thread;
pub struct ThreadPool {
workers: Vec<Worker>,
sender: Option<mpsc::Sender<Job>>,
}
type Job = Box<dyn FnOnce() + Send + 'static>;
impl ThreadPool {
pub 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: Some(sender),
}
}
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
let job = Box::new(f);
self.sender.as_ref().unwrap().send(job).unwrap();
}
}
impl Drop for ThreadPool {
fn drop(&mut self) {
drop(self.sender.take());
for worker in &mut self.workers {
println!("Shutting down worker {}", worker.id);
if let Some(thread) = worker.thread.take() {
thread.join().unwrap();
}
}
}
}
struct Worker {
id: usize,
thread: Option<thread::JoinHandle<()>>,
}
impl Worker {
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
let thread = thread::spawn(move || loop {
let message = receiver.lock().unwrap().recv();
match message {
Ok(job) => {
println!("Worker {id} got a job; executing.");
job();
}
Err(_) => {
println!("Worker {id} disconnected; shutting down.");
break;
}
}
});
Worker {
id,
thread: Some(thread),
}
}
}
重点来了
- 创建线程：用
mpsc::channel搭个通道，让主线程和工作线程能通信。工作线程就等着通道里有任务过来，接收到就执行。这就像是有个传送带，任务一过来，工人（线程）就拿起来干活。 - 分配任务：
execute方法把任务扔进通道，工作线程就能拿到任务干活了。这就像是你把任务扔到传送带上，工人会自己去拿。 - 线程管理：服务器要关机的时候，
Drop实现能保证所有工作线程都乖乖退出，不会留下“孤儿线程”。这就像是下班的时候，你得确保所有工人都安全离开，不会有人被锁在工厂里。
二、监听网络连接
Rust 里监听网络连接有俩常用方法：listener.incoming() 和 listener.accept()。这俩有啥区别呢？想象一下，你开了一家餐厅，listener.incoming() 就像是门口的迎宾，一直站在那里迎接客人，每次有客人进来就通知后厨；而 listener.accept() 就像是你每次只接待一个客人，等这个客人坐下后，再去门口看看有没有下一个客人。
listener.incoming()
这哥们儿是个迭代器，会一直监听新的连接，每次迭代返回一个 Result<TcpStream, std::io::Error>。用起来就像这样：
for stream in listener.incoming() {
let stream = stream.unwrap();
pool.execute(|| {
handle_connection(stream);
});
}
listener.accept()
这哥们儿每次调用会阻塞当前线程，直到有新的连接过来，返回一个 Result<(TcpStream, SocketAddr), std::io::Error>。用起来是这样：
loop {
match listener.accept() {
Ok((stream, _)) => {
pool.execute(|| {
handle_connection(stream);
});
}
Err(e) => {
eprintln!("Failed to accept connection: {:?}", e);
}
}
}
选谁好呢？
listener.incoming()：适合需要持续监听多个连接的情况，用在循环里很顺手。这就像是餐厅门口的迎宾，能同时处理多个客人的到来。listener.accept()：适合每次只处理一个连接的情况，用在同步逻辑里很方便。这就像是你每次只接待一个客人，处理完再去接待下一个。
三、优雅关机
服务器跑着跑着，说不定会收到外部信号，比如 Ctrl+C 或 SIGTERM，这时候咱得让它能优雅地关机，而不是直接“死机”。这咋整呢？用 signal-hook 库来捕获信号呗！这就像是你给服务器设置了一个“紧急停止”按钮，按下按钮的时候，服务器能安全地关闭，而不是直接断电。
use signal_hook::iterator::Signals;
use std::sync::{Arc, atomic::{AtomicBool, Ordering}};
fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
let pool = ThreadPool::new(4);
// 搞个原子布尔变量，控制服务器要不要关机
let running = Arc::new(AtomicBool::new(true));
let r = running.clone();
// 捕获信号
let signals = Signals::new(&[signal_hook::SIGINT, signal_hook::SIGTERM]).unwrap();
thread::spawn(move || {
for _ in signals.forever() {
println!("Received termination signal, shutting down...");
r.store(false, Ordering::SeqCst);
}
});
// 主循环
for stream in listener.incoming() {
if !running.load(Ordering::SeqCst) {
println!("Shutting down.");
break;
}
let stream = stream.unwrap();
pool.execute(|| {
handle_connection(stream);
});
}
println!("Shutting down.");
}
重点来了
- 捕获信号：用
signal-hook捕获SIGINT和SIGTERM信号。这就像是给服务器装了个“耳朵”，能听到外部的“关机”信号。 - 原子布尔变量：用
AtomicBool来控制服务器要不要继续跑。这就像是服务器的“心跳”，一旦“心跳”停止，服务器就知道该关机了。 - 主循环检查：每次迭代都瞅瞅
running的值，要是false，就退出循环关机。这就像是你每次接待客人的时候，都会看看“紧急停止”按钮有没有被按下。
四、处理连接
服务器接收到连接后，得处理啊。下面是个简单的 handle_connection 函数，处理 HTTP 请求并返回响应：
fn handle_connection(mut stream: TcpStream) {
let mut buffer = [0; 1024];
stream.read(&mut buffer).unwrap();
let response = "HTTP/1.1 200 OK\r\n\r\nHello, world!";
stream.write(response.as_bytes()).unwrap();
stream.flush().unwrap();
}
重点来了
- 读取请求：从
TcpStream里把客户端发的数据读出来。这就像是你从客人那里拿到订单。 - 构造响应：根据请求整一个简单的 HTTP 响应。这就像是你根据订单准备食物。
- 发送响应：把响应发回客户端。这就像是你把准备好的食物送到客人桌上。
五、总结
好啦，今天咱们用 Rust 搭了个多线程服务器，还学会了怎么让它能优雅地关机。关键点都给你唠清楚了：
- 线程池：用
mpsc::channel和Arc<Mutex<...>>搭个线程池，任务分配妥妥的。这就像是你有了一个高效的“任务分配中心”。 - 网络监听：用
listener.incoming()或listener.accept()监听网络连接，看场景选合适的。这就像是你选择是让迎宾接待客人，还是自己一个个接待。 - 信号处理：用
signal-hook捕获信号，AtomicBool控制服务器运行状态。这就像是你给服务器装了个“紧急停止”按钮。 - 连接处理：在工作线程里处理连接，读请求、整响应、发回去。这就像是你在餐厅里接待客人，处理订单，送食物。
有了这些知识，你就能搭一个高性能、靠谱的网络服务啦，还能在收到终止信号的时候优雅地关机。赶紧试试吧，有问题随时问我哦！
