使用Filebeat替代Logstash HTTP输入优化ELK日志收集性能

背景

在之前的文章Docker-Compose部署ELK中，我们介绍了通过Logstash的HTTP端口直接收集SpringBoot应用日志的方案。虽然这种方案实现简单，但在高并发场景下存在一些性能问题：

• 业务应用需要同步发送日志到Logstash，可能阻塞业务线程
• 网络波动可能影响应用性能
• Logstash的HTTP输入处理大量连接时资源消耗较大

Filebeat的优势

Filebeat作为Elastic Stack的轻量级日志采集器，具有以下优势：

1. 低资源消耗：Go语言编写，内存占用小
2. 断点续传：支持日志位置记录，重启后从断点继续
3. 背压敏感：能够检测到Logstash或Elasticsearch过载并降低发送速率
4. 异步采集：与业务应用完全解耦，不影响应用性能

Filebeat配置详解

全部配置

# ============================== Filebeat inputs ===============================
filebeat.inputs:
- type: filestream
  enabled: true
  paths:
    - /opt/sc_fare/logs/*.json
    - /opt/sc_fare/logs/callog/*.json
  parsers:
    - ndjson: 
        target: ""  # 解析到根级别
        overwrite_keys: true
        add_error_key: true
  json:
    keys_under_root: true      # 将JSON字段提升到根级别
    overwrite_keys: false      # 不覆盖现有字段
    add_error_key: true        # JSON解析错误时添加错误字段
    message_key: "message"     # 指定消息字段
  # 文件处理设置
  close_inactive: 5m
  close_renamed: false
  close_removed: true
  close_eof: false
  clean_inactive: 72h
  
# ==================== Processors ====================
processors:
  - add_host_metadata: ~
  # 时间戳处理 - 使用日志中的@timestamp
  - timestamp:
      field: "@timestamp"
      layouts:
        - '2025-11-20T14:29:47.865+08:00'
      timezone: Asia/Shanghai
        
  - drop_fields:
      fields:
        # 删除所有agent相关字段
        - "agent.ephemeral_id"
        - "agent.id"
        - "agent.name"
        - "agent.type"
        - "agent.version"
        # 删除所有host相关字段（因为您图片中没有host.ip）
        - "host.architecture"
        - "host.containerized"
        - "host.hostname"
        - "host.os.type"
        - "host.os.kernel"
        - "host.os.platform"
        - "host.os.family"
        - "host.os.version"
        - "host.os.codename"
        - "host.os.name"
        - "host.mac"
        - "host.id"
        - "host.name"  # 删除host.name，因为您图片中显示有这个字段
        # 删除文件相关字段
        - "log.path"
        - "log.file.device_id"
        - "log.file.inode"
        - "log.offset"
        - "input.type"
        - "ecs.version"
        # 删除logstashHost（关键：这是您要删除的字段）
        - "logstashHost"
        - "logstashPort"
      ignore_missing: true
  
# ============================== Outputs =====================================
# 输出到 Logstash
output.logstash:
  hosts: ["192.168.53.130:5044"]
  bulk_max_size: 2048

输入配置

# ============================== Filebeat inputs ===============================
filebeat.inputs:
- type: filestream
  enabled: true
  paths:
    - /opt/sc_fare/logs/*.json
    - /opt/sc_fare/logs/callog/*.json

关键参数说明：

• filestream类型：适合持续增长的日志文件
• paths：支持通配符，可配置多个路径模式
• 文件监控设置确保高效的文件句柄管理

JSON解析配置

parsers:
  - ndjson: 
      target: ""
      overwrite_keys: true
      add_error_key: true
json:
  keys_under_root: true
  overwrite_keys: false
  add_error_key: true
  message_key: "message"

解析策略：

• ndjson解析器：处理每行一个JSON对象的日志格式
• keys_under_root: true：将JSON字段提升到根级别，便于检索
• 保留原始@timestamp字段，确保时间准确性

处理器配置

processors:
  - add_host_metadata: ~
  - timestamp:
      field: "@timestamp"
      layouts:
        - '2025-11-20T14:29:47.865+08:00'
      timezone: Asia/Shanghai
  - drop_fields:
      fields:
        # 精简字段，减少存储空间
        - "agent.ephemeral_id"
        - "agent.id"
        # ... 其他字段
      ignore_missing: true

处理器作用：

• 时间戳处理：标准化时间格式
• 字段精简：删除不必要的元数据，优化存储
• 主机信息：保留必要的主机标识信息

输出配置

output.logstash:
  hosts: ["192.168.53.130:5044"]
  bulk_max_size: 2048

优化参数：

• bulk_max_size: 2048：调整批量发送大小，平衡吞吐量和延迟
• 支持多主机配置，实现负载均衡和高可用

系统服务配置

创建Systemd服务文件，确保Filebeat稳定运行：

[Unit]
Description=Filebeat sends log files to Logstash or directly to Elasticsearch.
Documentation=https://www.elastic.co/beats/filebeat
Wants=network-online.target
After=network-online.target

[Service]
ExecStart=/soft/filebeat-8.15.0/filebeat -c /soft/filebeat-8.15.0/filebeat.yml -path.home /soft/filebeat-8.15.0 -path.config /soft/filebeat-8.15.0 -path.data /soft/filebeat-8.15.0/data -path.logs /soft/filebeat-8.15.0/logs
WorkingDirectory=/soft/filebeat-8.15.0
Restart=always
RestartSec=10
User=root
Group=root

服务管理命令：

# 启动服务
systemctl start filebeat

# 设置开机自启
systemctl enable filebeat

# 查看服务状态
systemctl status filebeat

# 查看日志
journalctl -u filebeat -f

性能优化建议

1. 资源限制

# 在filebeat.yml中添加
queue:
  mem:
    events: 4096
    flush.min_events: 512
    flush.timeout: 1s

2. 多工作器配置

# 对于高吞吐量场景
output.logstash:
  hosts: ["logstash1:5044", "logstash2:5044"]
  worker: 4
  loadbalance: true

3. 日志轮转策略

filebeat.inputs:
- type: filestream
  # ... 其他配置
  close_inactive: 5m
  clean_inactive: 72h
  close_renamed: true
  close_removed: true

监控与调试

启用监控

# 启用内部监控
monitoring:
  enabled: true
  period: 10s

日志级别调整

# 调试时调整日志级别
logging.level: debug
logging.to_files: true
logging.files:
  path: /soft/filebeat-8.15.0/logs
  name: filebeat.log
  keepfiles: 7

与Logstash集成

在Logstash端需要配置对应的Beats输入：

input {
  beats {
    port => 5044
    ssl => false
  }
}

filter {
  # 可在此处添加额外的过滤规则
}

output {
  elasticsearch {
    hosts => ["elasticsearch:9200"]
    index => "logs-%{+YYYY.MM.dd}"
  }
}

总结

通过使用Filebeat替代直接的HTTP日志收集，我们实现了：

1. 性能提升：业务应用与日志收集解耦
2. 可靠性增强：支持断点续传和背压控制
3. 资源优化：轻量级采集，减少系统负担
4. 维护简便：标准化的服务管理和监控

这种方案特别适合生产环境的高并发场景，能够有效提升整个日志收集系统的稳定性和性能。

---

注：实际部署时请根据具体环境调整路径、IP地址和端口号等参数。

SpringBoot Logback配置详解

完整的Logback配置文件示例

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <!-- 应用名称 -->
    <property name="appName" value="sc-fare"/>
    <!-- 日志文件路径 -->
    <property name="log.path" value="/opt/sc_fare/logs"/>
    
    <!-- JSON格式日志输出配置 -->
    <appender name="file_json" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>${log.path}/${appName}.json</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
            <fileNamePattern>${log.path}/${appName}-%d{yyyy-MM-dd}.%i.json</fileNamePattern>
            <maxFileSize>100MB</maxFileSize>
            <maxHistory>7</maxHistory>
            <totalSizeCap>1GB</totalSizeCap>
        </rollingPolicy>
        <encoder class="net.logstash.logback.encoder.LogstashEncoder">
            <providers>
                <mdc>
                    <includeMdcKeyName>traceId</includeMdcKeyName>
                </mdc>
                <timestamp/>
                <version/>
                <message/>
                <loggerName/>
                <threadName/>
                <logLevel/>
                <callerData/>
            </providers>
            <customFields>{"appName":"${appName}"}</customFields>
        </encoder>
    </appender>

    <!-- 异步Appender包装 -->
    <appender name="file_json_async" class="ch.qos.logback.classic.AsyncAppender">
        <!-- 队列大小：根据业务量调整 -->
        <queueSize>20000</queueSize>
        <!-- 重要：队列快满时不阻塞应用线程 -->
        <neverBlock>true</neverBlock>
        <!-- 当队列剩余20%容量时，丢弃TRACE/DEBUG级别日志 -->
        <discardingThreshold>0</discardingThreshold>
        <!-- 最大等待刷新时间（毫秒） -->
        <maxFlushTime>30000</maxFlushTime>
        <!-- 添加队列满时的警告 -->
        <includeCallerData>false</includeCallerData>
        <!-- 包含的appender -->
        <appender-ref ref="file_json"/>
    </appender>

    <!-- 传统文本格式日志（可选） -->
    <appender name="file_info" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>${log.path}/${appName}.log</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
            <fileNamePattern>${log.path}/${appName}-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
            <maxFileSize>100MB</maxFileSize>
            <maxHistory>7</maxHistory>
            <totalSizeCap>1GB</totalSizeCap>
        </rollingPolicy>
        <encoder>
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n</pattern>
        </encoder>
    </appender>

    <!-- 错误日志单独输出 -->
    <appender name="file_error" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>${log.path}/${appName}-error.log</file>
        <filter class="ch.qos.logback.classic.filter.LevelFilter">
            <level>ERROR</level>
            <onMatch>ACCEPT</onMatch>
            <onMismatch>DENY</onMismatch>
        </filter>
        <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
            <fileNamePattern>${log.path}/${appName}-error-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
            <maxFileSize>100MB</maxFileSize>
            <maxHistory>30</maxHistory>
            <totalSizeCap>2GB</totalSizeCap>
        </rollingPolicy>
        <encoder>
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n</pattern>
        </encoder>
    </appender>

    <!-- 日志级别配置 -->
    <root level="info">
        <appender-ref ref="file_info"/>
        <appender-ref ref="file_error"/>
        <appender-ref ref="file_json_async"/>
    </root>
</configuration>

配置详解

1. JSON日志输出配置

LogstashEncoder关键特性：

• 自动生成结构化JSON日志
• 包含完整的日志上下文信息
• 支持MDC（Mapped Diagnostic Context）字段

自定义字段说明：

<customFields>{"appName":"${appName}"}</customFields>

在每条日志中自动添加应用标识，便于在ELK中区分不同应用日志。

2. 异步日志配置优化

异步Appender参数调优：

参数	推荐值	说明
queueSize	20000	根据业务峰值调整，避免内存溢出
neverBlock	true	队列满时不阻塞业务线程
discardingThreshold	0	队列剩余20%时丢弃低级别日志
maxFlushTime	30000	应用关闭时最大等待时间

3. 依赖配置

在pom.xml中添加必要依赖：

<dependencies>
    <!-- Logstash Logback Encoder -->
    <dependency>
        <groupId>net.logstash.logback</groupId>
        <artifactId>logstash-logback-encoder</artifactId>
        <version>7.4</version>
    </dependency>
    
    <!-- 异步Appender支持 -->
    <dependency>
        <groupId>ch.qos.logback</groupId>
        <artifactId>logback-classic</artifactId>
        <version>1.4.14</version>
    </dependency>
</dependencies>

日志格式示例

生成的JSON日志示例：

{
  "@timestamp": "2025-11-20T14:29:47.865+08:00",
  "@version": "1",
  "message": "用户登录成功",
  "logger_name": "com.example.UserController",
  "thread_name": "http-nio-8080-exec-1",
  "level": "INFO",
  "level_value": 20000,
  "appName": "sc-fare",
  "traceId": "a1b2c3d4e5f6",
  "caller_class_name": "com.example.UserController",
  "caller_method_name": "login",
  "caller_file_name": "UserController.java",
  "caller_line_number": 128
}

性能优化建议

1. 根据业务场景调整参数

高并发场景：

<appender name="file_json_async" class="ch.qos.logback.classic.AsyncAppender">
    <queueSize>50000</queueSize>
    <discardingThreshold>0</discardingThreshold>
    <neverBlock>true</neverBlock>
    <includeCallerData>false</includeCallerData>
    <appender-ref ref="file_json"/>
</appender>

低资源环境：

<appender name="file_json_async" class="ch.qos.logback.classic.AsyncAppender">
    <queueSize>5000</queueSize>
    <discardingThreshold>20</discardingThreshold>
    <neverBlock>false</neverBlock>
    <includeCallerData>false</includeCallerData>
    <appender-ref ref="file_json"/>
</appender>

2. MDC跟踪配置

在代码中使用MDC添加跟踪信息：

import org.slf4j.MDC;

// 在请求处理开始时设置traceId
MDC.put("traceId", UUID.randomUUID().toString().substring(0, 8));

// 请求处理完成后清理
MDC.clear();

3. 日志文件管理策略

滚动策略说明：

• maxFileSize: 100MB：单个文件最大100MB
• maxHistory: 7：保留最近7天的日志文件
• totalSizeCap: 1GB：日志目录总大小限制

与Filebeat集成注意事项

1. 文件路径匹配

确保Filebeat配置的路径与Logback输出路径一致：

paths:
  - /opt/sc_fare/logs/*.json
  - /opt/sc_fare/logs/callog/*.json

2. 时间戳格式

Logback的@timestamp格式与Filebeat时间戳处理器匹配：

- timestamp:
    field: "@timestamp"
    layouts:
      - '2025-11-20T14:29:47.865+08:00'
    timezone: Asia/Shanghai

通过以上配置，SpringBoot应用能够高效生成结构化JSON日志，与Filebeat无缝集成，实现高性能的日志收集方案。