springboot - 2.7.3版本 - （八）ELK整合Kafka

接之前的内容进行整改，加入kafka：springboot - 2.7.3版本 - （五）整合Elasticsearch+Logstash+Kibana（ELK）_小小小爬虫的博客-CSDN博客

原来顺序： logback - logstash - elasticSearch 

变更顺序：logback - kafka - logstash - elasticSearch

一，pom.xml添加依赖包

<dependency>
	<groupId>com.github.danielwegenergroupId>
	<artifactId>logback-kafka-appenderartifactId>
	<version>0.2.0-RC2version>
dependency>

二，修改logback-spring.xml，将logstash-appender改为kafka-appender

<appender name="KAFKA" class="com.github.danielwegener.logback.kafka.KafkaAppender">
	<encoder class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder">
		<providers class="net.logstash.logback.composite.loggingevent.LoggingEventJsonProviders">
			<pattern>
				<pattern>
					{"app":"${APP}",
					"profile":"${PROFILES_ACTIVE}",
					"thread": "%thread",
					"logger": "%logger{5}",
					"message":"%msg",
					"app_name":"${APP_NAME}",
					"env_name":"${ENV_NAME}",
					"hostname":"${HOSTNAME}",
					"captain_seq":"${CAPTAIN_SEQ}",
					"captain_gen":"${CAPTAIN_GEN}",
					"build_name":"${BUILD_NAME}",
					"build_git_version":"${BUILD_GIT_VERSION}",
					"build_git_hash":"${BUILD_GIT_HASH}",
					"build_timestamp":"${BUILD_TIMESTAMP}",
					"date":"%d{yyyy-MM-dd HH:mm:ss.SSS}",
					"level":"%level",
					"stack_trace":"%exception"
					}
				pattern>
			pattern>
		providers>
	encoder>
	<topic>appLogtopic>
	<keyingStrategy class="com.github.danielwegener.logback.kafka.keying.NoKeyKeyingStrategy" />
	<deliveryStrategy class="com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy" />
	<producerConfig>bootstrap.servers=localhost:9092producerConfig>
	<producerConfig>retries=1producerConfig>
　　　 　  <producerConfig>batch-size=16384producerConfig>
　　　 　  <producerConfig>buffer-memory=33554432producerConfig>
　　　 　  <producerConfig>properties.max.request.size==2097152producerConfig>
	<appender-ref ref="CONSOLE"/>
appender>
 
<root level="INFO">
	<appender-ref ref="KAFKA" />
	<appender-ref ref="FILE" />
	<appender-ref ref="CONSOLE" />
root>

三，logstash中/config添加文件logstash-kafka-es.conf，改为从kafka输入

# Sample Logstash configuration for creating a simple
# Logback -> kafka -> Logstash -> Elasticsearch pipeline.
 
input {
  kafka {
    id => "my_plugin_id"
    bootstrap_servers => "127.0.0.1:9092"
    topics => ["appLog"]
    auto_offset_reset => "latest"
  }
}
 
output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "logstash-test-applog-%{+YYYY.MM.dd}"
    #user => "elastic"
    #password => "changeme"
  }
}

四，启动测试

- 启动zookeeper

- 启动kafka 【.\bin\windows\kafka-server-start.bat .\config\server.properties】

- 启动elasticSearch

- 启动logstash 【logstash -f ./config/logstash-kafka-es.conf】

- 启动kibana

- 启动springboot项目

 

 

五，源代码下载： https://download.csdn.net/download/MyNoteBlog/86729194