C++并发编程与线程安全

C++并发编程与线程安全

并发编程是现代C++中的重要主题。通过多线程技术，程序可以充分利用多核处理器的性能，提高程序的响应速度和吞吐量。

std::thread是C++11引入的标准线程库，提供了跨平台的线程创建和管理功能。

#include
#include
#include
#include

void worker_function(int id) {
std::cout << "Thread " << id << " starting\n";
std::this_thread::sleep_for(std::chrono::milliseconds(100));
std::cout << "Thread " << id << " finished\n";
}

void basic_thread_example() {
std::vector threads;

for (int i = 0; i < 5; ++i) {
threads.emplace_back(worker_function, i);
}

for (auto& t : threads) {
t.join();
}
}

互斥锁是保护共享数据的基本机制，确保同一时刻只有一个线程访问临界区。

#include

class Counter {
int value_;
mutable std::mutex mutex_;

public:
Counter() : value_(0) {}

void increment() {
std::lock_guard lock(mutex_);
++value_;
}

void add(int delta) {
std::lock_guard lock(mutex_);
value_ += delta;
}

int get() const {
std::lock_guard lock(mutex_);
return value_;
}
};

void mutex_example() {
Counter counter;
std::vector threads;

for (int i = 0; i < 10; ++i) {
threads.emplace_back([&counter]() {
for (int j = 0; j < 1000; ++j) {
counter.increment();
}
});
}

for (auto& t : threads) {
t.join();
}

std::cout << "Final count: " << counter.get() << "\n";
}

条件变量用于线程间的同步，允许线程等待特定条件成立。

#include
#include

template
class ThreadSafeQueue {
std::queue queue_;
mutable std::mutex mutex_;
std::condition_variable cond_;

public:
void push(T value) {
std::lock_guard lock(mutex_);
queue_.push(std::move(value));
cond_.notify_one();
}

bool try_pop(T& value) {
std::lock_guard lock(mutex_);
if (queue_.empty()) {
return false;
}
value = std::move(queue_.front());
queue_.pop();
return true;
}

void wait_and_pop(T& value) {
std::unique_lock lock(mutex_);
cond_.wait(lock, [this] { return !queue_.empty(); });
value = std::move(queue_.front());
queue_.pop();
}

bool empty() const {
std::lock_guard lock(mutex_);
return queue_.empty();
}
};

void producer_consumer_example() {
ThreadSafeQueue queue;

std::thread producer([&queue]() {
for (int i = 0; i < 10; ++i) {
queue.push(i);
std::cout << "Produced: " << i << "\n";
std::this_thread::sleep_for(std::chrono::milliseconds(50));
}
});

std::thread consumer([&queue]() {
for (int i = 0; i < 10; ++i) {
int value;
queue.wait_and_pop(value);
std::cout << "Consumed: " << value << "\n";
}
});

producer.join();
consumer.join();
}

原子操作提供了无锁的线程安全机制，适用于简单的数据类型。

#include

class AtomicCounter {
std::atomic value_;

public:
AtomicCounter() : value_(0) {}

void increment() {
value_.fetch_add(1, std::memory_order_relaxed);
}

int get() const {
return value_.load(std::memory_order_relaxed);
}

bool compare_and_swap(int expected, int desired) {
return value_.compare_exchange_strong(expected, desired);
}
};

void atomic_example() {
AtomicCounter counter;
std::vector threads;

for (int i = 0; i < 10; ++i) {
threads.emplace_back([&counter]() {
for (int j = 0; j < 1000; ++j) {
counter.increment();
}
});
}

for (auto& t : threads) {
t.join();
}

std::cout << "Atomic count: " << counter.get() << "\n";
}

读写锁允许多个读者同时访问，但写者独占访问。

#include

class SharedData {
mutable std::shared_mutex mutex_;
std::vector data_;

public:
void write(int value) {
std::unique_lock lock(mutex_);
data_.push_back(value);
}

int read(size_t index) const {
std::shared_lock lock(mutex_);
if (index < data_.size()) {
return data_[index];
}
return -1;
}

size_t size() const {
std::shared_lock lock(mutex_);
return data_.size();
}
};

void reader_writer_example() {
SharedData data;

std::thread writer([&data]() {
for (int i = 0; i < 100; ++i) {
data.write(i);
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
});

std::vector readers;
for (int i = 0; i < 5; ++i) {
readers.emplace_back([&data, i]() {
for (int j = 0; j < 50; ++j) {
size_t size = data.size();
if (size > 0) {
int value = data.read(size - 1);
std::cout << "Reader " << i << " read: " << value << "\n";
}
std::this_thread::sleep_for(std::chrono::milliseconds(20));
}
});
}

writer.join();
for (auto& r : readers) {
r.join();
}
}

线程池可以复用线程，避免频繁创建和销毁线程的开销。

#include

class ThreadPool {
std::vector workers_;
ThreadSafeQueue> tasks_;
std::atomic stop_;

public:
explicit ThreadPool(size_t num_threads) : stop_(false) {
for (size_t i = 0; i < num_threads; ++i) {
workers_.emplace_back([this]() {
while (!stop_.load()) {
std::function task;
if (tasks_.try_pop(task)) {
task();
} else {
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
}
});
}
}

~ThreadPool() {
stop_.store(true);
for (auto& worker : workers_) {
if (worker.joinable()) {
worker.join();
}
}
}

template
void submit(F&& task) {
tasks_.push(std::forward(task));
}
};

void thread_pool_example() {
ThreadPool pool(4);

for (int i = 0; i < 20; ++i) {
pool.submit([i]() {
std::cout << "Task " << i << " executing\n";
std::this_thread::sleep_for(std::chrono::milliseconds(100));
});
}

std::this_thread::sleep_for(std::chrono::seconds(3));
}

future和promise提供了异步操作的结果传递机制。

#include

int compute_value(int x) {
std::this_thread::sleep_for(std::chrono::milliseconds(100));
return x * x;
}

void future_promise_example() {
std::future result = std::async(std::launch::async, compute_value, 10);

std::cout << "Computing...\n";
int value = result.get();
std::cout << "Result: " << value << "\n";

std::promise promise;
std::future future = promise.get_future();

std::thread([&promise]() {
std::this_thread::sleep_for(std::chrono::milliseconds(100));
promise.set_value(42);
}).detach();

std::cout << "Waiting for promise...\n";
std::cout << "Promise value: " << future.get() << "\n";
}

死锁是并发编程中常见的问题，需要通过正确的锁顺序或超时机制来避免。

class BankAccount {
int balance_;
mutable std::mutex mutex_;

public:
explicit BankAccount(int balance) : balance_(balance) {}

void transfer(BankAccount& to, int amount) {
std::lock(mutex_, to.mutex_);
std::lock_guard lock1(mutex_, std::adopt_lock);
std::lock_guard lock2(to.mutex_, std::adopt_lock);

if (balance_ >= amount) {
balance_ -= amount;
to.balance_ += amount;
}
}

int balance() const {
std::lock_guard lock(mutex_);
return balance_;
}
};

void deadlock_avoidance_example() {
BankAccount acc1(1000);
BankAccount acc2(2000);

std::thread t1([&]() {
for (int i = 0; i < 100; ++i) {
acc1.transfer(acc2, 10);
}
});

std::thread t2([&]() {
for (int i = 0; i < 100; ++i) {
acc2.transfer(acc1, 10);
}
});

t1.join();
t2.join();

std::cout << "Account 1: " << acc1.balance() << "\n";
std::cout << "Account 2: " << acc2.balance() << "\n";
}

并发编程需要仔细设计和测试，以确保程序的正确性和性能。理解各种同步机制的特点和适用场景，是编写高质量并发代码的关键。