微服务日志之Spring Boot Kafka实现日志收集

目录#

准备工作
创建Spring Boot项目
配置Kafka
日志生产端实现
日志消费端实现
常见实践与最佳实践
总结
参考

1. 准备工作#

1.1 安装Kafka#

下载Kafka安装包，解压到本地目录。
启动Zookeeper（Kafka依赖Zookeeper进行协调管理）：bin/zookeeper-server-start.sh config/zookeeper.properties
启动Kafka服务：bin/kafka-server-start.sh config/server.properties

1.2 引入依赖#

在Spring Boot项目的pom.xml中添加以下依赖：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-logging</artifactId>
</dependency>

2. 创建Spring Boot项目#

使用Spring Initializr（https://start.spring.io/）创建一个Spring Boot项目，选择Web、Kafka等相关依赖。

3. 配置Kafka#

3.1 配置文件（application.properties）#

spring.kafka.bootstrap-servers=localhost:9092
# 生产者配置
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
# 消费者配置
spring.kafka.consumer.group-id=log-collector-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringSerializer

4. 日志生产端实现#

4.1 创建日志生产者类#

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
 
@Component
public class LogProducer {
 
    private static final String TOPIC = "log-topic";
 
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
 
    public void sendLog(String logMessage) {
        kafkaTemplate.send(TOPIC, logMessage);
    }
}

4.2 在业务代码中使用#

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
 
@RestController
public class HelloController {
 
    private static final Logger logger = LoggerFactory.getLogger(HelloController.class);
 
    @Autowired
    private LogProducer logProducer;
 
    @GetMapping("/hello")
    public String hello() {
        String logMessage = "This is a test log message";
        logger.info(logMessage);
        logProducer.sendLog(logMessage);
        return "Hello, World!";
    }
}

5. 日志消费端实现#

5.1 创建日志消费者类#

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
 
@Component
public class LogConsumer {
 
    @KafkaListener(topics = "log-topic", groupId = "log-collector-group")
    public void consume(String logMessage) {
        System.out.println("Received log: " + logMessage);
        // 这里可以进一步处理日志，比如存储到数据库、文件等
    }
}

6. 常见实践与最佳实践#

6.1 分区策略#

根据业务需求合理设置Kafka主题的分区数。如果日志量较大，可以增加分区数来提高并行处理能力。
可以根据日志的某些特征（如服务名称等）进行自定义分区，方便后续处理。

6.2 消息可靠性#

生产者可以设置acks参数（如acks = all）来确保消息发送成功。
消费者可以手动提交偏移量（enable.auto.commit = false，然后在合适的时候调用commitSync或commitAsync），避免消息丢失或重复消费。

6.3 日志格式规范#

统一日志格式，比如包含时间戳、日志级别、服务名称、具体日志内容等信息，方便后续分析和处理。

7. 总结#

通过本文的介绍，我们了解了如何使用Spring Boot与Kafka实现日志收集。从准备工作、项目创建、配置到生产端和消费端的实现，以及常见的实践和最佳实践。在实际项目中，可以根据具体需求进一步优化和扩展日志收集功能。

目录#