Rust线程池管理库deadpool-sync的使用，deadpool-sync提供高效同步连接池解决方案

Deadpool是一个用于任何类型连接和对象的极其简单的异步池。

这个crate提供了用于为不支持异步且需要在线程内运行的对象编写池的辅助工具。

注意： 这个crate旨在使deadpool-* crate的开发更容易。其他库和二进制项目通常不应直接使用它，而应使用使用它的crate提供的重新导出。

特性

许可证

根据以下任一许可证授权

• Apache License, Version 2.0
• MIT license

由您选择。

完整示例代码

use deadpool_sync::SyncWrapper;
use std::thread;
use std::time::Duration;

// 创建一个同步对象包装器
let wrapper = SyncWrapper::new(|| {
    // 这里创建或初始化您的同步对象
    // 例如：数据库连接、HTTP客户端等
    let connection = SomeExpensiveSyncResource::new();
    connection
});

// 在线程池中使用同步对象
let handle = thread::spawn(move || {
    // 获取同步对象
    let mut guard = wrapper.get().unwrap();
    
    // 使用同步对象执行操作
    guard.do_something_sync();
    
    // 执行一些耗时操作
    thread::sleep(Duration::from_secs(1));
});

// 等待线程完成
handle.join().unwrap();

use deadpool_sync::Pool;
use std::sync::Arc;
use std::thread;

// 创建连接池配置
let pool = Pool::builder()
    .max_size(10) // 最大连接数
    .build(|| {
        // 连接工厂函数
        Arc::new(ExpensiveSyncConnection::new())
    })
    .unwrap();

// 在多线程环境中使用连接池
let handles: Vec<_> = (0..5)
    .map(|i| {
        let pool = pool.clone();
        thread::spawn(move || {
            // 从池中获取连接
            let connection = pool.get().unwrap();
            
            // 使用连接执行操作
            println!("Thread {} using connection: {:?}", i, connection);
            
            // 连接会在离开作用域时自动返回池中
        })
    })
    .collect();

// 等待所有线程完成
for handle in handles {
    handle.join().unwrap();
}

use deadpool_sync::SyncWrapper;
use std::sync::{Arc, Mutex};
use std::thread;

// 创建线程安全的计数器示例
let counter = Arc::new(SyncWrapper::new(|| Mutex::new(0)));

let mut handles = vec![];

for _ in 0..10 {
    let counter = Arc::clone(&counter);
    let handle = thread::spawn(move || {
        let mut guard = counter.get().unwrap();
        let mut num = guard.lock().unwrap();
        *num += 1;
    });
    handles.push(handle);
}

for handle in handles {
    handle.join().unwrap();
}

// 验证结果
let result = counter.get().unwrap().lock().unwrap();
println!("Final counter value: {}", *result);

完整示例demo

use deadpool_sync::Pool;
use std::sync::Arc;
use std::thread;
use std::time::Duration;

// 定义一个模拟的昂贵同步连接
struct ExpensiveConnection {
    id: u32,
}

impl ExpensiveConnection {
    fn new(id: u32) -> Self {
        println!("Creating new connection {}", id);
        // 模拟创建连接的耗时操作
        thread::sleep(Duration::from_millis(100));
        ExpensiveConnection { id }
    }

    fn execute_query(&self, query: &str) {
        println!("Connection {} executing query: {}", self.id, query);
        // 模拟查询执行时间
        thread::sleep(Duration::from_millis(50));
    }
}

fn main() {
    // 创建连接池，最大容量为5
    let pool = Pool::builder()
        .max_size(5)
        .build(|| {
            static COUNTER: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(1);
            let id = COUNTER.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            Arc::new(ExpensiveConnection::new(id))
        })
        .unwrap();

    println!("连接池创建完成，开始模拟多线程查询...");

    let mut handles = vec![];

    // 创建10个线程并发执行查询
    for i in 0..10 {
        let pool = pool.clone();
        let handle = thread::spawn(move || {
            // 从连接池获取连接
            let connection = pool.get().unwrap();
            
            // 执行查询操作
            connection.execute_query(&format!("SELECT * FROM table WHERE id = {}", i));
            
            // 连接会在作用域结束时自动返回池中
            println!("Thread {} finished using connection {}", i, connection.id);
        });
        handles.push(handle);
    }

    // 等待所有线程完成
    for handle in handles {
        handle.join().unwrap();
    }

    println!("所有查询执行完成！");
    println!("当前池中连接数: {}", pool.status().size);
    println!("可用连接数: {}", pool.status().available);
}

use deadpool_sync::SyncWrapper;
use std::sync::{Arc, Mutex};
use std::thread;

// 使用SyncWrapper包装线程安全的数据结构示例
struct SharedData {
    counter: i32,
    data: Vec<String>,
}

impl SharedData {
    fn new() -> Self {
        SharedData {
            counter: 0,
            data: Vec::new(),
        }
    }

    fn increment(&mut self) {
        self.counter += 1;
    }

    fn add_data(&mut self, item: String) {
        self.data.push(item);
    }

    fn get_stats(&self) -> (i32, usize) {
        (self.counter, self.data.len())
    }
}

fn main() {
    // 使用SyncWrapper包装共享数据
    let shared_data = Arc::new(SyncWrapper::new(|| Mutex::new(SharedData::new())));

    let mut handles = vec![];

    // 创建多个线程并发修改共享数据
    for i in 0..8 {
        let data_ref = Arc::clone(&shared_data);
        let handle = thread::spawn(move || {
            // 获取数据的可变访问权
            let mut guard = data_ref.get().unwrap();
            let mut data = guard.lock().unwrap();
            
            // 修改数据
            data.increment();
            data.add_data(format!("Item from thread {}", i));
            
            println!("Thread {} modified shared data", i);
        });
        handles.push(handle);
    }

    // 等待所有线程完成
    for handle in handles {
        handle.join().unwrap();
    }

    // 读取最终结果
    let guard = shared_data.get().unwrap();
    let data = guard.lock().unwrap();
    let (counter, items_count) = data.get_stats();
    
    println!("最终计数器值: {}", counter);
    println!("数据项数量: {}", items_count);
    println!("所有数据: {:?}", data.data);
}