如何使用Serilog实现日志消息的防抖处理？

一、基于日志框架扩展（单应用场景首选） #

如果是单应用内的轮询日志，直接基于常用日志框架扩展是最便捷的方式，以Logback为例：

1. 自定义聚合Appender
   实现一个继承自AppenderBase<ILoggingEvent>的自定义Appender，内部维护线程安全的Map存储日志消息与计数，配合定时任务每隔5分钟输出汇总并清空计数：

public class AggregatingAppender extends AppenderBase<ILoggingEvent> {
    private final ConcurrentHashMap<String, AtomicInteger> logCounter = new ConcurrentHashMap<>();
    private ScheduledExecutorService scheduler;
    private long intervalMinutes = 5;

    @Override
    public void start() {
        super.start();
        scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.scheduleAtFixedRate(this::flushAggregatedLogs, intervalMinutes, intervalMinutes, TimeUnit.MINUTES);
    }

    @Override
    protected void append(ILoggingEvent event) {
        String message = event.getFormattedMessage();
        // 只聚合特定标记的轮询日志，比如含"轮询检查"关键词
        if (message.contains("轮询检查")) {
            logCounter.computeIfAbsent(message, k -> new AtomicInteger(0)).incrementAndGet();
        } else {
            // 非聚合日志直接输出
            addInfo(message);
        }
    }

    private void flushAggregatedLogs() {
        if (logCounter.isEmpty()) return;
        StringBuilder sb = new StringBuilder("过去").append(intervalMinutes).append("分钟内以下轮询日志已记录：\n");
        logCounter.forEach((msg, count) -> sb.append("- ").append(msg).append("：").append(count.get()).append("次\n"));
        addInfo(sb.toString());
        logCounter.clear();
    }

    @Override
    public void stop() {
        scheduler.shutdown();
        super.stop();
    }
}

在Logback配置文件中引入该Appender，替代原有输出目标即可。

2. 利用TurboFilter做前置过滤聚合
   通过TurboFilter拦截特定日志，先完成计数，到达时间窗口后统一输出，适合无需保留原始日志的场景。

二、独立日志聚合服务（分布式架构场景） #

如果是多实例分布式架构，用独立日志管道工具做聚合更合适：

Logstash配置示例 #

使用metrics插件实现窗口计数，配合ruby过滤器生成汇总日志：

input {
  beats {
    port => 5044
  }
}

filter {
  if [message] =~ /轮询检查/ {
    metrics {
      meter => "log.count.%{message}"
      add_tag => "aggregated"
      flush_interval => 300 # 5分钟窗口，单位秒
    }
  }
}

output {
  if "aggregated" in [tags] {
    ruby {
      code => '
        event.set("summary_message", "过去5分钟内轮询日志「#{event.get("[message]")}」已记录#{event.get("[metrics][count][log.count.#{event.get("[message]")}]")}次")
      '
    }
    elasticsearch {
      hosts => ["localhost:9200"]
      index => "aggregated-logs-%{+YYYY.MM.dd}"
    }
  } else {
    elasticsearch {
      hosts => ["localhost:9200"]
      index => "normal-logs-%{+YYYY.MM.dd}"
    }
  }
}

Fluentd配置思路 #

用filter_record_transformer配合time_slice_tag做窗口聚合，统计相同日志出现次数后输出汇总消息。

三、业务代码层面直接处理 #

如果不想改动日志框架，可在轮询逻辑里直接做计数：

public class PollingService {
    private final AtomicInteger pollLogCount = new AtomicInteger(0);
    private final String pollMessage = "轮询检查：系统状态正常";

    public void doPoll() {
        // 轮询业务逻辑...
        pollLogCount.incrementAndGet();
    }

    public PollingService() {
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.scheduleAtFixedRate(() -> {
            int count = pollLogCount.getAndSet(0);
            if (count > 0) {
                LoggerFactory.getLogger(PollingService.class).info("过去5分钟内「{}」已记录{}次", pollMessage, count);
            }
        }, 5, 5, TimeUnit.MINUTES);
    }
}

这种方式最直接，但耦合业务代码，适合单一固定轮询日志的场景。

内容的提问来源于stack exchange，提问作者Ynv