一、从需求出发：为什么需要分布式日志系统

说实话，我做了这么多年C++后端，日志系统几乎是每个项目都绕不开的坎。小到单机服务，大到几百台机器的集群，日志就是系统的「黑匣子」。但单机日志好写，分布式环境下的日志收集，那才叫真正的挑战。

我在上一家公司就遇到过这么个场景：线上有几十台机器，每台都在疯狂输出日志。出问题的时候，运维同学要一台台登录上去看，用grep翻半天。有时候问题定位完了，黄花菜都凉了。说白了，我们需要一个能统一收集、存储、分析日志的系统。

核心需求拆解

  • 日志采集：从多个节点实时拉取日志，不能丢数据
  • 统一存储：所有日志集中存放，方便检索
  • 聚合分析：按时间、级别、关键字做统计
  • 多线程安全：高并发写入时不能崩
  • 性能调优：千万级日志量下，延迟控制在毫秒级

我的经验之谈：很多团队一开始图省事，用数据库存日志。结果日志量一上来，数据库先扛不住了。日志系统，本质上是个「写多读少」的场景，用STL容器+内存缓冲，比数据库靠谱得多。

二、架构设计：分层解耦才是王道

架构这东西，我习惯先画图再写代码。你想想看，如果一上来就堆代码，后面改起来多痛苦。

分布式日志收集系统架构图 日志采集层 Agent 节点 1..N std::queue 缓冲 传输层 TCP / 消息队列 批量发送 存储层 std::vector 环形缓冲 内存 + 文件持久化 聚合分析层 std::map 统计 std::sort 排序 查询接口层 按时间/级别/关键字 std::find_if 过滤 多线程安全 std::mutex 保护 无锁队列优化 核心容器：std::vector / std::map / std::queue / std::string 核心算法：std::sort / std::find_if / std::count_if / std::for_each

这个架构图我画了好几个版本，最终定下来四层：采集、传输、存储、分析。每一层职责清晰，层与层之间通过接口解耦。嗯，这里要注意：千万别把逻辑都揉在一个类里，否则后面维护起来想哭。

三、STL容器选型：选对容器，事半功倍

容器选型这块，我踩过不少坑。以前图省事全用std::vector，结果在频繁插入删除的场景下性能惨不忍睹。后来学乖了，按场景选容器。

3.1 日志存储：std::vector + 环形缓冲

日志写入是典型的「尾部追加」模式。std::vector的push_back在尾部操作，均摊O(1)，完美匹配。但内存会不断增长怎么办？我习惯用环形缓冲：预分配固定大小，满了就覆盖最旧的。

// 环形缓冲实现
template<typename T>
class RingBuffer {
    std::vector<T> buffer_;
    size_t head_ = 0;
    size_t tail_ = 0;
    size_t capacity_;
    std::mutex mtx_;
    
public:
    explicit RingBuffer(size_t cap) : capacity_(cap) {
        buffer_.resize(cap);
    }
    
    void push(const T& item) {
        std::lock_guard<std::mutex> lock(mtx_);
        buffer_[tail_] = item;
        tail_ = (tail_ + 1) % capacity_;
        if (tail_ == head_) {
            head_ = (head_ + 1) % capacity_;  // 覆盖最旧
        }
    }
    
    std::vector<T> drain() {
        std::lock_guard<std::mutex> lock(mtx_);
        std::vector<T> result;
        while (head_ != tail_) {
            result.push_back(buffer_[head_]);
            head_ = (head_ + 1) % capacity_;
        }
        return result;
    }
};

避坑指南：我曾经在环形缓冲里用std::list，结果内存碎片化严重，GC频繁。换成vector预分配后，性能提升了3倍。记住：日志场景下，连续内存是王道。

3.2 聚合统计：std::map vs std::unordered_map

统计每种日志级别的数量，用map最顺手。但这里有个选择：

容器适用场景我的建议
std::map需要有序遍历（如按时间排序）日志按时间聚合时用
std::unordered_map只查不排序，追求速度按级别统计时用，O(1)查找
// 按日志级别统计
std::unordered_map<std::string, int> levelCount;
levelCount["ERROR"] = 0;
levelCount["WARN"] = 0;
levelCount["INFO"] = 0;

// 聚合时
for (const auto& log : logs) {
    levelCount[log.level]++;
}

// 按数量降序输出
std::vector<std::pair<std::string, int>> sorted(levelCount.begin(), levelCount.end());
std::sort(sorted.begin(), sorted.end(), 
    [](const auto& a, const auto& b) { return a.second > b.second; });

四、多线程安全：别让并发成为噩梦

多线程写日志，最怕什么？数据竞争、死锁、性能下降。我见过一个项目，日志系统加了锁之后，写入性能直接掉了80%。

4.1 生产者-消费者模型

我习惯用「多生产者-单消费者」模型。多个线程往队列里写，一个后台线程负责批量刷盘。

class LogCollector {
    std::queue<LogEntry> queue_;
    std::mutex mtx_;
    std::condition_variable cv_;
    std::thread worker_;
    bool running_ = true;
    
public:
    LogCollector() {
        worker_ = std::thread([this] { flushLoop(); });
    }
    
    void submit(const LogEntry& entry) {
        {
            std::lock_guard<std::mutex> lock(mtx_);
            queue_.push(entry);
        }
        cv_.notify_one();  // 唤醒消费者
    }
    
    void flushLoop() {
        while (running_) {
            std::unique_lock<std::mutex> lock(mtx_);
            cv_.wait(lock, [this] { return !queue_.empty() || !running_; });
            
            // 批量取出
            std::vector<LogEntry> batch;
            while (!queue_.empty() && batch.size() < 1000) {
                batch.push_back(queue_.front());
                queue_.pop();
            }
            lock.unlock();
            
            // 批量写入（无锁操作）
            writeBatch(batch);
        }
    }
};

注意：condition_variable的wait一定要配合lambda判断条件，防止虚假唤醒。我曾经因为这个bug，线上日志重复写入了好几万条，排查了一整天才找到原因。

五、性能调优：从毫秒到微秒的进化

性能调优这事，说白了就是「找瓶颈、干掉它」。我一般用perf工具先跑一遍，看看热点在哪。

5.1 减少锁竞争

锁是性能杀手。我常用的优化手段：

  • 细粒度锁：一个队列一把锁改成多个队列多把锁
  • 无锁队列：用std::atomic实现，适合高吞吐场景
  • 批量操作：攒够一批再写，减少锁次数

5.2 内存分配优化

频繁的new/delete会导致性能抖动。我的做法：

  • 预分配内存（reserve/resize）
  • 使用对象池复用LogEntry对象
  • 避免std::string的频繁拷贝，用移动语义
// 对象池示例
class LogEntryPool {
    std::vector<LogEntry> pool_;
    std::atomic<size_t> index_{0};
    
public:
    LogEntry* acquire() {
        size_t idx = index_.fetch_add(1);
        if (idx >= pool_.size()) {
            pool_.resize(pool_.size() * 2);
        }
        return &pool_[idx];
    }
};

六、完整代码实现：把理论落地

好了，理论说完了，咱们看看完整的实现。这个版本我简化了一些细节，但核心逻辑都在。

#include <iostream>
#include <vector>
#include <map>
#include <queue>
#include <mutex>
#include <thread>
#include <algorithm>
#include <chrono>
#include <sstream>

// 日志条目
struct LogEntry {
    std::string timestamp;
    std::string level;   // ERROR, WARN, INFO
    std::string message;
    int nodeId;
};

// 分布式日志收集器
class DistributedLogCollector {
    RingBuffer<LogEntry> buffer_{100000};  // 10万条环形缓冲
    std::vector<LogEntry> archive_;        // 归档存储
    std::mutex archiveMtx_;
    
public:
    void ingest(const LogEntry& entry) {
        buffer_.push(entry);
    }
    
    void flushToArchive() {
        auto batch = buffer_.drain();
        std::lock_guard<std::mutex> lock(archiveMtx_);
        archive_.insert(archive_.end(), batch.begin(), batch.end());
    }
    
    // 按级别统计
    std::map<std::string, int> countByLevel() {
        std::lock_guard<std::mutex> lock(archiveMtx_);
        std::map<std::string, int> result;
        for (const auto& log : archive_) {
            result[log.level]++;
        }
        return result;
    }
    
    // 按关键字搜索
    std::vector<LogEntry> searchByKeyword(const std::string& keyword) {
        std::lock_guard<std::mutex> lock(archiveMtx_);
        std::vector<LogEntry> result;
        std::copy_if(archive_.begin(), archive_.end(), 
                     std::back_inserter(result),
                     [&](const LogEntry& e) {
                         return e.message.find(keyword) != std::string::npos;
                     });
        return result;
    }
    
    // 获取最近N条
    std::vector<LogEntry> recent(size_t n) {
        std::lock_guard<std::mutex> lock(archiveMtx_);
        size_t start = (archive_.size() > n) ? archive_.size() - n : 0;
        return std::vector<LogEntry>(archive_.begin() + start, archive_.end());
    }
};

// 模拟多节点写入
void simulateNode(DistributedLogCollector& collector, int nodeId) {
    std::vector<std::string> levels = {"ERROR", "WARN", "INFO"};
    for (int i = 0; i < 1000; i++) {
        LogEntry entry;
        entry.timestamp = std::to_string(
            std::chrono::system_clock::now().time_since_epoch().count());
        entry.level = levels[rand() % 3];
        entry.message = "Node " + std::to_string(nodeId) + 
                        " log #" + std::to_string(i);
        entry.nodeId = nodeId;
        collector.ingest(entry);
    }
}

int main() {
    DistributedLogCollector collector;
    
    // 启动10个模拟节点
    std::vector<std::thread> threads;
    for (int i = 0; i < 10; i++) {
        threads.emplace_back(simulateNode, std::ref(collector), i);
    }
    
    // 等待写入完成
    for (auto& t : threads) t.join();
    
    // 刷入归档
    collector.flushToArchive();
    
    // 聚合分析
    auto stats = collector.countByLevel();
    std::cout << "=== 日志统计 ===" << std::endl;
    for (const auto& [level, count] : stats) {
        std::cout << level << ": " << count << std::endl;
    }
    
    // 搜索ERROR日志
    auto errors = collector.searchByKeyword("ERROR");
    std::cout << "\n=== ERROR日志数量: " << errors.size() << " ===" << std::endl;
    
    return 0;
}

七、总结与思考

写这个系统，我最大的体会是：STL容器不是万能的，但用好了能省90%的功夫。选对容器、控制好锁粒度、批量操作，这三点做到位，性能基本不会差。

如果你在实际项目中遇到日志量特别大的场景（比如每秒百万级），可以考虑用无锁队列替代std::queue，或者用内存映射文件做持久化。但大多数场景下，上面这套方案已经足够用了。

最后说一句：日志系统一定要在项目初期就设计好，别等到线上出问题了再补。我见过太多团队因为日志没做好，排查问题花了好几天——那滋味，真不好受。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →