news 2026/7/10 12:37:46

Kafka 和springboot 整合Logback日志

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
Kafka 和springboot 整合Logback日志

Spring Boot 整合 Kafka 进行日志处理是一个常见的任务，可以帮助你更好地管理和分析应用程序的日志。以下是一个基本的步骤指南，帮助你完成这个整合。

本文介绍了如何在SpringBoot应用中整合Kafka，利用Logback收集日志并发送到Kafka。详细步骤包括添加Kafka相关依赖，配置logback.xml，自定义KafkaAppender以及提供测试。

首先安装kafka.。之前已经安装过了，我们这里不再讲。，可以参考前面讲的类容：

https://blog.csdn.net/lchmyhua88/article/details/155640953?spm=1001.2014.3001.5502

接着我们按照下面的步骤开始：

  • 1.pom依赖
  • 2.logback.xml配置
  • 3.配置 kafka消费者
  • 4.监听消费消息&测试代码

1. 添加依赖

<!-- Spring Boot Starter Kafka --> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> <!-- Logback Kafka Appender --> <dependency> <groupId>com.github.danielwegener</groupId> <artifactId>logback-kafka-appender</artifactId> <version>0.2.0-RC2</version> </dependency> <!-- Kafka Clients --> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>3.0.0</version> </dependency>

2.配置 Logback

<configuration> <!-- 控制台输出appender --> <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender"> <encoder> <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern> </encoder> </appender> <!-- Kafka appender --> <appender name="KAFKA" class="com.github.danielwegener.logback.kafka.KafkaAppender"> <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder"> <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern> </encoder> <!-- Kafka topic名称 --> <topic>log-topic</topic> <!-- Kafka bootstrap servers --> <keyingStrategy class="com.github.danielwegener.logback.kafka.keying.NoKeyKeyingStrategy" /> <deliveryStrategy class="com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy" /> <!-- Kafka producer配置 --> <producerConfig>bootstrap.servers=192.168.33.10:9092</producerConfig> <producerConfig>acks=0</producerConfig> <producerConfig>linger.ms=1000</producerConfig> <producerConfig>max.block.ms=0</producerConfig> <!-- 过滤器设置（可选） --> <filter class="ch.qos.logback.classic.filter.ThresholdFilter"> <level>INFO</level> </filter> </appender> <!-- 根logger配置 --> <root level="INFO"> <appender-ref ref="CONSOLE"/> <appender-ref ref="KAFKA"/> </root> </configuration>

3.配置 kafka消费者，为了方便测试我们把生产者消费者都配置上

spring.kafka.bootstrap-servers=192.168.33.10:9092 #spring.kafka.consumer.group-id=my-group spring.kafka.consumer.auto-offset-reset=earliest spring.kafka.consumer.enable-auto-commit=true spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer spring.kafka.consumer.properties.spring.json.trusted.packages=* spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonDeserializer

4.注入kafka bean ,方便后面调试生产者。

@Configuration public class KafkaProducerConfig { @Value("${spring.kafka.bootstrap-servers}") private String bootstrapServers; @Bean public ProducerFactory<String, Object> producerFactory() { Map<String, Object> configProps = new HashMap<>(); configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class); return new DefaultKafkaProducerFactory<>(configProps); } @Bean public KafkaTemplate<String, Object> kafkaTemplate() { return new KafkaTemplate<>(producerFactory()); } }

5.监听消费消息

@Component public class LogConsumer { @KafkaListener(topics = "log-topic",groupId = "my-group") public void listen(String message) { System.out.println("收到日志消息: " + message); } }

6.添加日志打印

@RestController @RequestMapping("/kafka") public class LogController { @Autowired private KafkaProducerServiceImpl producerService; private static final Logger logger = (Logger) LoggerFactory.getLogger(LogController.class); @GetMapping("/log") public String log() { //producerService.sendLogMessage("This is a test log message"); logger.info("这是一条测试日志消息"); logger.warn("这是一条警告日志消息"); logger.error("这是一条错误日志消息"); return "Log message sent"; } }

下面我们开始测试，通过调用接口：

控制台可以看到我写3条日志，kafka消费3条消息

这样我们就通过logback把日志收集到kafka里面。方便数据分析。当然你也可以写入es进行分析画像。

我们可以通过kafka manger控制台来监控消息。

下载kafka-manager，下载地址： https://github.com/yahoo/kafka-manager

解压后配置application.conf kafka地址：

保存后，指定8888端口启动：

bin/kafka-manager -Dconfig.file=conf/application.conf -Dhttp.port=8888

启动后打开控制面板，可以看到kafka集群节点信息和分区，消费偏移等信息：

http://192.168.33.10:8888/

详情可以去官网看看介绍：

https://github.com/yahoo/CMAK

版权声明: 本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：809451989@qq.com进行投诉反馈，一经查实，立即删除！
网站建设 2026/7/9 14:49:01

25、Ubuntu 网络应用全攻略

Ubuntu 网络应用全攻略 1. Firefox 浏览器使用技巧 Firefox 支持标签式窗口，提供了多种打开新标签的方式： - 点击“New Tab”按钮（现有标签右侧的“+”符号）。 - 按住“Ctrl”键并点击链接，可在新标签中打开。 - 按下“Ctrl - T”组合键。 - 若鼠标有中键，有时点击…

作者头像 李华
网站建设 2026/7/8 14:00:24

26、开启Ubuntu系统下OpenOffice.org的高效办公之旅

开启Ubuntu系统下OpenOffice.org的高效办公之旅 在Ubuntu系统中，我们可以借助网络完成许多操作。若想使用的程序或功能未被提及，可通过Synaptic软件包管理器或Ubuntu软件中心进行搜索，大概率能找到所需的程序或具有类似功能的替代程序。 而开源软件强大功能与能力的一个绝…

作者头像 李华
网站建设 2026/7/8 18:38:37

27、Ubuntu 软件与游戏使用指南

Ubuntu 软件与游戏使用指南 一、OpenOffice.org 软件介绍 OpenOffice.org 是一套功能强大的办公软件套件，可与其他商业套件相媲美。它是开源软件，有众多开发者持续对其更新，自 2006 年以来每年至少更新四次。 （一）任务框架（The Tasks Frame） 任务框架提供了各种任务…

作者头像 李华
网站建设 2026/7/9 20:58:48

AI营销增长：4大核心能力+前端落地指南

AI赋能营销增长&#xff1a;4大核心能力前端落地全指南&#xff08;小白友好版&#xff09; 之前我们聊了营销增长的业务场景&#xff0c;今天聚焦“AI到底怎么帮我们干活”——从写文案、分用户到投广告、做客服&#xff0c;AI能把“费人力、耗时间”的活儿变简单。这篇会把A…

作者头像 李华
网站建设 2026/7/10 2:35:36

基于 Rust 实现单向网闸环境下的 MQTT 消息透明传输

摘要&#xff1a;在工业互联网与关键信息基础设施保护&#xff08;CIIP&#xff09;场景中&#xff0c;物理单向网闸&#xff08;Data Diode/光闸&#xff09;的部署虽然提升了安全边界&#xff0c;但也阻断了 TCP/IP 协议的双向通信能力&#xff0c;导致 Kafka、MQTT 等通信协…

作者头像 李华
网站建设 2026/7/10 1:35:45

31、Shell编程：从基础到高级应用

Shell编程：从基础到高级应用 1. Here文档的使用 在输出文本时，除了使用 echo 命令的两种方法外，还有一种称为Here文档（Here Document）或Here脚本的方式。Here文档是一种额外的I/O重定向形式，它允许我们将一段文本嵌入到脚本中，并将其作为命令的标准输入。其基本语法…

作者头像 李华