带背压的异步日志核心是在日志生产端和异步消费端之间加入流量控制机制，当日志队列达到容量上限时，通过阻塞生产线程或直接丢弃日志的方式，避免队列无限增长导致内存溢出，同时减少对主业务的影响。
核心设计思路
整个模块分为三个部分：线程安全的日志队列、日志生产接口、日志消费线程。背压逻辑主要在日志队列的入队操作中实现，根据选择的策略决定队列满时的行为。
线程安全队列基础实现
首先实现一个支持最大容量限制的线程安全队列，作为日志的中转容器，内部使用std::mutex和std::condition_variable保证线程安全。
#include <queue>
#include <mutex>
#include <condition_variable>
#include <string>
#include <thread>
#include <iostream>
#include <chrono>
// 线程安全的有界队列模板
template <typename T>
class BoundedQueue {
private:
std::queue<T> queue_;
size_t max_size_;
std::mutex mutex_;
std::condition_variable not_empty_cv_; // 消费者等待队列非空
std::condition_variable not_full_cv_; // 生产者等待队列非满
bool stop_flag_ = false; // 停止标志，用于退出消费线程
public:
explicit BoundedQueue(size_t max_size) : max_size_(max_size) {}
// 入队：队列满时阻塞
bool push_block(const T& item) {
std::unique_lock<std::mutex> lock(mutex_);
// 等待队列不满且未停止
not_full_cv_.wait(lock, [this]() { return queue_.size() < max_size_ || stop_flag_; });
if (stop_flag_) return false;
queue_.push(item);
not_empty_cv_.notify_one(); // 通知消费者队列有新元素
return true;
}
// 入队：队列满时直接丢弃
bool push_drop(const T& item) {
std::unique_lock<std::mutex> lock(mutex_);
if (queue_.size() >= max_size_ || stop_flag_) {
return false; // 队列满或已停止，丢弃日志
}
queue_.push(item);
not_empty_cv_.notify_one();
return true;
}
// 出队，队列为空时阻塞
bool pop(T& item) {
std::unique_lock<std::mutex> lock(mutex_);
not_empty_cv_.wait(lock, [this]() { return !queue_.empty() || stop_flag_; });
if (stop_flag_ && queue_.empty()) return false;
item = std::move(queue_.front());
queue_.pop();
not_full_cv_.notify_one(); // 通知生产者队列有空位
return true;
}
// 停止队列，唤醒所有等待的线程
void stop() {
std::unique_lock<std::mutex> lock(mutex_);
stop_flag_ = true;
not_full_cv_.notify_all();
not_empty_cv_.notify_all();
}
};
日志模块实现
基于上面的有界队列，封装日志写入接口和异步消费逻辑，日志内容使用std::string存储，消费线程负责将日志写入文件。
日志消费线程
消费线程循环从队列中取出日志，写入到本地文件，当队列停止且所有日志消费完成后退出线程。
class AsyncLogger {
private:
BoundedQueue<std::string>* log_queue_;
std::thread consumer_thread_;
std::ofstream log_file_;
// 消费线程执行函数
void consume_loop() {
std::string log_item;
while (true) {
if (!log_queue_->pop(log_item)) {
// 队列停止且为空，退出循环
break;
}
// 写入日志文件，实际场景可添加时间戳、日志级别等信息
if (log_file_.is_open()) {
log_file_ << log_item << std::endl;
}
}
}
public:
AsyncLogger(BoundedQueue<std::string>* queue, const std::string& file_path)
: log_queue_(queue) {
log_file_.open(file_path, std::ios::app);
if (!log_file_.is_open()) {
std::cerr << "无法打开日志文件: " << file_path << std::endl;
}
// 启动消费线程
consumer_thread_ = std::thread(&AsyncLogger::consume_loop, this);
}
~AsyncLogger() {
// 停止队列，等待消费线程退出
log_queue_->stop();
if (consumer_thread_.joinable()) {
consumer_thread_.join();
}
if (log_file_.is_open()) {
log_file_.close();
}
}
};
两种背压策略的使用示例
下面分别演示阻塞策略和丢弃策略的使用方式，模拟高并发日志写入场景。
队列满时阻塞策略
int main() {
// 队列最大容量为100
BoundedQueue<std::string> log_queue(100);
// 创建异步日志器，日志写入到test_block.log
AsyncLogger logger(&log_queue, "test_block.log");
// 模拟10个线程并发写入日志
std::vector<std::thread> write_threads;
for (int i = 0; i < 10; ++i) {
write_threads.emplace_back([i, &log_queue]() {
for (int j = 0; j < 50; ++j) {
std::string log_content = "线程" + std::to_string(i) + " 日志" + std::to_string(j);
// 使用阻塞入队，队列满时写入线程会阻塞等待
log_queue.push_block(log_content);
// 模拟业务耗时
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
});
}
for (auto& t : write_threads) {
if (t.joinable()) t.join();
}
// 等待所有日志消费完成
std::this_thread::sleep_for(std::chrono::seconds(1));
return 0;
}
队列满时丢弃策略
int main() {
// 队列最大容量为100
BoundedQueue<std::string> log_queue(100);
// 创建异步日志器，日志写入到test_drop.log
AsyncLogger logger(&log_queue, "test_drop.log");
// 模拟10个线程并发写入日志
std::vector<std::thread> write_threads;
int drop_count = 0;
std::mutex drop_mutex;
for (int i = 0; i < 10; ++i) {
write_threads.emplace_back([i, &log_queue, &drop_count, &drop_mutex]() {
for (int j = 0; j < 50; ++j) {
std::string log_content = "线程" + std::to_string(i) + " 日志" + std::to_string(j);
// 使用丢弃入队，队列满时直接丢弃日志
if (!log_queue.push_drop(log_content)) {
std::unique_lock<std::mutex> lock(drop_mutex);
drop_count++;
}
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
});
}
for (auto& t : write_threads) {
if (t.joinable()) t.join();
}
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "丢弃的日志数量: " << drop_count << std::endl;
return 0;
}
两种策略的选择建议
- 阻塞策略适合日志重要性高、不能丢失的场景，比如交易流水、错误日志，但需要注意避免阻塞关键业务线程，必要时可单独开辟日志写入线程池。
- 丢弃策略适合日志允许部分丢失、对性能要求极高的场景，比如高频监控指标、调试日志，避免日志写入拖慢主业务流程。
注意事项
实际生产环境中，还需要考虑日志文件的滚动切割、日志格式的统一封装、队列容量的动态调优等问题，本文示例仅展示核心背压逻辑的实现，可根据业务需求扩展功能。
队列的停止逻辑需要保证所有已入队的日志都能被消费完成，避免程序退出时丢失日志。如果消费速度长期低于生产速度，需要根据业务情况调整队列容量或者优化日志消费的性能，比如批量写入文件、使用更快的存储设备。