springcloud 微服务日志写入kafka

本文涉及的产品
注册配置 MSE Nacos/ZooKeeper,118元/月
云原生网关 MSE Higress,422元/月
任务调度 XXL-JOB 版免费试用,400 元额度,开发版规格
简介: springcloud 微服务日志写入kafka

SpringCloud微服务日志可以写入file 再通过filebeat写入logstash, 或者直接写入logstash。
日志写入kafka,可以利用kafak的高吞吐量高性能来降低系统延迟,之后再异步写入logstash.

启动kafka服务

kafka介绍与安装

启动zookeeper

./bin/zookeeper-server-start.sh ./config/zookeeper.properties

启动kafka

./bin/kafka-server-start.sh ./config/server.properties

创建topic

bin/kafka-topics.sh --create --topic logger-channel --bootstrap-server localhost:9092

消费端

./kafka-console-consumer.sh --topic logger-channel --bootstrap-server localhost:9092

springBoot服务

引入依赖


org.apache.kafka
kafka-clients
2.8.0

编写logback-spring.xml

使用了LayoutKafkaMessageEncoder进行消息编码
AsyncAppender 异步日志
topic 指定主题logger-channel

<?xml version="1.0" encoding="UTF-8"?>
<!-- scan:当此属性设置为true时,配置文件如果发生改变,将会被重新加载,默认值为true。 scanPeriod:设置监测配置文件是否有修改的时间间隔,如果没有给出时间单位,
 默认单位是毫秒当scan为true时,此属性生效。默认的时间间隔为1分钟。 debug:当此属性设置为true时,将打印出logback内部日志信息,实时查看logback运行状态。
 默认值为false。 -->
<!-- <configuration scan="false" scanPeriod="60 seconds" debug="false"> -->
<configuration>

  <!--设置上下文名称,用于区分不同应用程序的记录。一旦设置不能修改, 可以通过%contextName来打印日志上下文名称 -->
  <contextName>kafka-log-test</contextName>

  <property name="applicationName" value="${spring.application.name:-paw-kelk}"/>
  <!-- 定义日志的根目录 -->
  <property name="logDir" value="/Users/rubble/logs/${applicationName}" />
  <!-- 定义日志文件名称 -->
  <property name="logName" value="${applicationName}"/>
  <!--  profile -->
  <property name="profileActive" value="${spring.profile.active:-dev}" />

  <!-- ConsoleAppender 表示控制台输出 -->
  <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
    <!-- 日志输出格式: %d表示日期时间, %thread表示线程名, %-5level:级别从左显示5个字符宽度, %logger{50}
        表示logger名字最长50个字符,否则按照句点分割。 %msg:日志消息, %n是换行符 -->
    <encoder>
      <pattern>%d{HH:mm:ss.SSS}  %contextName [%thread] %-5level %logger{36}  - %msg%n</pattern>
    </encoder>
  </appender>

  <!-- 异常错误日志记录到文件  -->
  <appender name="logfile" class="ch.qos.logback.core.rolling.RollingFileAppender">
    <!-- <Encoding>UTF-8</Encoding> -->
    <File>${logDir}/${logName}.log</File>
    <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
      <FileNamePattern>${logDir}/history/${myspringboottest_log}.%d{yyyy-MM-dd}.rar</FileNamePattern>
      <maxHistory>30</maxHistory>
    </rollingPolicy>
    <encoder>
      <pattern>%d{HH:mm:ss.SSS} %contextName [%thread] %-5level %logger{36}  - %msg%n</pattern>
    </encoder>
  </appender>


  <appender name="kafkaAppender" class="com.github.danielwegener.logback.kafka.KafkaAppender">
    <encoder class="com.github.danielwegener.logback.kafka.encoding.LayoutKafkaMessageEncoder">
      <layout class="net.logstash.logback.layout.LogstashLayout" >
        <includeContext>false</includeContext>
        <includeCallerData>true</includeCallerData>
        <customFields>{"appName":"${applicationName}","env":"${profileActive}"}</customFields>
        <fieldNames class="net.logstash.logback.fieldnames.ShortenedFieldNames"/>
<!--        <fieldNames class="net.logstash.logback.fieldnames.LogstashFieldNames"/>-->
      </layout>
      <charset>UTF-8</charset>
    </encoder>
    <!--kafka topic 需要与配置文件里面的topic一致 否则kafka会沉默并鄙视你-->
    <topic>logger-channel</topic>
    <keyingStrategy class="com.github.danielwegener.logback.kafka.keying.HostNameKeyingStrategy" />
    <deliveryStrategy class="com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy" />
    <producerConfig>bootstrap.servers=localhost:9092</producerConfig>
    <!-- 如果kafka不可用则输出到 appender -->
    <appender-ref ref="logfile" />
  </appender>


  <!--异步写入kafka,尽量不占用主程序的资源-->
  <appender name="ASYNC" class="ch.qos.logback.classic.AsyncAppender">
    <neverBlock>true</neverBlock>
    <includeCallerData>true</includeCallerData>
    <discardingThreshold>0</discardingThreshold>
    <queueSize>2048</queueSize>
    <appender-ref ref="kafkaAppender" />
  </appender>

  <root level="INFO">
    <appender-ref ref="CONSOLE" />
<!--    <appender-ref ref="logfile"/>-->
        <!-- kafka 日志    -->
    <appender-ref ref="ASYNC" />
  </root>

</configuration>

测试类controller

@GetMapping("/kafka")
public String kafka(){
  int time = RandomUtil.randomInt(0, 100);
  log.info("cost time: "+time);
  log.debug("debug time: "+time);
  return "cost time: "+time;
}

消费端输出两条消息 info、debug

{"@timestamp":"2021-08-01T14:47:52.068+08:00","@version":1,"message":"cost time: 32","logger":"com.paw.kafka.elk.controller.KafkaLogController","thread":"http-nio-8080-exec-2","level":"INFO","levelVal":20000,"caller":{"class":"com.paw.kafka.elk.controller.KafkaLogController","method":"kafka","file":"KafkaLogController.java","line":35},"appName":"paw-kelk","env":"dev"}
{"@timestamp":"2021-08-01T14:47:52.068+08:00","@version":1,"message":"debug time: 32","logger":"com.paw.kafka.elk.controller.KafkaLogController","thread":"http-nio-8080-exec-2","level":"DEBUG","levelVal":10000,"caller":{"class":"com.paw.kafka.elk.controller.KafkaLogController","method":"kafka","file":"KafkaLogController.java","line":36},"appName":"paw-kelk","env":"dev"}

springcloud服务的日志已写入到kafka中。

相关文章
|
4月前
|
存储 消息中间件 大数据
大数据-69 Kafka 高级特性 物理存储 实机查看分析 日志存储一篇详解
大数据-69 Kafka 高级特性 物理存储 实机查看分析 日志存储一篇详解
71 4
|
4月前
|
存储 消息中间件 大数据
大数据-70 Kafka 高级特性 物理存储 日志存储 日志清理: 日志删除与日志压缩
大数据-70 Kafka 高级特性 物理存储 日志存储 日志清理: 日志删除与日志压缩
71 1
|
4月前
|
存储 消息中间件 大数据
大数据-68 Kafka 高级特性 物理存储 日志存储概述
大数据-68 Kafka 高级特性 物理存储 日志存储概述
48 1
|
5月前
|
消息中间件 Kafka API
python之kafka日志
python之kafka日志
52 3
|
5月前
|
消息中间件 存储 监控
Kafka的logs目录下的文件都是什么日志?
Kafka的logs目录下的文件都是什么日志?
299 11
|
6月前
|
存储 Java Spring
【Azure Spring Cloud】Azure Spring Cloud服务,如何获取应用程序日志文件呢?
【Azure Spring Cloud】Azure Spring Cloud服务,如何获取应用程序日志文件呢?
|
6月前
|
消息中间件 Java Kafka
【Azure 事件中心】开启 Apache Flink 制造者 Producer 示例代码中的日志输出 (连接 Azure Event Hub Kafka 终结点)
【Azure 事件中心】开启 Apache Flink 制造者 Producer 示例代码中的日志输出 (连接 Azure Event Hub Kafka 终结点)
|
17天前
|
消息中间件 存储 缓存
kafka 的数据是放在磁盘上还是内存上,为什么速度会快?
Kafka的数据存储机制通过将数据同时写入磁盘和内存,确保高吞吐量与持久性。其日志文件按主题和分区组织,使用预写日志(WAL)保证数据持久性,并借助操作系统的页缓存加速读取。Kafka采用顺序I/O、零拷贝技术和批量处理优化性能,支持分区分段以实现并行处理。示例代码展示了如何使用KafkaProducer发送消息。
|
4月前
|
消息中间件 存储 运维
为什么说Kafka还不是完美的实时数据通道
【10月更文挑战第19天】Kafka 虽然作为数据通道被广泛应用,但在实时性、数据一致性、性能及管理方面存在局限。数据延迟受消息堆积和分区再平衡影响;数据一致性难以达到恰好一次;性能瓶颈在于网络和磁盘I/O;管理复杂性涉及集群配置与版本升级。
180 1
|
4月前
|
消息中间件 Java Kafka
Flink-04 Flink Java 3分钟上手 FlinkKafkaConsumer消费Kafka数据 进行计算SingleOutputStreamOperatorDataStreamSource
Flink-04 Flink Java 3分钟上手 FlinkKafkaConsumer消费Kafka数据 进行计算SingleOutputStreamOperatorDataStreamSource
83 1