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

构建高并发应用的基石：Rust无锁队列（从零开始掌握Rust无锁队列实现与并发编程）

主机测评网
Rust
2025-12-21
725

在现代多线程编程中，Rust无锁队列 是提升程序并发性能的关键数据结构。它避免了传统锁机制带来的上下文切换开销和死锁风险，特别适用于高频读写场景。本文将带你从基础概念出发，逐步实现一个简易但功能完整的无锁队列，即使你是 Rust 新手，也能轻松上手！

什么是无锁队列？

无锁（Lock-Free）队列是一种不依赖互斥锁（Mutex）来保证线程安全的数据结构。它通过原子操作（如 compare-and-swap, CAS）实现多个线程对共享资源的安全访问。相比加锁方式，无锁结构能显著减少线程阻塞，提高系统吞吐量。

构建高并发应用的基石：Rust无锁队列（从零开始掌握Rust无锁队列实现与并发编程） Rust无锁队列并发编程 Rust内存安全高性能数据结构第1张

为什么选择 Rust 实现无锁队列？

Rust 语言凭借其内存安全和零成本抽象的特性，成为实现高性能并发结构的理想选择。其所有权系统和借用检查器能在编译期防止数据竞争，而标准库提供的 std::sync::atomic 模块则为无锁编程提供了坚实基础。

动手实现：一个简单的无锁队列

我们将使用单生产者-单消费者（SPSC）模型来简化逻辑。这种模型常见于消息传递、任务调度等场景。

首先，添加必要依赖（无需外部 crate，仅用标准库）：

use std::ptr;use std::sync::atomic::{AtomicPtr, Ordering};use std::sync::Arc;

定义节点结构：

struct Node<T> {    data: T,    next: AtomicPtr<Node<T>>,}impl<T> Node<T> {    fn new(data: T) -> *mut Node<T> {        Box::into_raw(Box::new(Node {            data,            next: AtomicPtr::new(ptr::null_mut()),        }))    }}

定义无锁队列主体：

pub struct LockFreeQueue<T> {    head: AtomicPtr<Node<T>>,    tail: AtomicPtr<Node<T>>,}impl<T> LockFreeQueue<T> {    pub fn new() -> Self {        let dummy = Node::new(unsafe { std::mem::zeroed() });        Self {            head: AtomicPtr::new(dummy),            tail: AtomicPtr::new(dummy),        }    }    pub fn push(&self, data: T) {        let new_node = Node::new(data);        loop {            let tail = self.tail.load(Ordering::Acquire);            let next = unsafe { (*tail).next.load(Ordering::Acquire) };            if next.is_null() {                // 尝试链接新节点                if unsafe { (*tail).next.compare_exchange(                    ptr::null_mut(),                    new_node,                    Ordering::Release,                    Ordering::Relaxed,                ).is_ok() } {                    // 更新 tail 指针                    self.tail.compare_exchange(                        tail,                        new_node,                        Ordering::Release,                        Ordering::Relaxed,                    );                    break;                }            } else {                // 帮助其他线程推进 tail                self.tail.compare_exchange(                    tail,                    next,                    Ordering::Release,                    Ordering::Relaxed,                );            }        }    }    pub fn pop(&self) -> Option {        loop {            let head = self.head.load(Ordering::Acquire);            let tail = self.tail.load(Ordering::Acquire);            let next = unsafe { (*head).next.load(Ordering::Acquire) };            if head == tail {                if next.is_null() {                    return None; // 队列为空                }                // 帮助推进 tail                self.tail.compare_exchange(                    tail,                    next,                    Ordering::Release,                    Ordering::Relaxed,                );            } else {                if !next.is_null() {                    // 尝试弹出节点                    let data = unsafe { (*next).data };                    if self.head.compare_exchange(                        head,                        next,                        Ordering::Release,                        Ordering::Relaxed,                    ).is_ok() {                        unsafe {                            drop(Box::from_raw(head));                        }                        return Some(data);                    }                }            }        }    }}impl<T> Drop for LockFreeQueue<T> {    fn drop(&mut self) {        while self.pop().is_some() {}        unsafe {            drop(Box::from_raw(self.head.load(Ordering::Relaxed)));        }    }}

测试我们的无锁队列

下面是一个简单的多线程测试示例，展示如何在生产者-消费者模型中使用该队列：

use std::thread;fn main() {    let queue = Arc::new(LockFreeQueue::new());    let queue_clone = Arc::clone(&queue);    let producer = thread::spawn(move || {        for i in 0..10 {            queue_clone.push(i);            println!("Produced: {}", i);        }    });    let consumer = thread::spawn(move || {        while let Some(val) = queue.pop() {            println!("Consumed: {}", val);        }    });    producer.join().unwrap();    consumer.join().unwrap();}