Linux多线程编程：从基础到高级实践

1. 线程等待：理解线程生命周期管理

在Linux多线程编程中，线程等待是一个基础但至关重要的概念。我们先从一个简单的例子开始：

cpp复制#include <pthread.h>
#include <iostream>

void* threadFunction(void* arg) {
    const char* name = static_cast<const char*>(arg);
    for(int i=0; i<3; ++i) {
        std::cout << name << " is running..." << std::endl;
        sleep(1);
    }
    return (void*)42; // 返回一个特殊值
}

int main() {
    pthread_t threadId;
    pthread_create(&threadId, nullptr, threadFunction, (void*)"Worker");

    void* threadResult;
    int joinResult = pthread_join(threadId, &threadResult);
    
    if(joinResult == 0) {
        std::cout << "Thread returned: " << (long)threadResult << std::endl;
    } else {
        perror("pthread_join failed");
    }
    
    return 0;
}

1.1 pthread_join的深层机制

pthread_join实际上执行了三个关键操作：

1. 阻塞调用线程，直到目标线程终止
2. 回收目标线程的资源（特别是线程描述符）
3. 获取目标线程的返回值

重要提示：未join的线程会导致资源泄漏，类似于僵尸进程。但不同于进程，Linux内核不提供查看"僵尸线程"的命令，因为线程资源主要在用户空间的pthread库中管理。

1.2 线程终止的三种方式

1. 自然终止：线程函数执行return语句
2. 显式终止：调用pthread_exit()
3. 强制终止：其他线程调用pthread_cancel()

cpp复制// 强制终止线程示例
pthread_cancel(threadId);
void* result;
pthread_join(threadId, &result);
if(result == PTHREAD_CANCELED) {
    std::cout << "Thread was canceled" << std::endl;
}

1.3 线程返回值处理技巧

线程可以返回任何类型的指针值，但需要注意：

• 不要返回指向栈内存的指针（函数退出后失效）
• 返回动态分配的内存时要确保接收方负责释放
• 可以使用类型安全的C++方式封装返回值

cpp复制// 安全的返回值处理示例
struct ThreadResult {
    int status;
    std::string message;
};

void* worker(void*) {
    auto result = new ThreadResult{0, "Success"};
    return result;
}

int main() {
    pthread_t tid;
    pthread_create(&tid, nullptr, worker, nullptr);
    
    ThreadResult* result;
    pthread_join(tid, (void**)&result);
    std::cout << result->message << std::endl;
    delete result;
}

2. 分离线程：自动资源回收机制

2.1 分离线程的核心概念

分离线程(detached thread)是指不需要其他线程调用pthread_join来回收资源的线程。线程终止时，系统会自动回收其资源。

cpp复制// 创建即分离的线程
pthread_attr_t attr;
pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
pthread_t tid;
pthread_create(&tid, &attr, threadFunc, nullptr);
pthread_attr_destroy(&attr);

2.2 分离线程的适用场景

分离线程特别适合以下情况：

• 不需要获取线程返回值
• 线程生命周期独立于创建者
• 执行后台任务的"守护线程"

实践建议：对于长期运行的服务程序，优先考虑使用分离线程，避免因忘记join导致的资源泄漏。

2.3 分离线程的注意事项

1. 不可逆性：一旦线程被分离，不能再变为joinable
2. 错误处理：对分离线程调用pthread_join会返回EINVAL错误
3. 资源回收时机：分离线程的资源回收时机不确定，可能不会立即进行

cpp复制// 动态分离示例
void* worker(void* arg) {
    pthread_detach(pthread_self()); // 自行分离
    // ...工作代码...
    return nullptr;
}

3. 线程ID与进程地址空间深度解析

3.1 线程ID的本质

在Linux的pthread实现中，线程ID实际上是指向线程控制块(TCB)的指针。TCB包含：

• 线程状态
• 栈指针
• 寄存器保存区
• 返回值存储区
• 其他线程属性

cpp复制// 获取线程ID的两种方式对比
std::cout << "POSIX ID: " << pthread_self() << std::endl;
std::cout << "System ID: " << syscall(SYS_gettid) << std::endl;

3.2 进程地址空间布局

典型的Linux进程地址空间布局（从低地址到高地址）：

1. 代码段(.text)
2. 数据段(.data, .bss)
3. 堆(heap)
4. 共享库映射区
5. 栈(stack)
6. 内核空间

线程会共享进程的大部分地址空间，但每个线程有自己的：

• 栈空间（通过mmap分配）
• 线程局部存储(TLS)
• 寄存器状态

3.3 线程局部存储(TLS)详解

TLS允许每个线程拥有变量的独立副本：

cpp复制// TLS使用示例
__thread int threadSpecificVar = 0;

void* worker(void* arg) {
    threadSpecificVar = reinterpret_cast<long>(arg);
    std::cout << "Thread var: " << threadSpecificVar << std::endl;
    return nullptr;
}

TLS的实现机制：

1. 编译器为TLS变量生成特殊的访问指令
2. 线程创建时为每个TLS变量分配独立存储
3. 通过FS/GS段寄存器实现快速访问

4. C++多线程实战与跨平台考量

4.1 std::thread基础用法

cpp复制#include <thread>
#include <iostream>

void worker(int id) {
    std::cout << "Worker " << id << " started" << std::endl;
    // ...工作代码...
}

int main() {
    std::thread t1(worker, 1);
    std::thread t2(worker, 2);
    
    t1.join();
    t2.join();
    return 0;
}

4.2 C++线程与POSIX线程的关系

C++标准库的线程实现通常基于平台原生API：

• Linux下：封装pthread
• Windows下：封装Win32线程API
• macOS下：封装pthread或GCD

4.3 跨平台线程编程最佳实践

1. 优先使用标准库：std::thread, std::mutex等
2. 平台特定功能：通过条件编译处理
3. 性能关键部分：考虑平台优化

cpp复制// 条件编译示例
#ifdef __linux__
    // Linux特定优化
#elif defined(_WIN32)
    // Windows特定代码
#endif

4.4 现代C++多线程特性

C++11后引入的重要多线程组件：

1. <thread>：线程管理
2. <mutex>：互斥锁
3. <atomic>：原子操作
4. <future>：异步结果
5. <condition_variable>：线程同步

cpp复制// 使用async的异步编程示例
auto future = std::async(std::launch::async, []{
    std::this_thread::sleep_for(1s);
    return 42;
});
std::cout << "Result: " << future.get() << std::endl;

5. 多线程编程常见陷阱与调试技巧

5.1 线程安全问题排查

常见线程安全问题：

1. 数据竞争
2. 死锁
3. 活锁
4. 优先级反转

调试工具：

• Valgrind Helgrind
• ThreadSanitizer (TSan)
• gdb多线程调试命令

bash复制# 使用TSan编译和运行
g++ -fsanitize=thread -g program.cpp -o program
./program

5.2 性能优化策略

1. 减少锁竞争：

   • 使用读写锁(std::shared_mutex)
   • 减小临界区范围
   • 考虑无锁数据结构

2. 线程池模式：

   • 避免频繁创建销毁线程
   • 合理设置线程数量(通常=CPU核心数)

cpp复制// 简单线程池示例
std::vector<std::thread> pool;
for(int i=0; i<std::thread::hardware_concurrency(); ++i) {
    pool.emplace_back(worker);
}

5.3 资源管理最佳实践

1. RAII包装线程：

cpp复制class ThreadGuard {
    std::thread t;
public:
    explicit ThreadGuard(std::thread t_) : t(std::move(t_)) {
        if(!t.joinable()) throw std::logic_error("No thread");
    }
    ~ThreadGuard() { if(t.joinable()) t.join(); }
    // 禁止拷贝
};

2. 异常安全处理：

• 确保线程退出时释放资源
• 使用shared_ptr管理共享资源

6. 深入理解Linux线程模型

6.1 1:1线程模型详解

Linux采用1:1线程模型，即：

• 每个用户态线程对应一个内核调度实体(LWP)
• 线程调度由内核完成
• 创建/销毁线程涉及系统调用

优势：

• 调度效率高
• 多核利用充分
• 实现相对简单

劣势：

• 创建大量线程时系统开销大
• 某些操作(如信号处理)较复杂

6.2 线程与信号的关系

信号处理注意事项：

1. 信号是发送到进程的，但由任意线程处理
2. 使用pthread_sigmask设置线程信号掩码
3. 专门的信号处理线程是常见模式

cpp复制// 设置信号处理线程
void* signalHandler(void*) {
    sigset_t set;
    sigfillset(&set);
    while(true) {
        int sig;
        sigwait(&set, &sig);
        // 处理信号
    }
}

int main() {
    // 屏蔽所有信号
    sigset_t set;
    sigfillset(&set);
    pthread_sigmask(SIG_BLOCK, &set, nullptr);
    
    // 创建信号处理线程
    std::thread handler(signalHandler);
    // ...
}

6.3 线程与进程资源的对比

共享资源：

• 地址空间
• 文件描述符表
• 信号处理
• 用户/组ID

独立资源：

• 线程ID
• 寄存器状态
• 栈空间
• 信号掩码
• 线程特定数据

7. 高级线程控制技术

7.1 线程属性精细控制

cpp复制// 设置线程栈大小示例
pthread_attr_t attr;
pthread_attr_init(&attr);
pthread_attr_setstacksize(&attr, 2*1024*1024); // 2MB栈

pthread_t tid;
pthread_create(&tid, &attr, threadFunc, nullptr);
pthread_attr_destroy(&attr);

其他可配置属性：

• 调度策略(SCHED_FIFO, SCHED_RR, SCHED_OTHER)
• 调度优先级
• 栈地址(自定义栈位置)
• 继承属性

7.2 线程取消与清理

安全的线程取消需要设置取消点和清理函数：

cpp复制void cleanup(void* arg) {
    std::cout << "Cleaning up: " << (const char*)arg << std::endl;
}

void* worker(void*) {
    pthread_cleanup_push(cleanup, (void*)"ResourceA");
    // ...工作代码...
    pthread_testcancel(); // 显式取消点
    pthread_cleanup_pop(1);
    return nullptr;
}

7.3 线程绑定CPU核心

提高缓存亲和性的技术：

cpp复制#ifdef __linux__
#include <sched.h>
void bindToCore(int coreId) {
    cpu_set_t cpuset;
    CPU_ZERO(&cpuset);
    CPU_SET(coreId, &cpuset);
    pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
}
#endif

8. 多线程设计模式与实践

8.1 生产者-消费者模式

cpp复制#include <queue>
#include <mutex>
#include <condition_variable>

template<typename T>
class SafeQueue {
    std::queue<T> queue;
    std::mutex mtx;
    std::condition_variable cv;
public:
    void push(T item) {
        std::lock_guard<std::mutex> lock(mtx);
        queue.push(std::move(item));
        cv.notify_one();
    }
    
    T pop() {
        std::unique_lock<std::mutex> lock(mtx);
        cv.wait(lock, [this]{ return !queue.empty(); });
        T item = std::move(queue.front());
        queue.pop();
        return item;
    }
};

8.2 线程池高级实现

cpp复制class ThreadPool {
    std::vector<std::thread> workers;
    SafeQueue<std::function<void()>> tasks;
    std::atomic<bool> stop{false};
    
    void workerThread() {
        while(!stop) {
            auto task = tasks.pop();
            if(task) task();
        }
    }
    
public:
    ThreadPool(size_t threads) {
        for(size_t i=0; i<threads; ++i) {
            workers.emplace_back([this]{ workerThread(); });
        }
    }
    
    ~ThreadPool() {
        stop = true;
        for(auto& worker : workers) {
            if(worker.joinable()) worker.join();
        }
    }
    
    template<class F>
    void enqueue(F&& f) {
        tasks.push(std::forward<F>(f));
    }
};

8.3 无锁编程基础

cpp复制#include <atomic>

class LockFreeStack {
    struct Node {
        int value;
        Node* next;
    };
    
    std::atomic<Node*> head{nullptr};
    
public:
    void push(int value) {
        Node* newNode = new Node{value, nullptr};
        newNode->next = head.load();
        while(!head.compare_exchange_weak(newNode->next, newNode)) {
            // CAS失败，重试
        }
    }
    
    bool pop(int& value) {
        Node* oldHead = head.load();
        while(oldHead && 
              !head.compare_exchange_weak(oldHead, oldHead->next)) {
            // CAS失败，重试
        }
        if(!oldHead) return false;
        value = oldHead->value;
        delete oldHead;
        return true;
    }
};

9. 性能分析与调优实战

9.1 多线程性能指标

关键性能指标：

1. 吞吐量(Throughput)
2. 延迟(Latency)
3. 可扩展性(Scalability)
4. 资源利用率

9.2 锁竞争分析工具

1. perf工具：

bash复制perf record -g -p <pid>
perf report

2. Lockstat：

bash复制echo 1 > /proc/sys/kernel/lock_stat
# 运行程序
echo 0 > /proc/sys/kernel/lock_stat
dmesg | less

9.3 缓存友好设计

提高缓存命中率的技巧：

1. 数据对齐
2. 伪共享(false sharing)避免
3. 数据局部性优化

cpp复制// 避免伪共享的填充技术
struct AlignedCounter {
    std::atomic<long> count;
    char padding[64 - sizeof(std::atomic<long>)];
};

10. 现代C++并发新特性

10.1 C++17并行算法

cpp复制#include <algorithm>
#include <execution>
#include <vector>

int main() {
    std::vector<int> data(1000000);
    std::sort(std::execution::par, data.begin(), data.end());
    return 0;
}

10.2 C++20协程基础

cpp复制#include <coroutine>
#include <iostream>

Generator<int> range(int start, int end) {
    for(int i=start; i<end; ++i)
        co_yield i;
}

int main() {
    for(int i : range(1, 10)) {
        std::cout << i << std::endl;
    }
    return 0;
}

10.3 原子操作内存模型

cpp复制std::atomic<int> counter{0};

void increment() {
    for(int i=0; i<1000; ++i) {
        counter.fetch_add(1, std::memory_order_relaxed);
    }
}

int main() {
    std::thread t1(increment);
    std::thread t2(increment);
    t1.join();
    t2.join();
    std::cout << counter.load() << std::endl;
    return 0;
}

11. 多线程调试高级技巧

11.1 gdb多线程调试

常用命令：

• info threads：查看所有线程
• thread <id>：切换线程
• thread apply all bt：所有线程的调用栈
• break <location> thread <id>：线程特定断点

11.2 死锁检测技术

1. 预防性设计：
   • 固定锁获取顺序
   • 使用std::scoped_lock
   • 设置锁超时

cpp复制std::timed_mutex mtx1, mtx2;

if(mtx1.try_lock_for(100ms)) {
    if(mtx2.try_lock_for(100ms)) {
        // 成功获取两个锁
        mtx2.unlock();
    }
    mtx1.unlock();
}

11.3 条件变量陷阱

常见错误：

1. 虚假唤醒
2. 通知丢失
3. 条件检查与等待之间的竞态

正确模式：

cpp复制std::mutex mtx;
std::condition_variable cv;
bool ready = false;

// 等待方
std::unique_lock<std::mutex> lock(mtx);
cv.wait(lock, []{ return ready; });

// 通知方
{
    std::lock_guard<std::mutex> lock(mtx);
    ready = true;
}
cv.notify_one();

12. 线程安全设计模式

12.1 单例模式线程安全实现

cpp复制class Singleton {
    static std::atomic<Singleton*> instance;
    static std::mutex mtx;
    
    Singleton() = default;
    
public:
    static Singleton* getInstance() {
        Singleton* tmp = instance.load(std::memory_order_acquire);
        if(tmp == nullptr) {
            std::lock_guard<std::mutex> lock(mtx);
            tmp = instance.load(std::memory_order_relaxed);
            if(tmp == nullptr) {
                tmp = new Singleton();
                instance.store(tmp, std::memory_order_release);
            }
        }
        return tmp;
    }
};

12.2 发布-订阅模式线程安全实现

cpp复制#include <map>
#include <functional>
#include <shared_mutex>

class EventBus {
    std::map<std::string, std::vector<std::function<void()>>> handlers;
    mutable std::shared_mutex mtx;
    
public:
    void subscribe(const std::string& event, std::function<void()> handler) {
        std::unique_lock<std::shared_mutex> lock(mtx);
        handlers[event].push_back(std::move(handler));
    }
    
    void publish(const std::string& event) {
        std::shared_lock<std::shared_mutex> lock(mtx);
        auto it = handlers.find(event);
        if(it != handlers.end()) {
            for(auto& handler : it->second) {
                handler();
            }
        }
    }
};

12.3 读写锁应用模式

cpp复制class ThreadSafeConfig {
    std::map<std::string, std::string> config;
    mutable std::shared_mutex mtx;
    
public:
    std::string get(const std::string& key) const {
        std::shared_lock<std::shared_mutex> lock(mtx);
        auto it = config.find(key);
        return it != config.end() ? it->second : "";
    }
    
    void set(const std::string& key, const std::string& value) {
        std::unique_lock<std::shared_mutex> lock(mtx);
        config[key] = value;
    }
};

13. 多线程与网络编程

13.1 Reactor模式实现

cpp复制class Reactor {
    int epollFd;
    std::unordered_map<int, std::function<void()>> handlers;
    std::mutex mtx;
    std::atomic<bool> running{false};
    
public:
    Reactor() : epollFd(epoll_create1(0)) {}
    
    void registerHandler(int fd, std::function<void()> handler) {
        std::lock_guard<std::mutex> lock(mtx);
        handlers[fd] = std::move(handler);
        
        epoll_event ev;
        ev.events = EPOLLIN | EPOLLET;
        ev.data.fd = fd;
        epoll_ctl(epollFd, EPOLL_CTL_ADD, fd, &ev);
    }
    
    void run() {
        running = true;
        while(running) {
            epoll_event events[10];
            int n = epoll_wait(epollFd, events, 10, 100);
            for(int i=0; i<n; ++i) {
                std::lock_guard<std::mutex> lock(mtx);
                if(auto it = handlers.find(events[i].data.fd); it != handlers.end()) {
                    it->second();
                }
            }
        }
    }
    
    void stop() { running = false; }
};

13.2 线程安全的连接池

cpp复制class ConnectionPool {
    std::vector<Connection*> pool;
    std::mutex mtx;
    std::condition_variable cv;
    
public:
    Connection* getConnection() {
        std::unique_lock<std::mutex> lock(mtx);
        cv.wait(lock, [this]{ return !pool.empty(); });
        auto conn = pool.back();
        pool.pop_back();
        return conn;
    }
    
    void releaseConnection(Connection* conn) {
        std::lock_guard<std::mutex> lock(mtx);
        pool.push_back(conn);
        cv.notify_one();
    }
};

14. 多线程与异步I/O

14.1 io_uring基础使用

cpp复制#include <liburing.h>

void ioUringExample() {
    io_uring ring;
    io_uring_queue_init(32, &ring, 0);
    
    int fd = open("test.txt", O_RDONLY);
    char buf[4096];
    
    // 提交读请求
    io_uring_sqe* sqe = io_uring_get_sqe(&ring);
    io_uring_prep_read(sqe, fd, buf, sizeof(buf), 0);
    io_uring_submit(&ring);
    
    // 等待完成
    io_uring_cqe* cqe;
    io_uring_wait_cqe(&ring, &cqe);
    if(cqe->res > 0) {
        std::cout << "Read " << cqe->res << " bytes" << std::endl;
    }
    
    io_uring_queue_exit(&ring);
    close(fd);
}

14.2 异步文件操作模式

cpp复制#include <aio.h>

void asyncFileIO() {
    int fd = open("file.txt", O_RDONLY);
    char buf[4096];
    
    struct aiocb cb = {0};
    cb.aio_fildes = fd;
    cb.aio_buf = buf;
    cb.aio_nbytes = sizeof(buf);
    
    aio_read(&cb);
    
    while(aio_error(&cb) == EINPROGRESS) {
        // 可以做其他工作
    }
    
    if(aio_error(&cb) == 0) {
        ssize_t bytes = aio_return(&cb);
        std::cout << "Read " << bytes << " bytes" << std::endl;
    }
    
    close(fd);
}

15. 多线程与GPU计算

15.1 CUDA线程模型

cpp复制__global__ void addKernel(int* c, const int* a, const int* b) {
    int i = threadIdx.x;
    c[i] = a[i] + b[i];
}

void launchKernel() {
    const int arraySize = 5;
    int a[arraySize] = {1, 2, 3, 4, 5};
    int b[arraySize] = {10, 20, 30, 40, 50};
    int c[arraySize] = {0};
    
    int *dev_a, *dev_b, *dev_c;
    cudaMalloc(&dev_a, arraySize * sizeof(int));
    cudaMalloc(&dev_b, arraySize * sizeof(int));
    cudaMalloc(&dev_c, arraySize * sizeof(int));
    
    cudaMemcpy(dev_a, a, arraySize * sizeof(int), cudaMemcpyHostToDevice);
    cudaMemcpy(dev_b, b, arraySize * sizeof(int), cudaMemcpyHostToDevice);
    
    addKernel<<<1, arraySize>>>(dev_c, dev_a, dev_b);
    
    cudaMemcpy(c, dev_c, arraySize * sizeof(int), cudaMemcpyDeviceToHost);
    
    for(int i=0; i<arraySize; ++i) {
        std::cout << c[i] << " ";
    }
    
    cudaFree(dev_a);
    cudaFree(dev_b);
    cudaFree(dev_c);
}

15.2 OpenMP与GPU加速

cpp复制#include <omp.h>

void matrixMultiply(float* A, float* B, float* C, int N) {
    #pragma omp target teams distribute parallel for \
        map(to: A[0:N*N], B[0:N*N]) map(from: C[0:N*N])
    for(int i=0; i<N; ++i) {
        for(int j=0; j<N; ++j) {
            float sum = 0.0f;
            for(int k=0; k<N; ++k) {
                sum += A[i*N+k] * B[k*N+j];
            }
            C[i*N+j] = sum;
        }
    }
}

16. 多线程与容器技术

16.1 容器中的线程隔离

容器线程注意事项：

1. 每个容器有自己的PID命名空间
2. 线程仍然共享相同的内核调度器
3. cgroup限制适用于容器内所有线程

bash复制# 设置容器CPU限制
docker run --cpus=2 your_image

16.2 Kubernetes中的线程调度

Kubernetes调度单元是Pod，一个Pod内的容器：

• 共享相同的网络命名空间
• 共享相同的IPC命名空间
• 可以通过共享卷通信

最佳实践：

• 将紧密协作的线程放在同一Pod
• 独立服务使用不同Pod
• 合理设置资源请求和限制

yaml复制apiVersion: v1
kind: Pod
metadata:
  name: threaded-app
spec:
  containers:
  - name: worker
    image: your-worker-image
    resources:
      requests:
        cpu: "2"
        memory: "1Gi"
      limits:
        cpu: "4"
        memory: "2Gi"

17. 多线程与实时系统

17.1 实时线程调度策略

Linux实时调度策略：

1. SCHED_FIFO：先进先出，无时间片
2. SCHED_RR：轮转调度，有时间片
3. SCHED_DEADLINE：基于截止时间

cpp复制// 设置实时优先级
struct sched_param param;
param.sched_priority = sched_get_priority_max(SCHED_FIFO);
pthread_setschedparam(pthread_self(), SCHED_FIFO, &param);

17.2 实时系统线程设计原则

1. 最小化锁使用
2. 避免动态内存分配
3. 使用无等待数据结构
4. 严格控制线程优先级
5. 确保最坏执行时间可预测

cpp复制// 实时安全的环形缓冲区
template<typename T, size_t Size>
class RingBuffer {
    std::array<T, Size> buffer;
    std::atomic<size_t> head{0}, tail{0};
    
public:
    bool push(const T& item) {
        size_t currTail = tail.load(std::memory_order_relaxed);
        size_t nextTail = (currTail + 1) % Size;
        if(nextTail == head.load(std::memory_order_acquire)) {
            return false; // 满
        }
        buffer[currTail] = item;
        tail.store(nextTail, std::memory_order_release);
        return true;
    }
    
    bool pop(T& item) {
        size_t currHead = head.load(std::memory_order_relaxed);
        if(currHead == tail.load(std::memory_order_acquire)) {
            return false; // 空
        }
        item = buffer[currHead];
        head.store((currHead + 1) % Size, std::memory_order_release);
        return true;
    }
};

18. 多线程与安全编程

18.1 线程安全函数设计

线程安全函数的特征：

1. 不依赖全局或静态变量
2. 所有数据通过参数传递
3. 使用线程安全库函数
4. 可重入(reentrant)

cpp复制// 线程安全的strtok替代品
char* strtok_r(char* str, const char* delim, char** saveptr) {
    // 实现略...
}

18.2 安全的多线程日志系统

cpp复制class ThreadSafeLogger {
    std::ofstream logFile;
    std::mutex mtx;
    
public:
    ThreadSafeLogger(const std::string& filename) 
        : logFile(filename, std::ios::app) {}
    
    void log(const std::string& message) {
        std::lock_guard<std::mutex> lock(mtx);
        auto now = std::chrono::system_clock::now();
        auto now_time = std::chrono::system_clock::to_time_t(now);
        logFile << std::put_time(std::localtime(&now_time), "%F %T")
                << " [" << std::this_thread::get_id() << "] "
                << message << std::endl;
    }
};

19. 多线程与机器学习

19.1 并行训练模式

cpp复制void parallelTraining(std::vector<Data>& dataset, Model& model, int epochs) {
    auto worker = [&](int start, int end) {
        for(int i=start; i<end; ++i) {
            model.trainOnSample(dataset[i]);
        }
    };
    
    const int numThreads = std::thread::hardware_concurrency();
    const int chunkSize = dataset.size() / numThreads;
    
    std::vector<std::thread> threads;
    for(int i=0; i<numThreads; ++i) {
        int start = i * chunkSize;
        int end = (i == numThreads-1) ? dataset.size() : start + chunkSize;
        threads.emplace_back(worker, start, end);
    }
    
    for(auto& t : threads) t.join();
}

19.2 参数服务器模式

cpp复制class ParameterServer {
    std::vector<float> parameters;
    mutable std::shared_mutex mtx;
    
public:
    void update(const std::vector<float>& gradients) {
        std::unique_lock<std::shared_mutex> lock(mtx);
        for(size_t i=0; i<parameters.size(); ++i) {
            parameters[i] -= 0.01f * gradients[i];
        }
    }
    
    std::vector<float> getParameters() const {
        std::shared_lock<std::shared_mutex> lock(mtx);
        return parameters;
    }
};

20. 多线程与区块链

20.1 并行挖矿实现

cpp复制void mineBlock(Block& block, int difficulty, std::atomic<bool>& found) {
    std::string target(difficulty, '0');
    while(!found) {
        block.nonce++;
        std::string hash = block.calculateHash();
        if(hash.substr(0, difficulty) == target) {
            found = true;
            std::cout << "Block mined by thread " 
                      << std::this_thread::get_id() << std::endl;
        }
    }
}

void parallelMining(Block& block, int difficulty) {
    std::atomic<bool> found{false};
    const int numThreads = std::thread::hardware_concurrency();
    
    std::vector<std::thread> miners;
    for(int i=0; i<numThreads; ++i) {
        miners.emplace_back(mineBlock, std::ref(block), 
                           difficulty, std::ref(found));
    }
    
    for(auto& t : miners) t.join();
}

20.2 智能合约并行执行

cpp复制class ParallelExecutor {
    std::vector<Contract> contracts;
    std::mutex resultMtx;
    std::vector<Result> results;
    
    void executeContract(size_t index) {
        Result r = contracts[index].execute();
        std::lock_guard<std::mutex> lock(resultMtx);
        results[index] = r;
    }
    
public:
    std::vector<Result> executeAll() {
        results.resize(contracts.size());
        std::vector<std::thread> workers;
        
        for(size_t i=0; i<contracts.size(); ++i) {
            workers.emplace_back(&ParallelExecutor::executeContract, this, i);
        }
        
        for(auto& t : workers) t.join();
        return results;
    }
};