高频策略的Rust内存优化：对象池与零分配设计模式

1
2
3
4
5

// 每次处理一个tick都创建新对象
for tick in ticks {
    let order = Order::new(tick.price, tick.volume);
    process_order(&order);
}  // order在这里被销毁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

use std::sync::Mutex;
use std::ops::{Deref, DerefMut};

pub struct ObjectPool<T> {
    pool: Mutex<Vec<T>>,
    factory: fn() -> T,
}

impl<T: Send> ObjectPool<T> { 
    pub fn new(capacity: usize, factory: fn() -> T) -> Self {
        let mut pool = Vec::with_capacity(capacity);
        for _ in 0..capacity {
            pool.push(factory());
        }
        Self {
            pool: Mutex::new(pool),
            factory,
        }
    }

    // 修复生命周期标注
    pub fn acquire(&self) -> PooledObject<'_, T> {
        let mut pool = self.pool.lock().expect("Mutex poisoned");
        let obj = pool.pop().unwrap_or_else(|| (self.factory)());
        PooledObject { obj: Some(obj), pool: self }
    }

    pub fn release(&self, obj: T) {
        let mut pool = self.pool.lock().expect("Mutex poisoned");
        pool.push(obj);
    }
}

pub struct PooledObject<'a, T: Send> {
    obj: Option<T>,
    pool: &'a ObjectPool<T>,
}

// 实现 Deref 使其可用
impl<'a, T: Send> Deref for PooledObject<'a, T> {
    type Target = T;
    fn deref(&self) -> &Self::Target {
        self.obj.as_ref().unwrap()
    }
}

impl<'a, T: Send> DerefMut for PooledObject<'a, T> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        self.obj.as_mut().unwrap()
    }
}

impl<'a, T: Send> Drop for PooledObject<'a, T> {
    fn drop(&mut self) {
        if let Some(obj) = self.obj.take() {
            self.pool.release(obj);
        }
    }
}

1
2
3
4
5

// 每个对象单独分配
let mut orders = Vec::new();
for _ in 0..1000 {
    orders.push(Box::new(Order::default()));
}

1
2
3
4
5
6
7
8
9

use bumpalo::Bump;
 
let arena = Bump::new();
 
// 所有对象在Arena中连续分配
let orders: Vec<&mut Order> = (0..1000)
    .map(|_| arena.alloc(Order::default()))
    .collect();
 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

pub struct MarketDataProcessor {
    // 预分配缓冲区
    buffer: Vec<u8>,
    orders: Vec<Order>,
}

impl MarketDataProcessor {
    pub fn new() -> Self {
        Self {
            buffer: vec![0u8; 1024 * 1024],  // 1MB缓冲区
            orders: Vec::with_capacity(1000),
        }
    }

    pub fn process(&mut self, data: &[u8]) {
        
        if data.len() > self.buffer.capacity() {
        // 记录警告或处理异常，而不是让 Vec 自动扩容导致丢帧
        eprintln!("Warning: Data size exceeds buffer capacity!");
    }
        self.buffer.extend_from_slice(data);

        // 重置订单列表，不重新分配
        self.orders.clear();
    }
}