📚 课程目录第4章日志采集

4. 日志采集：文件采集、网络采集、Kafka等

日志采集，说白了就是把散落在各处的日志收拢到一个地方。我刚开始做运维那会儿，还在用 scp 逐台拷贝日志，效率低不说，还经常漏数据。后来才明白——采集层设计得好不好，直接决定了你的巡检系统能跑多稳、查多快。

你想想看，日志是分散的，但分析必须集中。没有一套靠谱的采集方案，后面写的规则再漂亮也是白搭。这一章咱们就拆开聊聊三种主流采集方式：文件采集、网络采集和Kafka 采集。每种方式都有它的脾气，选对了事半功倍。

核心观点：采集不是"把日志搬过来"那么简单。你需要考虑实时性、资源消耗、可靠性，以及后续的扩展性。选型一旦定下来，后期改造成本很高。

4.1 文件采集：最基础，也最容易踩坑

文件采集是最传统的做法。应用把日志写到磁盘，采集程序从文件里读。听起来简单，但里面的门道不少。

常见的三种文件采集方式：

方式	原理	实时性	资源消耗	典型工具
tail -F	跟踪文件尾部追加	较高	低	tail 命令、Logstash file input
inotify	内核事件通知	高	极低	inotifywait、Python pyinotify
轮询	定期检查文件变化	取决于间隔	中等	自定义脚本、Logstash poll

我个人习惯优先用 inotify。它在 Linux 内核层面监听文件事件，不会漏掉任何一次写入。但要注意——inotify 对目录数量有限制，默认监控上限是 8192 个。我曾经在监控一个大规模集群时，因为忘记调这个参数，导致采集进程莫名其妙挂掉。嗯，这个坑我替你们踩过了。

来看看一个简单的 Python inotify 采集示例：

import pyinotify
import time

class LogHandler(pyinotify.ProcessEvent):
    def process_IN_MODIFY(self, event):
        with open(event.pathname, 'r') as f:
            f.seek(0, 2)  # 跳到文件末尾
            line = f.readline()
            while line:
                print(f"[采集] {event.pathname}: {line.strip()}")
                line = f.readline()

wm = pyinotify.WatchManager()
handler = LogHandler()
notifier = pyinotify.Notifier(wm, handler)

wm.add_watch('/var/log/app', pyinotify.IN_MODIFY, rec=True)
notifier.loop()

提示：文件采集最容易忽略的是日志轮转。你监控的日志文件可能被 rename 或删除，记得处理 IN_MOVE_SELF 和 IN_DELETE_SELF 事件，否则会丢数据。

4.2 网络采集：远程日志的标配

很多设备不支持写本地文件，或者你需要采集远程主机的日志。这时候就得靠网络采集了。说白了就是通过 TCP/UDP 或 HTTP 把日志发过来。

常见的网络采集场景：

网络设备（路由器、交换机）通过 syslog 发送日志
应用服务通过 HTTP/RESTful 接口上报状态
自定义代理通过 TCP 长连接批量传输

举个例子，一个基于 TCP Socket 的日志采集服务：

import socket
import threading

def handle_client(conn, addr):
    print(f"[连接] 客户端 {addr} 已接入")
    with conn:
        while True:
            data = conn.recv(4096)
            if not data:
                break
            print(f"[日志] {addr}: {data.decode().strip()}")

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('0.0.0.0', 9510))
server.listen(10)
print("[采集] TCP 日志采集服务已启动，端口 9510")

while True:
    conn, addr = server.accept()
    threading.Thread(target=handle_client, args=(conn, addr)).start()

注意：网络采集最大的风险是数据丢失。UDP 本身不保证送达，TCP 虽然可靠，但如果接收端处理不过来，连接可能被重置。我建议在高吞吐场景下引入本地缓存或确认重传机制。

网络采集的实时性通常很高，但代价是占用网络带宽和连接资源。你需要根据日志量合理规划采集端的并发能力。我曾经遇到过一个案例：某团队用 HTTP 上报日志，每次请求都新建连接，结果采集服务器被大量 TIME_WAIT 连接拖垮。后来改成 TCP 长连接，问题就解决了。

4.3 Kafka 采集：大规模日志的首选

当日志量达到 TB 级别，文件采集和网络采集都扛不住的时候，Kafka 就派上用场了。说白了，Kafka 是一个高吞吐的分布式消息队列，专门用来做数据管道。

为什么用 Kafka？

削峰填谷：日志产生速度不均匀，Kafka 能缓冲峰值流量
解耦：采集端和消费端互不影响
持久化：日志可以保留在 Kafka 中，方便重放和回溯
多消费者：一份日志可以被多个系统同时消费

一个典型的 Kafka 日志采集架构如下：

在实际项目中，我推荐用 Filebeat → Kafka → Logstash → Elasticsearch 这条链路。Filebeat 负责轻量级采集，Kafka 做缓冲，Logstash 做过滤和转换，ES 做存储和检索。每个环节职责单一，出了问题也好排查。

来看一个 Python Kafka 消费者的简单实现：

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'app-logs',
    bootstrap_servers=['kafka1:9092', 'kafka2:9092'],
    group_id='log-patrol-group',
    auto_offset_reset='latest',
    enable_auto_commit=True
)

for message in consumer:
    log_line = message.value.decode('utf-8')
    print(f"[Kafka] offset={message.offset}, value={log_line}")
    # 这里可以调用你的巡检规则引擎
    # patrol_engine.evaluate(log_line)

经验之谈：Kafka 采集的吞吐量取决于分区数和消费者组的设计。我个人建议分区数 = 消费者数 × 2，这样在消费者扩缩容时能保持均衡。另外，auto_offset_reset 一定要根据场景选对——线上用 latest，离线回溯用 earliest。

4.4 三种采集方式怎么选？

没有银弹，只有最适合你的场景。我根据实际经验整理了一个对比清单：

评估维度	文件采集	网络采集	Kafka 采集
日志量	小～中（≤100 GB/天）	中（≤1 TB/天）	大（≥1 TB/天）
实时性要求	中等（秒级）	高（毫秒级）	高（毫秒级）
资源消耗	低	中等（依赖连接数）	中等（依赖分区数）
数据可靠性	中等（需处理轮转）	中～高（需应用层确认）	高（副本机制）
运维复杂度	低	中	高（需管理 Kafka 集群）
扩展性	差（单机瓶颈）	中（可水平扩展采集端）	强（天然分布式）

避坑指南：我曾经在一个项目里，为了省事把所有日志都用网络采集直接发给 Logstash。结果某次业务高峰，日志量暴增 10 倍，Logstash 直接 OOM，导致整个采集链路瘫痪。后来老老实实加了 Kafka 做缓冲，再也没出过类似问题。记住——没有缓冲层的采集架构，扛不住突发流量。

总结一下我的选型思路：

日常巡检、日志量不大 → 文件采集 + inotify，简单可靠
远程设备、需要快速上报 → 网络采集，推荐 TCP 长连接
企业级、海量日志、多个消费方 → Kafka 采集，稳字当头

采集层是整个巡检系统的地基。地基没打好，上面的规则引擎再高效也跑不稳。希望这一章能帮你理清思路，找到适合自己场景的采集方案。