用 Rust 搭建一个优雅的多线程服务器：从零开始的详细指南

嘿，小伙伴们！今天咱们来聊聊怎么用 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 控制服务器运行状态。这就像是你给服务器装了个“紧急停止”按钮。
• 连接处理：在工作线程里处理连接，读请求、整响应、发回去。这就像是你在餐厅里接待客人，处理订单，送食物。

有了这些知识，你就能搭一个高性能、靠谱的网络服务啦，还能在收到终止信号的时候优雅地关机。赶紧试试吧，有问题随时问我哦！