log4j2日志發給kafka

首先搭建ELK，其中es和kibana沒有特別要求，能正常訪問即可
logstash需要配置接收kafka的消息

input {
  kafka {
    bootstrap_servers => "kafka_network:9092"
    auto_offset_reset => "latest"
    topics => ["kafka-test"]
    type => "kafka"
  }
}

filter {
    if [type] == "kafka" {
        json {
            source => "message"
        }
    }
    
}

output {
  elasticsearch {
    hosts => "elasticsearch_network:9200"
    index => "kafka-%{+YYYY.MM.dd}"
  }
}

log4j2配置

<Properties>
  <property name="kafkaServers" value="127.0.0.1:9092" />
</Properties>

<appenders>
        <!--異步-->
        <Kafka name="Kafka" topic="kafka-test" syncSend="false">
            <!--JsonTemplateLayout provides more capabilitites and should be used instead.-->
            <!--這個文件是官網ECS示例，然后時間改成了時間戳，刪除了ECS版本號，可以根據實際情況配置模板-->
            <JsonTemplateLayout eventTemplateUri="classpath:EcsLayout.json"/>

            <Property name="bootstrap.servers">${kafkaServers}</Property>
        </Kafka>
</appenders>

<loggers>
  <Logger name="org.apache.kafka" level="info" />
  <root level="info">
            <appender-ref ref="Kafka"/>
  </root>
</loggers>

依賴

<log4j.version>2.17.0</log4j.version>
<kafka-client.version>2.7.2</kafka-client.version>

<!-- kafka appender 需要額外的依賴 https://logging.apache.org/log4j/2.x/runtime-dependencies.html-->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-layout-template-json</artifactId>
        </dependency>

在resources目錄下創建EcsLayout.json文件

{
  "timestamp": {
    "$resolver": "timestamp",
    "epoch": {
      "unit": "millis",
      "rounded": true
    }
  },
  "log.level": {
    "$resolver": "level",
    "field": "name"
  },
  "log.logger": {
    "$resolver": "logger",
    "field": "name"
  },
  "message": {
    "$resolver": "message",
    "stringified": true,
    "field": "message"
  },
  "process.thread.name": {
    "$resolver": "thread",
    "field": "name"
  },
  "labels": {
    "$resolver": "mdc",
    "flatten": true,
    "stringified": true
  },
  "tags": {
    "$resolver": "ndc"
  },
  "error.type": {
    "$resolver": "exception",
    "field": "className"
  },
  "error.message": {
    "$resolver": "exception",
    "field": "message"
  },
  "error.stack_trace": {
    "$resolver": "exception",
    "field": "stackTrace",
    "stackTrace": {
      "stringified": true
    }
  }
}

寫一個測試方法

    @GetMapping(value = "/lyc_exception")
    public void lyc_exception() {
        try{
            MDC.put("logMDC", "1234567890qwertyuiop");
            log.info("sdasdad");
            int i = 1/0;
        }catch (Exception e){
            log.error(e.getMessage());
        }finally {
            MDC.remove("logMDC");
        }
    }

就可以運行了，結果

 

 

 

 這只是一個簡化版，一般還需要補充環境配置和失敗回落文件

<SpringProfile>需要用到此依賴（v2.17.0）
<dependency>
    <groupId>org.apache.logging.log4j</groupId>
    <artifactId>log4j-spring-boot</artifactId>
    <version>${log4j.version}</version>
</dependency>
若使用asyncLogger需要用到此依賴
<dependency>
    <groupId>com.lmax</groupId>
    <artifactId>disruptor</artifactId>
    <version>3.4.2</version>
</dependency>
------------------------------------------
    <appenders>
        <!--在需要受環境影響的地方使用SpringProfile標簽包裹即可-->
        <SpringProfile name="prod">
            <!--不使用syncSend=false-->
            <Kafka name="Kafka" topic="card-mini-program-manager" ignoreExceptions="false">
                <!--JsonTemplateLayout provides more capabilitites and should be used instead.-->
                <JsonTemplateLayout eventTemplateUri="classpath:EcsLayout.json"/>
                <Property name="bootstrap.servers">${kafkaServers}</Property>
                <Property name="max.block.ms">2000</Property>
                <Property name="timeout.ms">3000</Property>
            </Kafka>
            <RollingFile name="FailoverKafkaLog" fileName="${FILE_PATH}/${KAFKA_FAILOVER_FILE_NAME}.log"
                         filePattern="${FILE_PATH}/${KAFKA_FAILOVER_FILE_NAME}-%d{yyyy-MM-dd}">
                <ThresholdFilter level="INFO" onMatch="ACCEPT" onMismatch="DENY"/>
                <JsonTemplateLayout eventTemplateUri="classpath:EcsLayout.json"/>
                <Policies>
                    <TimeBasedTriggeringPolicy interval="1" />
                </Policies>
                <DefaultRolloverStrategy max="15"/>
            </RollingFile>
            <!--這60秒期間都是使用第二選擇，60秒后重試第一選擇-->
            <Failover name="FailoverKafka" primary="Kafka" retryIntervalSeconds="60">
                <Failovers>
                    <AppenderRef ref="FailoverKafkaLog"/>
                </Failovers>
            </Failover>

            <!--異步使用AsyncLogger或者AsyncAppender，是由於sendSync會被捕獲異常，導致failover認為沒有異常，就無法使用failover了-->
            <Async name="AsyncKafka">
                <AppenderRef ref="FailoverKafka"/>
            </Async>
        </SpringProfile>
    </appenders>

    <loggers>
        <root level="info">
            <SpringProfile name="prod">
                <appender-ref ref="AsyncKafka"/>
            </SpringProfile>
        </root>
    </loggers>

如果kafka是有配置賬號密碼的話，這里我沒測試過，但根據網上的一些資料，猜測是

 

猜測增加這幾個property
<Property name="security.protocol">SASL_PLAINTEXT</Property>
<Property name="sasl.mechanism">SCRAM-SHA-256</Property>
<Property name="sasl.jaas.config">org.apache.kafka.common.security.scram.ScramLoginModule required username="xxx" password="yyy"</Property>