使用log4j2 Kafka Appender输出日志到Kafka时如何限制发送速率？

以下方案均针对Kafka Appender单独生效，不会影响本地文件、控制台等其他Appender的日志输出。

方案1：使用log4j2自带BurstFilter（无需自定义代码，推荐） #

如果你使用的是log4j2（Flink 1.10及之后版本默认使用log4j2），可以直接用官方自带的BurstFilter实现速率控制，直接修改log4j2配置文件即可：

<Kafka name="KafkaAppender" topic="你的目标topic">
    <!-- 限流配置：每秒最多发送3条日志，超过的直接丢弃 -->
    <BurstFilter level="all" rate="3" maxBurst="3" onMatch="ACCEPT" onMismatch="DENY"/>
    <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
    <Property name="bootstrap.servers">你的kafka地址</Property>
</Kafka>

参数说明：

• rate：每秒允许通过的最大日志条数，按需求填3即可
• maxBurst：允许的最大突发日志数，如需严格限制每秒不超过3条，填3即可
• level="all"：表示对所有级别的日志都生效限流规则

方案2：自定义Filter实现更灵活的限流逻辑（适配log4j1、log4j2） #

如果需要更复杂的限流逻辑，或者你使用的是log4j1，可以基于Guava的RateLimiter自定义限流Filter，步骤如下：

1. 引入Guava依赖（如果Flink环境已经自带可以跳过）

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>31.1-jre</version>
</dependency>

2. 编写自定义Filter，以log4j2为例：

import com.google.common.util.concurrent.RateLimiter;
import org.apache.logging.log4j.core.Filter;
import org.apache.logging.log4j.core.LogEvent;
import org.apache.logging.log4j.core.config.plugins.Plugin;
import org.apache.logging.log4j.core.config.plugins.PluginAttribute;
import org.apache.logging.log4j.core.config.plugins.PluginFactory;

@Plugin(name = "RateLimitFilter", category = "Core", elementType = Filter.ELEMENT_TYPE)
public class RateLimitFilter extends AbstractFilter {
    private final RateLimiter rateLimiter;

    private RateLimitFilter(double permitsPerSecond, Result onMatch, Result onMismatch) {
        super(onMatch, onMismatch);
        this.rateLimiter = RateLimiter.create(permitsPerSecond);
    }

    @Override
    public Result filter(LogEvent event) {
        return rateLimiter.tryAcquire() ? onMatch : onMismatch;
    }

    @PluginFactory
    public static RateLimitFilter createFilter(
            @PluginAttribute("permitsPerSecond") double permitsPerSecond,
            @PluginAttribute("onMatch") Result onMatch,
            @PluginAttribute("onMismatch") Result onMismatch) {
        return new RateLimitFilter(permitsPerSecond, 
            onMatch == null ? Result.ACCEPT : onMatch,
            onMismatch == null ? Result.DENY : onMismatch);
    }
}

3. 配置到Kafka Appender中：

<Kafka name="KafkaAppender" topic="你的目标topic">
    <RateLimitFilter permitsPerSecond="3" onMatch="ACCEPT" onMismatch="DENY"/>
    <!-- 其余配置不变 -->
</Kafka>

注意事项 #

• 上述限流都是单进程（单Flink TaskManager）级别的限流，如果你的任务有N个TaskManager实例，总发送速率为3*N条/秒
• 超过限流阈值的日志默认直接丢弃，如果你需要缓存后延迟发送，可以修改自定义Filter的逻辑，替换tryAcquire为阻塞式的acquire即可，不过不推荐在日志链路加阻塞逻辑，可能影响Flink任务运行效率

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