本文专题主要是SpringBoot和ELK的实践，使用Kafka作为日志输送！
? ? ? ? 首先是ELK的安装，elasticsearch和logstash以及kibana的安装，这里我使用的是docker安装，下面我把我安装的版本贴出，具体安装过程以后再出，毕竟这不是我们开发需要考虑的东西，你说是不是。
首先elasticsearch和kibana版本保持一致，kafka版本和zookeeper自动按照docker获取即可。
????????
安装好之后，测试各项服务安装成功并且可用即可。
1.1 kinaba服务 执行 ip:5601，出现下图即成功
1.2 elasticsearch服务 执行 ip:9200,出现下图即成功
?1.3 下载 logstash， 解压到指定目录，该步骤网上有很多教程，不细说了, 解压后到 config 目录下，创建 logstash.conf 文件，文件为 springboot 集成时要用到的配置，文件内容为：
?然后去bin 目录下指定此配置文件 启动即可?logstash -f …/config/logstash.conf?
至此 ELK 的环境准备及测试完成，接下来进行springboot 集成 ELK 所需的配置
?配置
简单准备一个可以正常启动的 springboot 项目，且有 log 的输出。
本文只列出关于 ELK 的相关配置，其他依赖或者配置可以自由添加。
1.添加依赖：
?
<!--集成logstash-->
<dependency>
<groupId>net.logstash.logback</groupId>
<artifactId>logstash-logback-encoder</artifactId>
<version>5.3</version>
</dependency>
<!--引入es-->
<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch</artifactId>
<version>7.8.0</version>
</dependency>
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-high-level-client</artifactId>
<version>7.8.0</version>
</dependency>
<!--引入kafak和spring整合的jar-->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.8.2</version>
</dependency>
其他所需依赖可以自由添加
- application.yml 内容
spring.application.name=elk-user server.port=8080 #============== kafka =================== # 指定kafka 代理地址，可以多个 spring.kafka.bootstrap-servers=ip:9092 #=============== provider ======================= #重试次数 spring.kafka.producer.retries=0 # 每次批量发送消息的数量 spring.kafka.producer.batch-size=16384 spring.kafka.producer.buffer-memory=33554432 # 指定消息key和消息体的编解码方式 spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer #=============== consumer ======================= # 指定默认消费者group id spring.kafka.consumer.group-id=user-log-group spring.kafka.listener.missing-topics-fatal=false spring.kafka.consumer.auto-offset-reset=earliest #是否自动提交offer spring.kafka.consumer.enable-auto-commit=false spring.kafka.consumer.auto-commit-interval=100 # 指定消息key和消息体的编解码方式 spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
- resource 目录下添加文件 log4j.xml，文件内容为：
? ? ? ?
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="off" monitorInterval="10">
<properties>
<property name="KAFKA_SERVER">${env:KAFKA_SERVER:-124.223.87.70:9092}</property>
<property name="LOG_LEVEL">${env:LOG_LEVEL:-INFO}</property>
</properties>
<Appenders>
<Console name="Console" target="SYSTEM_OUT">
<PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %-5level %c{1.} - %msg%n"/>
<!--<JsonLayout stacktraceAsString ="true"/>-->
</Console>
<Kafka name="Kafka" topic="user-log">
<ThresholdFilter level="INFO" onMatch="ACCEPT" onMismatch="DENY" />
<Property name="bootstrap.servers">${KAFKA_SERVER}</Property>
<JsonLayout stacktraceAsString ="true">
<KeyValuePair key="hostname" value="${hostName}"/>
<KeyValuePair key="service_name" value="日志"/>
<KeyValuePair key="detail" value="$${ctx:detail:-}"/>
</JsonLayout>
</Kafka>
<Async name="Async">
<AppenderRef ref="Kafka"/>
<AppenderRef ref="Console"/>
</Async>
</Appenders>
<Loggers>
<Root level="${LOG_LEVEL}">
<AppenderRef ref="Async"/>
</Root>
<Logger name="org.apache.kafka" level="INFO" />
</Loggers>
</Configuration>
整体项目目录如下
先看Bean目录下的UserLog，这里仅仅记录一个实体，便于输出。?
然后是config目录 ：
ElasticSearchConfig和KafkaConfig
接着是Controller
?然后是Customer
????????
UserLogConsumer消费者
KafkaMessageListener这是kafka的一个监听类，他会自动去监听相同topic生产者所发的消息，然后插入es中。
Producer 生产者示例
我们通过postman去调用即可：localhost:8080/test/send，调用第一次之后我们需要去kinaba查看下：
我们发现es中已经写入了我们代码中所设置的索引，然后我们去?kibana中设置
这里因为我已经设着过user-log,所以我重新设置了另一个。设置完之后，去主页面即可看到我们的信息
?到这就完成了springboot+elk+kafka的整合了，实现了日志收集
